feat(sql,core): FROM-less SELECT handshake executed against Elasticsearch (story 20.9)

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

@@ -56,6 +56,7 @@ import app.softnetwork.elastic.sql.query.{
   DropWatcher,
   EnrichPolicyStatement,
   ExecuteEnrichPolicy,
+  FromlessSelect,
   Insert,
   LicenseStatement,
   MultiSearch,
@@ -541,16 +542,23 @@ class TableExecutor(
             Future.successful(
               ElasticResult.success(
                 QueryRows(
-                  mappings.map { case (index, mappings) =>
-                    ListMap(
-                      "name" -> index,
-                      "type" -> mappings.tableType.name.toUpperCase,
-                      "pk"   -> mappings.primaryKey.mkString(","),
-                      "partitioned" -> mappings.partitionBy
-                        .map(p => s"PARTITION BY ${p.column} (${p.granularity})")
-                        .getOrElse("")
-                    )
-                  }.toSeq
+                  mappings
+                    // Issue #251/AD-10 — the handshake index is infrastructure, not a table:
+                    // never listed, ANY pattern. This one seam also covers jdbc getTables and
+                    // Flight GET_TABLES (both execute SHOW TABLES through gateway.run).
+                    // DESCRIBE TABLE deliberately still works on it.
+                    .filterNot { case (index, _) => index == GatewayApi.HandshakeIndex }
+                    .map { case (index, mappings) =>
+                      ListMap(
+                        "name" -> index,
+                        "type" -> mappings.tableType.name.toUpperCase,
+                        "pk"   -> mappings.primaryKey.mkString(","),
+                        "partitioned" -> mappings.partitionBy
+                          .map(p => s"PARTITION BY ${p.column} (${p.granularity})")
+                          .getOrElse("")
+                      )
+                    }
+                    .toSeq
                 )
               )
             )
@@ -1677,14 +1685,49 @@ class LicenseExecutor(
     exp.map(_.toString).getOrElse("never")
 }
 
+/** Issue #251 (story 20.9) — FROM-less SELECT: Painless handshake AGAINST Elasticsearch. The
+  * statement's LIMIT/OFFSET are applied engine-side on the one assembled row, AFTER the ES
+  * round-trip (AD-12) — `LIMIT 0` still connection-checks (AD-8), and the internal rewrite always
+  * carries its own `LIMIT 1` so it can never reach scroll/PIT.
+  */
+class FromlessSelectExecutor(
+  evaluator: HandshakeEvaluator,
+  logger: Logger
+) extends Executor[FromlessSelect] {
+
+  override def execute(
+    statement: FromlessSelect
+  )(implicit system: ActorSystem): Future[ElasticResult[QueryResult]] = {
+    implicit val ec: ExecutionContext = system.dispatcher
+    // run(statement) never calls validate(): a programmatic FromlessSelect reaches this
+    // executor without the parser's gate — re-run the guards (surviving review critical #5).
+    statement.validate() match {
+      case Left(reason) =>
+        val error =
+          ElasticError(message = reason, statusCode = Some(400), operation = Some("sql"))
+        logger.error(s"❌ ${error.message}")
+        Future.successful(ElasticFailure(error))
+      case Right(_) =>
+        evaluator.evaluateHandshake(statement).map {
+          case ElasticSuccess(row) =>
+            val offset = statement.limit.flatMap(_.offset).map(_.offset).getOrElse(0)
+            val max = statement.limit.map(_.limit).getOrElse(1)
+            ElasticSuccess(QueryRows(Seq(row).drop(offset).take(max)))
+          case ElasticFailure(error) => ElasticFailure(error)
+        }
+    }
+  }
+}
+
 class DqlRouterExecutor(
   searchExec: SearchExecutor,
   pipelineExec: PipelineExecutor,
   tableExec: TableExecutor,
   watcherExec: WatcherExecutor,
   policyExec: EnrichPolicyExecutor,
   clusterExec: ClusterExecutor,
-  licenseExec: LicenseExecutor
+  licenseExec: LicenseExecutor,
+  fromlessExec: FromlessSelectExecutor // issue #251 — see the story 20.9 AD-4′ arity note
 ) extends Executor[DqlStatement] {
 
   override def execute(
@@ -1698,6 +1741,8 @@ class DqlRouterExecutor(
     case e: EnrichPolicyStatement => policyExec.execute(e)
     case c: ClusterStatement      => clusterExec.execute(c)
     case l: LicenseStatement      => licenseExec.execute(l)
+    // Issue #251 — FROM-less SELECT: Painless handshake AGAINST Elasticsearch.
+    case f: FromlessSelect => fromlessExec.execute(f)
 
     case _ =>
       Future.successful(
@@ -1775,14 +1820,27 @@ trait GatewayApi extends IndicesApi with ElasticClientHelpers {
     strategy = licenseRefreshStrategy // No longer Option — NopRefreshStrategy for Community
   )
 
+  /** Issue #251 — the seam behind the FROM-less SELECT handshake. Today's backend searches the
+    * dedicated handshake index; a future Painless-execute-API backend swaps in HERE, touching
+    * nothing else (AD-4′).
+    */
+  lazy val handshakeEvaluator: HandshakeEvaluator =
+    new SearchHandshakeEvaluator(api = this, logger = logger)
+
+  lazy val fromlessSelectExecutor = new FromlessSelectExecutor(
+    evaluator = handshakeEvaluator,
+    logger = logger
+  )
+
   lazy val dqlExecutor = new DqlRouterExecutor(
     searchExec = searchExecutor,
     pipelineExec = pipelineExecutor,
     tableExec = tableExecutor,
     watcherExec = watcherExecutor,
     policyExec = policyExecutor,
     clusterExec = clusterExecutor,
-    licenseExec = licenseExecutor
+    licenseExec = licenseExecutor,
+    fromlessExec = fromlessSelectExecutor
   )
 
   lazy val ddlExecutor = new DdlRouterExecutor(
@@ -2075,6 +2133,27 @@ object GatewayApi {
     s"$ParseRejectionPrefix [${excerpt(statement)}]: $safeReason"
   }
 
+  /** Issue #251 — the dedicated FROM-less-handshake index. NOT dot-prefixed (ES 8+
+    * deprecation-warns dot-prefixed non-system creation — and the deprecation LOG itself creates a
+    * `.ds-*` index, the measured all_templates trap); product-prefixed so operators can attribute
+    * it; excluded from SHOW TABLES by TableExecutor (AD-10).
+    */
+  val HandshakeIndex: String = "softclient4es_handshake"
+
+  /** 1 shard / 0 replicas: single-node clusters stay green. NEVER defaultSettings (ngram). */
+  private[client] val HandshakeSettings: String =
+    """{"index": {"number_of_shards": 1, "number_of_replicas": 0}}"""
+
+  /** index.hidden exists only from ES 7.7 — used when the cluster supports it (AD-9/AD-10). */
+  private[client] val HandshakeSettingsHidden: String =
+    """{"index": {"number_of_shards": 1, "number_of_replicas": 0, "hidden": true}}"""
+
+  private[client] val HandshakeMapping: String =
+    """{"properties": {"dummy": {"type": "keyword"}}}"""
+
+  private[client] val HandshakeDocId: String = "1"
+  private[client] val HandshakeDoc: String = """{"dummy": "dummy"}"""
+
   /** Split a normalized SQL string into statements on top-level `;`.
     *
     * This replaces `split(";\\s*$")`, which — `$` anchoring to the end of the whole (newline-free)

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

@@ -0,0 +1,237 @@
+/*
+ * Copyright 2025 SOFTNETWORK
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package app.softnetwork.elastic.client
+
+import akka.actor.ActorSystem
+import app.softnetwork.elastic.client.result._
+import app.softnetwork.elastic.sql.query.{FromlessSelect, SingleSearch}
+import org.slf4j.Logger
+
+import scala.collection.immutable.ListMap
+import scala.concurrent.{ExecutionContext, Future}
+import scala.util.Try
+
+/** THE SEAM (issue #251, AD-4′): "evaluate this FROM-less select-list's Painless against the
+  * cluster and give me ONE row". Today's backend searches the dedicated handshake index
+  * (script_fields on the seeded doc); the recorded future backend is the Painless execute API
+  * (`_scripts/painless/_execute`, no index needed) — swapping it must touch nothing but this
+  * trait's implementation wiring in GatewayApi.
+  */
+trait HandshakeEvaluator {
+  def evaluateHandshake(statement: FromlessSelect)(implicit
+    system: ActorSystem
+  ): Future[ElasticResult[ListMap[String, Any]]]
+}
+
+/** Search-backed implementation: lazily ensures the handshake index (probe-before-act, race-safe,
+  * memoized — AD-9), rewrites the statement to `SELECT <items> FROM <handshake> LIMIT 1` (AD-3′)
+  * and executes it through the UNMODIFIED FROM-ful pipeline, then assembles one row: script_fields
+  * arrays unwrapped, `__cN` keys renamed to the PD-2 output names (AD-11).
+  */
+class SearchHandshakeEvaluator(
+  api: SearchApi with IndicesApi with IndexApi with RefreshApi with VersionApi,
+  logger: Logger
+) extends HandshakeEvaluator {
+
+  import GatewayApi._
+
+  @volatile private[this] var ready: Boolean = false
+
+  /** Test seam only. */
+  private[client] def isReady: Boolean = ready
+
+  override def evaluateHandshake(statement: FromlessSelect)(implicit
+    system: ActorSystem
+  ): Future[ElasticResult[ListMap[String, Any]]] = {
+    implicit val ec: ExecutionContext = system.dispatcher
+    implicit val context: ConversionContext = NativeContext
+    ensureHandshakeIndex() match {
+      case ElasticFailure(error) => Future.successful(ElasticFailure(error))
+      case ElasticSuccess(_)     => runSearch(statement, retriesLeft = 1)
+    }
+  }
+
+  private def runSearch(statement: FromlessSelect, retriesLeft: Int)(implicit
+    ec: ExecutionContext,
+    context: ConversionContext
+  ): Future[ElasticResult[ListMap[String, Any]]] = {
+    val single = statement.toSingleSearch(HandshakeIndex)
+    api.searchAsync(single).flatMap {
+      case ElasticSuccess(response) =>
+        response.results.headOption match {
+          case Some(raw) =>
+            Future.successful(ElasticSuccess(assembleRow(statement, single, raw)))
+          case None if retriesLeft > 0 =>
+            // Pre-created-but-unseeded index, or the seed doc was deleted out-of-band:
+            // re-seed ONCE, then loud (never an empty-but-successful answer — #253 family).
+            ready = false
+            ensureHandshakeIndex(forceSeed = true) match {
+              case ElasticFailure(error) => Future.successful(ElasticFailure(error))
+              case _                     => runSearch(statement, retriesLeft - 1)
+            }
+          case None =>
+            Future.successful(ElasticFailure(handshakeCorruptError()))
+        }
+      case ElasticFailure(error) if indexNotFound(error) && retriesLeft > 0 =>
+        // Out-of-band index delete: reset the memo, re-ensure, retry ONCE.
+        ready = false
+        ensureHandshakeIndex() match {
+          case ElasticFailure(e) => Future.successful(ElasticFailure(e))
+          case _                 => runSearch(statement, retriesLeft - 1)
+        }
+      case ElasticFailure(error) =>
+        // The connection check doing its job: propagate the client/cluster failure verbatim.
+        Future.successful(ElasticFailure(error))
+    }
+  }
+
+  /** Probe-before-act (project_mv_metadata_index_contract): an ElasticFailure from the probe is
+    * NEVER read as "absent". A failed create re-probes ONCE (lost cross-process race => proceed)
+    * and otherwise propagates the ORIGINAL failure (a 403 stays a 403 — PD-6/OQ-6). Memoized per
+    * client; concurrent ensures are idempotent (PUT same doc id). Runs synchronously on the caller
+    * thread — the established extension-path shape, once per client lifecycle.
+    */
+  private[client] def ensureHandshakeIndex(forceSeed: Boolean = false): ElasticResult[Unit] = {
+    if (ready && !forceSeed) ElasticResult.success(())
+    else {
+      api.indexExists(HandshakeIndex, pattern = false) match {
+        case ElasticFailure(error) if error.statusCode.contains(403) =>
+          // A read-only account can be denied the EXISTS probe itself (ES security answers the
+          // exists action with 403 for an index the user has no privilege on) — the OQ-6/PD-6
+          // guidance must reach THIS failure too, not only the create/seed 403s, or exactly the
+          // read-only BI session the guidance exists for gets a bare security_exception.
+          withGuidance(ElasticFailure(error))
+        case ElasticFailure(error) =>
+          ElasticFailure(error) // outage != absence — propagate verbatim (never "absent")
+        case ElasticSuccess(true) if !forceSeed =>
+          ready = true
+          ElasticResult.success(())
+        case ElasticSuccess(existsNow) =>
+          val created: ElasticResult[_] =
+            if (existsNow) ElasticResult.success(true)
+            else
+              // mappings MUST be passed: without it the seed doc dynamic-maps `dummy` as
+              // text+keyword instead of the lead-mandated single keyword field (AC 5/AD-9),
+              // and the HandshakeMapping constant is dead code.
+              api.createIndex(
+                HandshakeIndex,
+                settings = handshakeSettings(),
+                mappings = Some(HandshakeMapping)
+              ) match {
+                case f @ ElasticFailure(_) =>
+                  api.indexExists(HandshakeIndex, pattern = false) match {
+                    case ElasticSuccess(true) => ElasticResult.success(true) // lost the race
+                    case _                    => withGuidance(f)
+                  }
+                case ok => ok
+              }
+          created match {
+            case ElasticFailure(error) => ElasticFailure(error)
+            case _ =>
+              api.index(HandshakeIndex, HandshakeDocId, HandshakeDoc) match {
+                case ElasticFailure(error) => withGuidance(ElasticFailure(error))
+                case _ =>
+                  api.refresh(HandshakeIndex) match {
+                    case ElasticFailure(error) => ElasticFailure(error)
+                    case _ =>
+                      ready = true
+                      logger.info(s"✅ Handshake index '$HandshakeIndex' ready")
+                      ElasticResult.success(())
+                  }
+              }
+          }
+      }
+    }
+  }
+
+  /** index.hidden exists only from ES 7.7 (AD-9); `api.version` caches successes. A version lookup
+    * failure — or an unparseable version string — falls back to the un-hidden settings (the create
+    * itself will surface any real outage) — defense-in-depth must not add a failure mode.
+    */
+  private def handshakeSettings(): String =
+    api.version match {
+      case ElasticSuccess(v) if Try(ElasticsearchVersion.isAtLeast(v, 7, 7)).getOrElse(false) =>
+        HandshakeSettingsHidden
+      case _ => HandshakeSettings
+    }
+
+  /** Bounded self-heal trigger. statusCode None != 404 (project_elastic_error_status_semantics) —
+    * the message probe covers status-less transports; a wrong trigger costs one retry, never a
+    * wrong answer.
+    */
+  private def indexNotFound(error: ElasticError): Boolean =
+    error.statusCode.contains(404) ||
+    Option(error.message).exists(_.contains("index_not_found"))
+
+  private def handshakeCorruptError(): ElasticError =
+    ElasticError(
+      message =
+        s"FROM-less SELECT handshake found index '$HandshakeIndex' present but empty and could " +
+        s"not re-seed it. Seed it once: PUT /$HandshakeIndex/_doc/$HandshakeDocId $HandshakeDoc",
+      statusCode = Some(500),
+      index = Some(HandshakeIndex),
+      operation = Some("handshake")
+    )
+
+  /** PD-6/OQ-6 recommendation (lead-confirmed default): keep the original failure — status
+    * included, never invented — and append the pre-creation guidance for read-only BI service
+    * accounts. Both routes are named (lead review of PR #268): the SQL one for an administrator
+    * connected through SoftClient4ES itself, the REST one for curl/Kibana.
+    */
+  private def withGuidance(f: ElasticFailure): ElasticFailure =
+    ElasticFailure(
+      f.elasticError.copy(
+        message = s"${f.elasticError.message} — FROM-less SELECT executes a Painless handshake " +
+          s"against index '$HandshakeIndex'. If this client must stay read-only, create it " +
+          s"once as an administrator — via SQL: CREATE TABLE IF NOT EXISTS $HandshakeIndex " +
+          s"""(dummy KEYWORD) OPTIONS (settings = (number_of_shards = "1", """ +
+          s"""number_of_replicas = "0")); INSERT INTO $HandshakeIndex (dummy) VALUES ('dummy'); """ +
+          s"or via REST: PUT /$HandshakeIndex " +
+          s"""{"settings": {"number_of_shards": 1, "number_of_replicas": 0}, """ +
+          s""""mappings": $HandshakeMapping} then PUT /$HandshakeIndex/_doc/$HandshakeDocId """ +
+          s"$HandshakeDoc — the driver then uses it read-only.",
+        index = Some(HandshakeIndex),
+        operation = Some("handshake")
+      )
+    )
+
+  /** Project EXACTLY the select-list outputs (PD-2 names), unwrapping the ES per-field
+    * script_fields array (AD-11): the generic parseSimpleHits row keeps the wrap AND appends the
+    * `dummy` _source entry (normalizeRow "extra fields"). Response keys = the rewrite's computed
+    * aliases (`__cN`) or explicit aliases — positionally zipped with the statement's output names
+    * (same Select instance, same order; key alignment is by construction: SingleSearch.scriptFields
+    * = fieldsWithComputedAliases.filter(_.isScriptField)).
+    */
+  private def assembleRow(
+    statement: FromlessSelect,
+    single: SingleSearch,
+    raw: ListMap[String, Any]
+  ): ListMap[String, Any] = {
+    val responseKeys =
+      single.select.fieldsWithComputedAliases.map(f =>
+        f.fieldAlias.map(_.alias).getOrElse(f.sourceField)
+      )
+    ListMap(statement.columnNames.zip(responseKeys).map { case (out, key) =>
+      out -> (raw.get(key) match {
+        case Some(l: Seq[_]) if l.size == 1 => l.head // the ES per-field array wrapper
+        case Some(l: Seq[_]) if l.isEmpty   => null
+        case Some(v)                        => v // defensive passthrough — never guess
+        case None                           => null // script returned null -> key absent
+      })
+    }: _*)
+  }
+}