feat(sql,core): correlated subqueries — cross-scope resolution and routing to the relational engine (story 22.3, elasticsql half)

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

@@ -139,7 +139,7 @@
     "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 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",
+    "A WHERE subquery that reads an outer alias (a correlated subquery) executes through the relational engine shipped in softclient4es-arrow-extensions, and is refused with HTTP 400 at every venue without it. An outer reference must be QUALIFIED with the outer table's alias - a bare column name inside the subquery is read as the subquery's own column - and an object path inside the body should be qualified with the body's own alias",
     "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": [

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

@@ -17,7 +17,7 @@
 package app.softnetwork.elastic.client
 
 import app.softnetwork.elastic.client.result.ElasticError
-import app.softnetwork.elastic.sql.query.{derivedTablesPresent, Statement}
+import app.softnetwork.elastic.sql.query.{closureSearches, derivedTablesPresent, Statement}
 
 /** The ONE rejection every venue WITHOUT the relational engine emits for a closure-shaped statement
   * — a cross-index JOIN or a derived table (epic 22 AD-5; #157's discipline, widened).
@@ -38,9 +38,28 @@ object RelationalClosureGuard {
     * did not choose.
     */
   def shapeOf(statement: Statement): String =
-    if (derivedTablesPresent(statement)) "A derived table (subquery in FROM/JOIN)"
+    if (closureSearches(statement).exists(_.hasCorrelatedSubqueries)) CorrelatedShape
+    else if (derivedTablesPresent(statement)) "A derived table (subquery in FROM/JOIN)"
     else "A cross-index JOIN"
 
+  /** Story 22.3 — reported FIRST, for the same reason the derived table outranks the JOIN: it is
+    * the construct with the narrowest remedy (qualify differently, or rewrite as a JOIN), and a
+    * statement that carries both is refused for the one the analyst is least likely to have chosen.
+    */
+  private val CorrelatedShape =
+    "A correlated subquery (a WHERE subquery that reads an outer alias)"
+
+  /** The same rejection for a caller that holds a NODE rather than a statement — story 22.3b's
+    * defensive arm in `SubqueryResolver`, which is reached only through the
+    * `GatewayApi.run(statement: Statement)` path that never validates. ONE message, never two.
+    */
+  def correlatedRejection: ElasticError =
+    ElasticError(
+      message = messageFor(CorrelatedShape),
+      statusCode = Some(400),
+      operation = Some("search")
+    )
+
   /** `operation` stays `"join"` on the gateway path for BOTH shapes: nothing downstream
     * distinguishes them (the JDBC driver relays `message` verbatim), so changing it would be a
     * second behaviour change with no consumer.
@@ -54,13 +73,15 @@ object RelationalClosureGuard {
     */
   def rejection(statement: Statement, operation: String = "join"): ElasticError =
     ElasticError(
-      message =
-        s"${shapeOf(statement)} requires the relational engine shipped in the $ExtensionJar jar " +
-        "(Java 11+); this venue has none, so the statement is refused rather than executed " +
-        s"against the first index it names. Put $ExtensionJar on the classpath (at the REPL: " +
-        "re-run the installer, or drop --no-extensions). See " +
-        "documentation/client/repl.md#extensions-cross-index-joins-materialized-views.",
+      message = messageFor(shapeOf(statement)),
       statusCode = Some(400),
       operation = Some(operation)
     )
+
+  private def messageFor(shape: String): String =
+    s"$shape requires the relational engine shipped in the $ExtensionJar jar " +
+    "(Java 11+); this venue has none, so the statement is refused rather than executed " +
+    s"against the first index it names. Put $ExtensionJar on the classpath (at the REPL: " +
+    "re-run the installer, or drop --no-extensions). See " +
+    "documentation/client/repl.md#extensions-cross-index-joins-materialized-views."
 }

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

@@ -253,21 +253,76 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
   private[client] def innerColumnType(inner: SingleSearch, name: String): Option[SQLType] =
     resolvedSchema(inner).flatMap(_.find(name).map(_.dataType))
 
