fix(sql): set operations take DuckDB's types and answer one class per column; aggregates without GROUP BY answer one row

THIRD-PARTY-NOTICES

@@ -0,0 +1,63 @@
+THIRD-PARTY NOTICES
+
+This project includes portions of the third-party software listed below, each distributed under
+its own license, reproduced here.
+
+================================================================================
+{fmt} 6.1.2
+https://github.com/fmtlib/fmt/tree/6.1.2
+
+Used in: core/src/main/scala/app/softnetwork/elastic/client/FloatingPointText.scala
+A Scala port of {fmt}'s floating-point formatting (include/fmt/format-inl.h: `fp`, `multiply`,
+`get_cached_power` and its tables, `bigint`, `grisu_gen_digits`, `grisu_shortest_handler`,
+`fallback_format`, the shortest branch of `format_float`; include/fmt/format.h: the `float_writer`
+layout of the `{}` specifier and `write_exponent`).
+
+License (https://raw.githubusercontent.com/fmtlib/fmt/6.1.2/LICENSE.rst):
+
+Copyright (c) 2012 - present, Victor Zverovich
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+--- Optional exception to the license ---
+
+As an exception, if, as a result of your compiling your source code, portions
+of this Software are embedded into a machine-executable object form of such
+source code, you may redistribute such embedded portions in such object form
+without including the above copyright and permission notices.
+
+================================================================================
+DuckDB 1.5.5
+https://github.com/duckdb/duckdb/tree/v1.5.5
+
+Used in: core/src/main/scala/app/softnetwork/elastic/client/TemporalText.scala
+The text layout of a DATE, a TIME and a TIMESTAMP cast to VARCHAR, translated from
+`DateToStringCast` and `TimeToStringCast` (src/include/duckdb/common/types/cast_helpers.hpp) and
+`StringFromTimestamp` / `StringAsTime` (src/common/operator/string_cast.cpp).
+
+License (https://raw.githubusercontent.com/duckdb/duckdb/v1.5.5/LICENSE):
+
+Copyright 2018-2025 Stichting DuckDB Foundation
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/BucketScriptParams.scala

@@ -18,8 +18,16 @@ package app.softnetwork.elastic.sql.bridge
 
 import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
 import com.fasterxml.jackson.databind.node.{ArrayNode, ObjectNode}
+import app.softnetwork.elastic.sql.query.SingleSearch
 import com.sksamuel.elastic4s.requests.searches.SearchRequest
-import com.sksamuel.elastic4s.requests.searches.aggs.{AbstractAggregation, Aggregation}
+import com.sksamuel.elastic4s.requests.searches.aggs.{
+  AbstractAggregation,
+  Aggregation,
+  CardinalityAggregation,
+  KeyedFiltersAggregation,
+  SumAggregation,
+  ValueCountAggregation
+}
 import com.sksamuel.elastic4s.requests.searches.aggs.pipeline.BucketScriptPipelineAgg
 
 import scala.jdk.CollectionConverters._
@@ -35,8 +43,11 @@ import scala.jdk.CollectionConverters._
   * The script is sent as the object Elasticsearch documents -- `source`, `lang`, `params` -- for
   * exactly the `bucket_script`s that carry params, so the source stays the SAME text from one
   * request to the next and Elasticsearch compiles it once: inlining the clock instead would compile
-  * a new script on every request, against the cluster's compilation rate limit. Every other body is
-  * returned untouched, unparsed.
+  * a new script on every request, against the cluster's compilation rate limit.
+  *
+  * The same builder never writes a `gap_policy` either (and elastic4s 7.17 to 9.0 knows only `skip`
+  * and `insert_zeros`), so the `keep_values` of a whole-table calculation is written here too
+  * ([[keepingValues]]). Every other body is returned untouched, unparsed.
   */
 object BucketScriptParams {
 
@@ -53,13 +64,63 @@ object BucketScriptParams {
       case _                   => Nil
     }.toMap
 
-  /** `body`, the JSON of `search`, with the params of its `bucket_script`s bound. */
+  /** The `bucket_script`s of the whole-table bucket (issue #413) whose every operand is a count or
+    * a sum -- `_count`, `value_count`, `cardinality`, `sum` -- by name.
+    *
+    * Over ZERO documents Elasticsearch takes every `buckets_path` but `_count` for a gap, and its
+    * default gap policy skips the calculation: `COUNT(x) * 2` answered NULL where SQL answers 0.
+    * `keep_values` hands the script the value instead -- 0, or 0.0 for a sum (core's `SUM` over no
+    * value, never NULL) -- and these four metrics are never `NaN` or infinite, so the script never
+    * reads one. A calculation over a `MIN`, a `MAX`, an `AVG`... keeps the default: there
+    * `keep_values` would hand the script `NaN`, which a cast turns into a value --
+    * `DATEDIFF(MAX(d), MIN(d))` reads `(long) NaN`, 0, and answered 0 days over no document instead
+    * of NULL (MEASURED on 8.18.3) -- while skipping answers NULL, as SQL does.
+    *
+    * 🔴 Elasticsearch 7.17 and later only: 6.8 refuses the policy (`Invalid gap policy:
+    * [keep_values], accepted values: [insert_zeros, skip]`), and its bridge does not send it.
+    */
+  private def keepingValues(aggs: Seq[AbstractAggregation]): Set[String] =
+    aggs
+      .collectFirst {
+        case wholeTable: KeyedFiltersAggregation
+            if wholeTable.name == SingleSearch.WholeTableHavingAgg =>
+          wholeTable.subaggs
+      }
+      .map { subaggs =>
+        val counts: Set[String] = subaggs.collect {
+          case count: ValueCountAggregation     => count.name
+          case distinct: CardinalityAggregation => distinct.name
+          case sum: SumAggregation              => sum.name
+        }.toSet
+        subaggs.collect {
+          case script: BucketScriptPipelineAgg
+              if script.bucketsPaths.values.forall(path => path == "_count" || counts(path)) =>
+            script.name
+        }.toSet
+      }
+      .getOrElse(Set.empty)
+
+  /** `body`, the JSON of `search`, with the params of its `bucket_script`s bound and the gap policy
+    * of its whole-table calculations written.
+    */
   def bind(body: String, search: SearchRequest): String = {
     val params = withParams(search.aggs)
-    if (params.isEmpty) body
+    val keepValues = keepingValues(search.aggs)
+    if (params.isEmpty && keepValues.isEmpty) body
     else {
       val root = mapper.readTree(body)
-      bind(root, params)
+      if (params.nonEmpty) bind(root, params)
+      if (keepValues.nonEmpty)
+        root.path("aggs").path(SingleSearch.WholeTableHavingAgg).path("aggs") match {
+          case wholeTable: ObjectNode =>
+            keepValues.foreach { name =>
+              wholeTable.path(name).path("bucket_script") match {
+                case bucketScript: ObjectNode => bucketScript.put("gap_policy", "keep_values")
+                case _                        =>
+              }
+            }
+          case _ =>
+        }
       mapper.writeValueAsString(root)
     }
   }

bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/ElasticAggregation.scala

@@ -65,7 +65,8 @@ import com.sksamuel.elastic4s.requests.searches.aggs.{
   NestedAggregation,
   StatsAggregation,
   TermsAggregation,
-  TermsOrder
+  TermsOrder,
+  ValueCountAggregation
 }
 import com.sksamuel.elastic4s.requests.searches.aggs.pipeline.BucketScriptPipelineAgg
 import com.sksamuel.elastic4s.requests.searches.sort.FieldSort
@@ -730,11 +731,15 @@ object ElasticAggregation {
         val script = metricSelectorForBucket(criteria, None, references)
         if (script.isEmpty) None
         else {
+          val documentCounts = documentCountNames(aggs)
           val bucketSelector =
             bucketSelectorAggregation(
               "having_filter",
               now(Script(script)),
-              extractMetricsPathForBucket(criteria, None, references)
+              readingDocumentCount(
+                extractMetricsPathForBucket(criteria, None, references),
+                documentCounts
+              )
             )
           Some(
             KeyedFiltersAggregation(
@@ -743,13 +748,91 @@ object ElasticAggregation {
                 app.softnetwork.elastic.sql.query.SingleSearch.WholeTableHavingBucket ->
                 matchAllQuery()
               ),
-              subaggs = aggs.map(_.agg) :+ bucketSelector
+              subaggs = aggs.map(agg => readingDocumentCount(agg.agg, documentCounts)) :+
+                bucketSelector
             )
           )
         }
       case _ => None
     }
 
+  /** The names of the aggregations of the whole-table bucket that count its DOCUMENTS: `COUNT(*)`
+    * (and `COUNT(_id)`, `COUNT(DISTINCT *)`), the `value_count` on `_index` that
+    * [[countAggregation]] emits -- a field every document carries exactly once.
+    */
+  private def documentCountNames(aggs: Seq[ElasticAggregation]): Set[String] =
+    aggs
+      .map(_.agg)
+      .collect {
+        case count: ValueCountAggregation
+            if count.field.contains("_index") && count.script.isEmpty =>
+          count.name
+      }
+      .toSet
+
+  /** A pipeline of the whole-table bucket reads `COUNT(*)` as the bucket's own document count,
+    * `_count` -- equal to it by construction -- because Elasticsearch never takes `_count` for a
+    * gap. Over ZERO documents every other `buckets_path` is one, so `COUNT(*) * 2` was skipped
+    * (NULL) and `HAVING COUNT(*) = 0` dropped the bucket (no row); SQL answers `0` and one row.
+    * MEASURED on 6.8.23, 7.17.29 and 8.18.3: `_count` is `0` there on every major.
+    */
+  private def readingDocumentCount(
+    paths: Map[String, String],
+    documentCounts: Set[String]
+  ): Map[String, String] =
+    if (documentCounts.isEmpty) paths
+    else
+      paths.map { case (param, path) =>
+        param -> (if (documentCounts.contains(path)) "_count" else path)
+      }
+
+  private def readingDocumentCount(
+    agg: AbstractAggregation,
+    documentCounts: Set[String]
+  ): AbstractAggregation =
+    agg match {
+      case script: BucketScriptPipelineAgg
+          if script.bucketsPaths.values.exists(documentCounts.contains) =>
+        script.copy(bucketsPaths = readingDocumentCount(script.bucketsPaths, documentCounts))
+      case other => other
+    }
+
+  /** The synthetic whole-table bucket for a calculation over aggregates with NO `GROUP BY` (issue
+    * #413, [[app.softnetwork.elastic.sql.query.SingleSearch.wholeTableCalculation]]).
+    *
+    * The calculation is a `bucket_script`, and Elasticsearch accepts one only inside a multi-bucket
+    * aggregation: at the top level of `aggs` the search was refused on every major (6.8: `Only
+    * sibling pipeline aggregations are allowed at the top level`; 7.17 and 8.18: `must be declared
+    * inside of another aggregation`). The root aggregations therefore move inside the keyed
+    * `filters` aggregation [[wholeTableHavingAggregation]] builds -- the same name, the same one
+    * `match_all` bucket, which `parseAggregations` already reads as transparent -- and the
+    * calculation is evaluated in that bucket exactly as it is in a `GROUP BY` bucket.
+    *
+    * Not under a `HAVING`: that statement is [[wholeTableHavingAggregation]] 's, and when that one
+    * declines (no condition of the predicate resolves at this level) wrapping here would hang the
+    * calculation WITHOUT the predicate -- a filter silently dropped. It keeps the emission it had.
+    */
+  def wholeTableCalculationAggregation(
+    request: app.softnetwork.elastic.sql.query.SingleSearch,
+    aggs: Seq[ElasticAggregation]
+  ): Option[Aggregation] =
+    if (
+      request.wholeTableCalculation && request.having.flatMap(_.criteria).isEmpty &&
+      aggs.exists(_.aggType.isBucketScript)
+    ) {
+      val documentCounts = documentCountNames(aggs)
+      Some(
+        KeyedFiltersAggregation(
+          app.softnetwork.elastic.sql.query.SingleSearch.WholeTableHavingAgg,
+          Seq(
+            app.softnetwork.elastic.sql.query.SingleSearch.WholeTableHavingBucket ->
+            matchAllQuery()
+          ),
+          subaggs = aggs.map(agg => readingDocumentCount(agg.agg, documentCounts))
+        )
+      )
+    } else None
+
   /** Generates the bucket_selector script for a given bucket
     */
   def metricSelectorForBucket(

bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/package.scala

@@ -233,10 +233,13 @@ package object bridge {
           // No buckets. A `HAVING` with no `GROUP BY` filters the ONE implicit whole-table group,
           // so the metric aggregations move inside a synthetic single-bucket `filters` aggregation
           // that the `having_filter` bucket_selector can hang from; without it the predicate was
-          // silently discarded (see ElasticAggregation.wholeTableHavingAggregation).
+          // silently discarded (see ElasticAggregation.wholeTableHavingAggregation). A calculation
+          // over the aggregates needs the same parent, its `bucket_script` being refused at the top
+          // level (issue #413, ElasticAggregation.wholeTableCalculationAggregation).
           case Nil =>
             ElasticAggregation
               .wholeTableHavingAggregation(request, aggs, notNestedReferences)
+              .orElse(ElasticAggregation.wholeTableCalculationAggregation(request, aggs))
               .map(Seq(_))
               .getOrElse(aggs.map(_.agg))
           case aggs =>

bridge/src/test/scala/app/softnetwork/elastic/sql/HavingFunctionEmissionSpec.scala

@@ -438,11 +438,13 @@ class HavingFunctionEmissionSpec extends AnyFlatSpec with Matchers {
   // ---------------------------------------------------------------------------------------------
 
   "a whole-table HAVING over a function of an aggregate" should "hang the selector off the synthetic bucket" in {
+    // `COUNT(*)` is read as the bucket's own `_count` there (issue #413): over no document every
+    // other path is a gap, and `HAVING COUNT(*) = 0` dropped its one row
     queryOf("SELECT COUNT(*) AS c FROM t HAVING GREATEST(COUNT(*), 0) > 1") shouldBe Seq(
       """{"query":{"match_all":{}},"size":0,"_source":false,""",
       """"aggs":{"__whole_table_having__":{"filters":{"filters":{"_all":{"match_all":{}}}},""",
       """"aggs":{"c":{"value_count":{"field":"_index"}},""",
-      """"having_filter":{"bucket_selector":{"buckets_path":{"c":"c"},""",
+      """"having_filter":{"bucket_selector":{"buckets_path":{"c":"_count"},""",
       """"script":{"source":"(params.c == null ? 0 : Math.max(params.c, 0)) > 1"}}}}}}}"""
     ).mkString
   }