Uncorrelated WHERE subqueries (IN / EXISTS / scalar / quantified) — elasticsql

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

@@ -31,10 +31,13 @@ import app.softnetwork.elastic.sql.query.{
   IsNotNullExpr,
   IsNullCriteria,
   IsNullExpr,
+  MatchAllCriteria,
   MatchCriteria,
+  MatchNoneCriteria,
   NestedElement,
   NestedElements,
-  Predicate
+  Predicate,
+  SubqueryCriteria
 }
 import com.sksamuel.elastic4s.ElasticApi._
 import com.sksamuel.elastic4s.requests.searches.queries.{InnerHit, Query}
@@ -186,6 +189,20 @@ case class ElasticBridge(filter: ElasticFilter) {
       case matchExpression: MatchCriteria => matchExpression
       case isNull: IsNullCriteria         => isNull
       case isNotNull: IsNotNullCriteria   => isNotNull
+      // Story 22.2 — the two RESOLVED sentinels a WHERE subquery collapses to.
+      case _: MatchAllCriteria  => matchAllQuery()
+      case _: MatchNoneCriteria => matchNoneQuery()
+      // 🔴 Story 22.2 — an UNRESOLVED subquery node must never reach a bridge. It is replaced by a
+      // literal criteria at `SearchApi.resolveWithSchema` (the ONE seam), so arriving here means a
+      // statement was handed straight to `singleSearch` / `singleSearchToJsonQuery`, which bypass
+      // it. Named rather than swallowed by the `Unsupported filter type` default below, because
+      // the alternative — emitting nothing for the predicate — is a silent wrong answer.
+      case s: SubqueryCriteria =>
+        throw new IllegalArgumentException(
+          s"Unresolved WHERE subquery reached the query builder: ${s.sql}. Subqueries are " +
+          "executed at SearchApi.resolveWithSchema before translation; a statement handed " +
+          "straight to singleSearch / singleSearchToJsonQuery bypasses it."
+        )
       case other =>
         throw new IllegalArgumentException(s"Unsupported filter type: ${other.getClass.getName}")
     }

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

@@ -1,7 +1,8 @@
 package app.softnetwork.elastic.sql
 
 import app.softnetwork.elastic.sql.bridge._
-import app.softnetwork.elastic.sql.query.Criteria
+import app.softnetwork.elastic.sql.operator.NOT
+import app.softnetwork.elastic.sql.query.{Criteria, InExpr, MatchAllCriteria, MatchNoneCriteria}
 import com.fasterxml.jackson.databind.JsonNode
 import com.sksamuel.elastic4s.ElasticApi.matchAllQuery
 import com.sksamuel.elastic4s.requests.searches.{SearchBodyBuilderFn, SearchRequest}
@@ -1340,4 +1341,45 @@ class SQLCriteriaSpec extends AnyFlatSpec with Matchers {
     json should not include "o.status"
   }
 
+  /** Story 22.2 — the two RESOLVED sentinels a WHERE subquery collapses to, and the loud refusal of
+    * an UNRESOLVED node.
+    *
+    * 🔴 The `terms` row pins the EXACT AST `SubqueryResolver` produces, so the resolver spec and
+    * this one share one shape: if the resolver ever built a differently typed `Values`, the emitted
+    * query would move and this row would say so.
+    */
+  private def asQueryOf(criteria: Criteria): String = {
+    import SQLImplicits._
+    implicit def timestamp: Long =
+      ZonedDateTime.parse("2025-12-31T00:00:00Z").toInstant.toEpochMilli
+    SearchBodyBuilderFn(SearchRequest("*") query criteria.asQuery()).string
+  }
+
+  it should "emit match_all / match_none for the resolved subquery sentinels (story 22.2)" in {
+    asQueryOf(MatchAllCriteria()).replaceAll("\\s", "") should include("\"match_all\":{}")
+    asQueryOf(MatchNoneCriteria()).replaceAll("\\s", "") should include("\"match_none\":{}")
+  }
+
+  it should "emit the terms clause for the resolver's IN shape (story 22.2)" in {
+    val in: Criteria =
+      InExpr(GenericIdentifier("customer_id"), LongValues(Seq(LongValue(3), LongValue(1))), None)
+    asQueryOf(in).replaceAll("\\s", "") should include("\"terms\":{\"customer_id\":[3,1]}")
+    asQueryOf(
+      InExpr(GenericIdentifier("customer_id"), LongValues(Seq(LongValue(3))), Some(NOT))
+    ).replaceAll("\\s", "") should include("\"must_not\"")
+  }
+
+  it should "refuse an UNRESOLVED subquery node BY NAME, never silently (story 22.2)" in {
+    implicit def timestamp: Long = 0L
+    val node = parser
+      .Parser("SELECT id FROM t WHERE a IN (SELECT a FROM u)")
+      .toOption
+      .collect { case s: query.SingleSearch => s }
+      .flatMap(_.where.flatMap(_.criteria))
+      .getOrElse(fail("expected a WHERE subquery"))
+    val ex = the[IllegalArgumentException] thrownBy node.asQuery()
+    ex.getMessage should include("Unresolved WHERE subquery")
+    ex.getMessage should include("resolveWithSchema")
+  }
+
 }