-  /** Story 22.2 (PD-2) — the SCHEMA-aware half of the correlation rule.
+  /** Story 22.3 (AD-3) — the SCHEMA-aware SCOPE check, generalised from story 22.2's
+    * `bareNameCorrelation`. Reached only for a statement the closure guard did NOT already route to
+    * the relational engine, so it answers the two questions parse time provably cannot:
     *
-    * A QUALIFIED reference to an outer alias is caught structurally at parse time
-    * (`SubqueryScope`). A BARE one cannot be: SQL resolves a bare name innermost-first, so `cid`
-    * inside the subquery is the INNER column whenever the inner index has it — and that is the
-    * right reading in the overwhelming majority of statements. When BOTH mappings are in hand,
-    * though, a bare name the inner index does NOT map while the outer one DOES is a correlated
-    * reference beyond reasonable doubt, and executing it as uncorrelated would send an unknown
-    * field to Elasticsearch and answer zero rows with HTTP 200 — the silent-wrong-answer mode epic
-    * 22 exists to close.
+    *   1. a BARE name the inner index does not map while the outer one DOES. SQL resolves a bare
+    *      name innermost-first, so `cid` inside the subquery is the INNER column whenever the inner
+    *      index has it — the right reading in the overwhelming majority of statements — but when
+    *      BOTH mappings are in hand the other case is a correlated reference beyond reasonable
+    *      doubt, and executing it as uncorrelated would send an unknown field to Elasticsearch and
+    *      answer zero rows with HTTP 200. It is refused with the remedy that makes it EXECUTABLE
+    *      since story 22.3b: qualify it, and the statement routes. 2. a QUALIFIED name that
+    *      resolves in NO scope of the chain and is not a mapped object field of the inner index
+    *      either. At parse time those two are indistinguishable (story 22.2's PD-2 assumes the
+    *      object path, which is why no arm rejects it there); with the mapping in hand they are
+    *      not, and the message NAMES every scope it searched.
     *
     * Any schema-absent condition answers `None` (assume inner — PD-2's documented boundary), so
     * this never turns a schema outage into a rejection.
     */