build.sbt

@@ -20,7 +20,7 @@ ThisBuild / organization := "app.softnetwork"
 
 name := "softclient4es"
 
-ThisBuild / version := "0.23.0"
+ThisBuild / version := "0.24.0-SNAPSHOT"
 
 ThisBuild / scalaVersion := scala213
 

core/src/main/resources/help/commands/dql/select.json

@@ -9,6 +9,17 @@
     "[JOIN table ON condition]",
     "[JOIN (SELECT ...) [AS] alias ON condition]",
     "[WHERE condition]",
+    "",
+    "-- A WHERE condition may carry an UNCORRELATED subquery. The inner query runs FIRST and its",
+    "-- values are pushed into the outer query as a terms query or a literal comparison:",
+    "SELECT ... WHERE col IN (SELECT col FROM ...)",
+    "SELECT ... WHERE col NOT IN (SELECT col FROM ...)",
+    "SELECT ... WHERE EXISTS (SELECT 1 FROM ...)",
+    "SELECT ... WHERE NOT EXISTS (SELECT 1 FROM ...)",
+    "SELECT ... WHERE col > (SELECT MAX(col) FROM ...)",
+    "SELECT ... WHERE col = ANY (SELECT col FROM ...)",
+    "SELECT ... WHERE col > ALL (SELECT col FROM ...)",
+    "",
     "[GROUP BY columns]",
     "[HAVING condition]",
     "[ORDER BY columns [ASC|DESC]]",
@@ -34,7 +45,12 @@
       "name": "JOIN",
       "description": "Combine rows from multiple tables",
       "optional": true,
-      "variants": ["LEFT JOIN", "RIGHT JOIN", "INNER JOIN", "OUTER JOIN"]
+      "variants": [
+        "LEFT JOIN",
+        "RIGHT JOIN",
+        "INNER JOIN",
+        "OUTER JOIN"
+      ]
     },
     {
       "name": "WHERE",
@@ -55,7 +71,12 @@
       "name": "ORDER BY",
       "description": "Sort result rows",
       "optional": true,
-      "modifiers": ["ASC", "DESC", "NULLS FIRST", "NULLS LAST"]
+      "modifiers": [
+        "ASC",
+        "DESC",
+        "NULLS FIRST",
+        "NULLS LAST"
+      ]
     },
     {
       "name": "UNION ALL",
@@ -104,19 +125,33 @@
       "description": "Aggregate in a subquery, filter the aggregate outside",
       "sql": "SELECT d.category, d.total FROM (SELECT category, SUM(amount) AS total FROM orders GROUP BY category) AS d WHERE d.total > 100",
       "output": null
+    },
+    {
+      "title": "Subquery in WHERE",
+      "description": "Filter by a set computed by another query: the subquery runs first and its values are pushed into the outer query",
+      "sql": "SELECT id, amount FROM orders WHERE customer_id IN (SELECT id FROM customers WHERE region = 'EU')",
+      "output": null
     }
   ],
   "notes": [
     "An alias must not be a reserved word: count, day, hour, now, today, offset, distance and every function name are rejected as aliases",
     "Result sets larger than the index max_result_window are paged automatically; the query itself needs no scroll or search_after handling",
     "Nested fields are addressed with JOIN UNNEST(<path>) AS <alias>; see the JOIN clause above",
     "A FROM-less SELECT is evaluated against a hidden softclient4es_handshake index created on first use, so a read-only account needs it pre-created",
-    "A derived table (a SELECT in FROM or JOIN) must carry an alias; a SELECT * body exposes an unchecked projection"
+    "A derived table (a SELECT in FROM or JOIN) must carry an alias; a SELECT * body exposes an unchecked projection",
+    "A WHERE subquery must be self-contained: a reference to an outer alias (a correlated subquery) is refused until the relational engine executes it, and a bare column name inside the subquery is read as the subquery's own column",
+    "NOT IN follows SQL: when the subquery's values contain NULL, no row matches. ANY/SOME over an empty subquery is false and ALL over an empty subquery is true"
   ],
   "limitations": [
-    "Derived tables parse and are validated but execute only through the relational engine (softclient4es-arrow-extensions); without it the statement is refused with HTTP 400"
+    "Derived tables parse and are validated but execute only through the relational engine (softclient4es-arrow-extensions); without it the statement is refused with HTTP 400",
+    "A WHERE subquery may return at most 65536 distinct values (Elasticsearch index.max_terms_count); a larger set must be written as a JOIN",
+    "Subqueries are accepted in WHERE only (SELECT, DELETE, UPDATE): not in HAVING, CASE, JOIN ON, MATERIALIZED VIEW or WATCHER definitions; UNION ALL and FROM-less subquery bodies are rejected, and NOT IN over a GROUP BY subquery cannot see a NULL group"
+  ],
+  "seeAlso": [
+    "INSERT",
+    "UPDATE",
+    "DELETE"
   ],