-  private[client] def bareNameCorrelation(
+  private[client] def scopeCorrelation(
+    outer: SingleSearch,
+    inner: SingleSearch
+  ): Option[String] =
+    bareNameCorrelation(outer, inner).orElse(unresolvedQualifier(outer, inner))
+
+  /** Question 2 of [[scopeCorrelation]]. `Schema.find(head)` answers for an OBJECT field too (it
+    * returns the object column itself), so one lookup separates `address.city` from a typo.
+    */
+  private def unresolvedQualifier(
+    outer: SingleSearch,
+    inner: SingleSearch
+  ): Option[String] =
+    // 🔴 An EMPTY mapping is a mapping GAP, never evidence that a name does not exist: an index
+    // created but not yet written to, or one whose dynamic mapping has not caught up, would
+    // otherwise turn every dotted object path in a subquery body into a 400. Same posture as the
+    // schema-absent conditions in `resolvedSchema` — a mapping we do not have must never become a
+    // rejection.
+    resolvedSchema(inner).filter(_.columns.nonEmpty).flatMap { innerMapping =>
+      val chain = SubqueryScope.chain(inner, Seq(outer))
+      inner.referencedIdentifiers.iterator
+        .filter(id =>
+          id.tableAlias.isEmpty && id.table.isEmpty && !id.nested && id.name.contains(".")
+        )
+        .flatMap { id =>
+          SubqueryScope.resolve(id, chain) match {
+            // 🔴 The resolver sees a name the DETECTOR did not report. That is possible because the
+            // two used to read different maps, and it is exactly the shape that answers HTTP 200
+            // with zero rows if it executes: Elasticsearch reads `o.region` as an object path.
+            // `correlationNames` now derives from `scopeOf`, so this should be unreachable —
+            // it is the belt for the day the two drift again, and it fails LOUD, never silent.
+            case SubqueryScope.Resolved(depth, _, _) if depth >= 1 =>
+              Some(
+                SubqueryScope.bareCorrelatedMessage(
+                  id.name,
+                  inner.sources.headOption.getOrElse("the subquery's table"),
+                  outer.sources.headOption.getOrElse("the outer table")
+                )
+              )
+            // Unresolved: a typo, or an object path. Only the mapping can tell, and it does.
+            case SubqueryScope.Unresolved
+                if innerMapping.find(id.name.split("\\.", 2)(0)).isEmpty =>
+              Some(SubqueryScope.unresolvedMessage(id, chain, inner.sql))
+            case _ => None
+          }
+        }
+        .toSeq
+        .headOption
+    }
+
+  private def bareNameCorrelation(
     outer: SingleSearch,
     inner: SingleSearch
   ): Option[String] =

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

@@ -85,7 +85,7 @@ object SubqueryResolver {
     resolve(
       single,
       s => api.search(s)(EntityContext), // an absent key stays ABSENT — never null-filled
-      api.bareNameCorrelation(single, _),
+      api.scopeCorrelation(single, _),
       api.innerColumnType
     )
 
@@ -154,6 +154,15 @@ object SubqueryResolver {
       // Unreachable after `validate()`; `GatewayApi.run(statement)` does not validate a
       // programmatically built statement, so the arm is a named 400 rather than a MatchError.
       case None => Left(bad(s"Unsupported subquery body in ${node.sql}"))
+      // Story 22.3b — DEFENSIVE, and UNREACHABLE through every route that exists today: a node with
+      // non-empty `correlatedRefs` implies `hasCorrelatedSubqueries`, which both
+      // `SearchApi.resolveWithSchema` (before phase one) and `CoreDqlExtension.execute` refuse
+      // first. It is kept because the invariant it protects is a SILENT wrong answer if it ever
+      // breaks — executing a correlated body as if it were self-contained — and because a future
+      // caller of this object need not know the guard exists one layer up. Do not read it as
+      // evidence of an open hole.
+      case Some(_) if node.correlatedRefs.nonEmpty =>
+        Left(RelationalClosureGuard.correlatedRejection)
       case Some(inner) =>
         correlation(inner) match {
           case Some(reason) => Left(bad(reason))

core/src/test/scala/app/softnetwork/elastic/client/RelationalClosureGuardSpec.scala

@@ -62,6 +62,8 @@ class RelationalClosureGuardSpec extends AnyFlatSpec with Matchers {
 
   private val DerivedSelect = "SELECT COL FROM (SELECT 1 AS COL) AS d"
   private val JoinSelect = "SELECT o.id, c.name FROM orders o JOIN customers c ON o.cid = c.id"
+  private val CorrelatedSelect =
+    "SELECT c.id FROM customers c WHERE EXISTS (SELECT 1 FROM orders o WHERE o.customer_id = c.id)"
 
   // ---- the ONE seam ------------------------------------------------------------------------
 
@@ -192,6 +194,45 @@ class RelationalClosureGuardSpec extends AnyFlatSpec with Matchers {
     RelationalClosureGuard.shapeOf(stmt) should include("derived table")
   }
 
+  /** Story 22.3b — the correlated shape is reported FIRST when several are present, for the same
+    * reason the derived table outranks the JOIN: it carries the narrowest remedy.
+    */
+  it should "name the CORRELATED shape, and prefer it over the others" in {
+    val correlated = searchStatement(CorrelatedSelect)
+    relationalClosureRequired(correlated) shouldBe true
+    RelationalClosureGuard.shapeOf(correlated) should include("A correlated subquery")
+    val both = searchStatement(
+      "SELECT o.id FROM orders o JOIN customers c ON o.cid = c.id " +
+      "WHERE EXISTS (SELECT 1 FROM refunds r WHERE r.oid = o.id)"
+    )
+    RelationalClosureGuard.shapeOf(both) should include("A correlated subquery")
+  }
+
+  it should "refuse a correlated statement at the seam, before phase one ever runs" in {
+    val err = refusalOf(client().search(searchStatement(CorrelatedSelect)))
+    err.statusCode shouldBe Some(400)
+    err.operation shouldBe Some("search")
+    err.message should include("A correlated subquery")
+    err.message should include(RelationalClosureGuard.ExtensionJar)
+    // 🔴 falsifiable in the right direction: the resolver would have reported the INNER statement's
+    // own failure ("Subquery …", as the uncorrelated row above asserts). Seeing the closure message
+    // instead is what proves the guard ran FIRST.
+    err.message should not include "Subquery"
+  }
+
+  it should "give DELETE ... WHERE EXISTS (correlated) the same refusal" in {
+    val err = refusalOf(
+      client()
+        .asInstanceOf[IndicesApi]
+        .deleteByQuery(
+          "orders",
+          "DELETE FROM orders WHERE EXISTS (SELECT 1 FROM refunds r WHERE r.oid = orders.id)"
+        )
+    )
+    err.statusCode shouldBe Some(400)
+    err.message should include("A correlated subquery")
+  }
+
   it should "say the statement was refused rather than executed against the first index (PD-2)" in {
     val msg = RelationalClosureGuard.rejection(searchStatement(JoinSelect)).message
     msg should include("refused rather than executed against the first index it names")