-  "seeAlso": ["INSERT", "UPDATE", "DELETE"],
   "minVersion": null,
   "aliases": []
 }

core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala

@@ -319,6 +319,11 @@ trait ElasticConversion {
     val aggsNode = Option(json.path("aggregations"))
       .filter(!_.isMissingNode)
 
+    // Story 22.2 — how many documents a top-level aggregation actually aggregated over. `size: 0`
+    // still reports `hits.total`, so this is present on every aggregation response and on every
+    // major (a bare number on ES 6, `{value, relation}` on ES 7+).
+    val rootDocCount: Option[Long] = docCountOf(json.path("hits"), "total")
+
     val rows = (hitsNode, aggsNode) match {
       case (Some(hits), None) if hits.nonEmpty =>
         // Case 1 : only hits
@@ -328,12 +333,12 @@ trait ElasticConversion {
 
       case (None, Some(aggs)) =>
         // Case 2 : only aggregations
-        val ret = parseAggregations(aggs, rowInvariants, fieldAliases, aggregations)
+        val ret = parseAggregations(aggs, rowInvariants, fieldAliases, aggregations, rootDocCount)
         combineAggregationRows(ret)
 
       case (Some(hits), Some(aggs)) if hits.isEmpty =>
         // Case 3 : aggregations with no hits
-        val ret = parseAggregations(aggs, rowInvariants, fieldAliases, aggregations)
+        val ret = parseAggregations(aggs, rowInvariants, fieldAliases, aggregations, rootDocCount)
         combineAggregationRows(ret)
 
       case (Some(hits), Some(aggs)) if hits.nonEmpty =>
@@ -635,11 +640,17 @@ trait ElasticConversion {
 
   /** Parse recursively aggregations from Elasticsearch response with parent context
     */
+  /** @param docCount
+    *   how many documents the enclosing scope aggregated over, threaded so an aggregate over ZERO
+    *   documents can answer ANSI NULL (story 22.2). `hits.total` at the root, a bucket's
+    *   `doc_count` inside one, a wrapper aggregation's own `doc_count` under that.
+    */
   def parseAggregations(
     aggsNode: JsonNode,
     parentContext: ListMap[String, Any],
     fieldAliases: ListMap[String, String],
-    aggregations: ListMap[String, ClientAggregation]
+    aggregations: ListMap[String, ClientAggregation],
+    docCount: Option[Long] = None
   ): Seq[ListMap[String, Any]] = {
 
     if (aggsNode.isMissingNode || !aggsNode.isObject) {
@@ -721,11 +732,17 @@ trait ElasticConversion {
         }
 
         // Recursively parse subaggregations
-        parseAggregations(subAggsNode, currentContext, fieldAliases, aggregations)
+        parseAggregations(
+          subAggsNode,
+          currentContext,
+          fieldAliases,
+          aggregations,
+          docCountOf(aggValue, "doc_count").orElse(docCount)
+        )
       }
     } else if (bucketAggs.isEmpty) {
       // No buckets : it is a leaf aggregation (metrics or top_hits)
-      val metrics = extractMetrics(aggsNode, aggregations)
+      val metrics = extractMetrics(aggsNode, aggregations, docCount)
       val allTopHits = extractAllTopHits(aggsNode, fieldAliases, aggregations)
 
       if (allTopHits.nonEmpty) {
@@ -739,7 +756,8 @@ trait ElasticConversion {
       // Handle each aggregation with buckets
       bucketAggs.flatMap { case (aggName, buckets, _) =>
         buckets.flatMap { bucket =>
-          val metrics = extractMetrics(bucket, aggregations)
+          val bucketDocCount = docCountOf(bucket, "doc_count")
+          val metrics = extractMetrics(bucket, aggregations, bucketDocCount)
           val allTopHits = extractAllTopHits(bucket, fieldAliases, aggregations)
 
           val bucketKey = extractBucketKey(bucket)
@@ -775,7 +793,13 @@ trait ElasticConversion {
             /*subAggFields.foreach { entry =>
               subAggsNode.set(entry.getKey, entry.getValue) // FIXME
             }*/
-            parseAggregations(subAggsNode, currentContext, fieldAliases, aggregations)
+            parseAggregations(
+              subAggsNode,
+              currentContext,
+              fieldAliases,
+              aggregations,
+              bucketDocCount
+            )
           } else {
             Seq(currentContext)
           }
@@ -784,6 +808,21 @@ trait ElasticConversion {
     }
   }
 
+  /** The document count a scope aggregated over, when the response states it.
+    *
+    * Handles both shapes Elasticsearch uses for `hits.total`: a bare number (ES 6) and `{"value":
+    * n, "relation": "eq"}` (ES 7+). A `doc_count` is always a bare number.
+    */
+  private[client] def docCountOf(node: JsonNode, field: String): Option[Long] = {
+    val n = node.path(field)
+    if (n.isMissingNode) None
+    else if (n.isNumber) Some(n.asLong())
+    else if (n.isObject) {
+      val v = n.path("value")
+      if (v.isNumber) Some(v.asLong()) else None
+    } else None
+  }
+
   /** Extract the bucket key with proper typing (String, Long, Double, DateTime, etc.)
     */
   def extractBucketKey(bucket: JsonNode): Any = {
@@ -902,9 +941,15 @@ trait ElasticConversion {
 
   /** Extract metrics from an aggregation node
     */
+  /** @param docCount
+    *   how many documents the enclosing scope aggregated over, when the response states it
+    *   (`hits.total` at the root, a bucket's `doc_count` inside one). `Some(0)` is what makes an
+    *   aggregate ANSI-NULL — see [[ClientAggregation.nullOverEmptyInput]].
+    */
   def extractMetrics(
     aggsNode: JsonNode,
-    aggregations: ListMap[String, ClientAggregation]
+    aggregations: ListMap[String, ClientAggregation],
+    docCount: Option[Long] = None
   ): ListMap[String, Any] = {
     aggsNode match {
       case n: ObjectNode =>
@@ -917,8 +962,25 @@ trait ElasticConversion {
               bucketRoot = Some(agg.bucketRoot)
             case _ =>
           }
-          // Detect simple metric values
+          // 🔴 Story 22.2 — ANSI: an aggregate computed over ZERO documents is NULL (COUNT and SUM
+          // excepted — `ClientAggregation.nullOverEmptyInput` is the ONE place that rule lives).
+          //
+          // This arm comes FIRST, before the value is read, because on ES 8 the value CANNOT be
+          // trusted here: that module reads the TYPED response, `SingleMetricAggregateBase` holds a
+          // primitive `double`, and Elasticsearch's `null` has already become `0.0` inside the
+          // vendor's model before any of our code runs. MEASURED: identical wire responses on 6.8 /
+          // 7.17 / 8.18 / 9.0 (`{"value":null}`), but ES 8 alone converted it to `0.0` — so
+          // `WHERE x > (SELECT MAX(y) FROM t WHERE <no match>)` reduced to `x > 0` and returned
+          // EVERY row there while returning none elsewhere.
+          //
+          // The document count is the recoverable signal and it is EXACT, not a heuristic: a
+          // genuine `MAX` of `0.0` needs at least one document, where this rule cannot fire. On the
+          // majors that were already correct the value is `null` anyway, so this is a no-op for
+          // them beyond making the NULL explicit rather than an absent key.
+          val emptyInput =
+            docCount.contains(0L) && aggregations.get(name).exists(_.nullOverEmptyInput)
           Option(value.get("value"))
+            .filter(_ => !emptyInput)
             .filter(!_.isNull)
             .map { metricValue =>
               val numericValue = if (metricValue.isIntegralNumber) {
@@ -989,7 +1051,12 @@ trait ElasticConversion {
               } else {
                 None
               }
-            } match {
+            }
+            // An empty input yields an EXPLICIT null column rather than an absent key, so every
+            // consumer sees the same thing on every major (an absent key was the pre-existing
+            // behaviour on 6.8 / 7.17 / 9.0 and reads as NULL only because `rowNormalizer`
+            // null-fills under NativeContext).
+            .orElse(if (emptyInput) Some(name -> (null: Any)) else None) match {
             case Some(m) =>
               // Skip auxiliary aggregations (from HAVING/WHERE/ORDER BY only, not in SELECT)
               val isAuxiliary = aggregations.get(m._1).exists(_.auxiliary)

core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala

@@ -321,7 +321,11 @@ trait ScrollApi extends ElasticClientHelpers with SchemaCacheTtlApi {
           ElasticQuery(
             single,
             collection.immutable.Seq(single.sources: _*),
-            sql = Some(single.sql),
+            // Story 22.2 — the statement AS WRITTEN, not the resolved one, exactly as `search`
+            // has always done (`sql = Some(query)` there). A resolved WHERE subquery carries up to
+            // 65,536 literals, and this render reaches the `Row query …` INFO line and
+            // `ElasticResponse.sql`.
+            sql = Some(parsed.sql),
             explodeNested = single.explodeNested
           )
         scrollWithMetrics(