fix(bridge,client,core): STDDEV/VARIANCE over a transformed expression — per-major emission (#222)

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

@@ -89,11 +89,23 @@ case class ElasticAggregation(
 
   // CHECK if it is a "global" metric (cardinality, etc.) or a bucket metric (avg, sum, etc.)
   val isGlobalMetric: Boolean = agg match {
-    case _: CardinalityAggregation   => true
-    case _: StatsAggregation         => true
-    case _: ExtendedStatsAggregation => true
-    case _                           => false
+    case _: CardinalityAggregation           => true
+    case _: StatsAggregation                 => true
+    case _: ExtendedStatsAggregation         => true
+    case _: ScriptedExtendedStatsAggregation => true
+    case _                                   => false
   }
+
+  /** True when this aggregation is `STDDEV` / `VARIANCE` (any of the `extended_stats` family) over
+    * a TRANSFORMED expression -- the shape a client module must either render with its script or
+    * refuse (issue #222).
+    *
+    * Nothing in the emission path consults this: on both doors the decision is taken by the
+    * injected [[SearchBodySerializer]], off the `SearchRequest` it is handed. This is the
+    * per-aggregation view of the same predicate, for a caller holding an [[ElasticAggregation]] --
+    * the client-module tests assert it on the `sqlQueryToAggregations` door.
+    */
+  def hasTransformExtendedStats: Boolean = ScriptedExtendedStatsAggregation.existsIn(Seq(agg))
 }
 
 object ElasticAggregation {
@@ -196,7 +208,11 @@ object ElasticAggregation {
         case STDDEV | STDDEV_SAMP | STDDEV_POP | VARIANCE | VAR_SAMP | VAR_POP =>
           aggWithFieldOrScript(
             extendedStatsAgg,
-            (name, s) => extendedStatsAgg(name, sourceField).script(s)
+            // Issue #222 -- a transform-bearing extended_stats is bound to the bridge's own marker:
+            // elastic4s's ExtendedStatsAggregationBuilder drops `script`, so the library type
+            // would serialise as the statistic of the raw field. See ScriptedExtendedStatsAggregation.
+            (name, s) =>
+              ScriptedExtendedStatsAggregation(extendedStatsAgg(name, sourceField).script(s))
           )
         case th: WindowFunction =>
           th.window match {
@@ -222,7 +238,11 @@ object ElasticAggregation {
             case STDDEV | STDDEV_SAMP | STDDEV_POP | VARIANCE | VAR_SAMP | VAR_POP =>
               aggWithFieldOrScript(
                 extendedStatsAgg,
-                (name, s) => extendedStatsAgg(name, sourceField).script(s)
+                // Issue #222 -- a transform-bearing extended_stats is bound to the bridge's own marker:
+                // elastic4s's ExtendedStatsAggregationBuilder drops `script`, so the library type
+                // would serialise as the statistic of the raw field. See ScriptedExtendedStatsAggregation.
+                (name, s) =>
+                  ScriptedExtendedStatsAggregation(extendedStatsAgg(name, sourceField).script(s))
               )
             case PERCENTILE_CONT | PERCENTILE_DISC =>
               // Both map to ES `percentiles` (TDigest). One call → one percent;

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

@@ -22,5 +22,11 @@ case class ElasticMultiSearchRequest(
   requests: Seq[ElasticSearchRequest],
   multiSearch: MultiSearchRequest
 ) {
+  // Not routed through SearchBodySerializer (issue #222): no production path serialises a
+  // multi-search here -- core builds `_msearch` bodies from each request's own
+  // `singleSearchToJsonQuery` (`ElasticQueries.multiQuery`). A transform-bearing extended_stats
+  // inside this body still fails LOUDLY on every elastic4s line (its default aggregation handler
+  // throws `NotImplementedError` on the ScriptedExtendedStatsAggregation marker); it can never
+  // leave as the statistic of the wrong field.
   def query: String = MultiSearchBuilderFn(multiSearch).replace("\"version\":true,", "") /*FIXME*/
 }

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

@@ -17,7 +17,7 @@
 package app.softnetwork.elastic.sql.bridge
 
 import app.softnetwork.elastic.sql.query.{Bucket, Criteria, Except, Field, FieldSort}
-import com.sksamuel.elastic4s.requests.searches.{SearchBodyBuilderFn, SearchRequest}
+import com.sksamuel.elastic4s.requests.searches.SearchRequest
 
 case class ElasticSearchRequest(
   sql: String,
@@ -30,7 +30,10 @@ case class ElasticSearchRequest(
   search: SearchRequest,
   buckets: Seq[Bucket] = Seq.empty,
   having: Option[Criteria] = None,
-  sorts: Seq[FieldSort] = Seq.empty
+  sorts: Seq[FieldSort] = Seq.empty,
+  // The body serializer the client module injected through the SingleSearch conversion (issue
+  // #222); Default = the one-argument elastic4s builder, refusing a transform-bearing extended_stats.
+  serializer: SearchBodySerializer = SearchBodySerializer.Default
 ) {
   def minScore(score: Option[Double]): ElasticSearchRequest = {
     score match {
@@ -39,6 +42,12 @@ case class ElasticSearchRequest(
     }
   }
 
+  /** True when this request carries `STDDEV` / `VARIANCE` (any of the `extended_stats` family) over
+    * a TRANSFORMED expression -- plain (`STDDEV(YEAR(x))`) or windowed (`... OVER (PARTITION BY
+    * ...)`). The shape a client module must either render with its script or refuse (issue #222).
+    */
+  def hasTransformExtendedStats: Boolean = SearchBodySerializer.hasTransformExtendedStats(search)
+
   def query: String =
-    SearchBodyBuilderFn(search).string.replace("\"version\":true,", "") /*FIXME*/
+    serializer.serialize(search).replace("\"version\":true,", "") /*FIXME*/
 }

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

@@ -0,0 +1,83 @@
+/*
+ * 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.sql.bridge
+
+import com.sksamuel.elastic4s.requests.searches.aggs.{
+  AbstractAggregation,
+  Aggregation,
+  ExtendedStatsAggregation
+}
+
+/** An `extended_stats` aggregation over a TRANSFORMED expression -- `STDDEV(YEAR(createdAt))`,
+  * `VARIANCE(ABS(salary))` and the rest of the family (issue #222).
+  *
+  * elastic4s's own `ExtendedStatsAggregationBuilder` never emits `agg.script` (the only metric
+  * builder of ten with the omission; fixed upstream once in elastic4s#2700 onto a dead branch,
+  * re-submitted as elastic4s#4100), so an `ExtendedStatsAggregation` carrying a script silently
+  * serialises as the statistic of the RAW field -- or as `extended_stats: {}` when there is no raw
+  * field to fall back on. The bridge therefore binds a transform-bearing extended_stats to this
+  * marker instead of to the library type it wraps. A marker is a type elastic4s does not know, so:
+  *
+  *   - `AggregationBuilderFn`'s typed arms never claim it, and the `customAggregation` handler of
+  *     the 8.x/9.x two-argument `SearchBodyBuilderFn.apply` IS consulted for it (the typed `case
+  *     agg: ExtendedStatsAggregation` arm runs BEFORE that handler, which is why the handler cannot
+  *     key on the library type). The ES 8 / ES 9 client modules render it with its script.
+  *   - the one-argument builders (elastic4s 6.x / 7.x, and the 8.x/9.x default handler) throw a
+  *     `NotImplementedError` on it -- the request can never leave as silently-wrong JSON.
+  *     [[SearchBodySerializer.Default]] refuses it earlier, with a named message; the ES 6 / ES 7
+  *     client modules refuse it with an `ElasticError` naming their major.
+  *
+  * `inner` is exactly the aggregation the default builder would have received (name, field, script,
+  * sigma, missing, sub-aggregations, metadata), so a rendering handler stays in parity with it.
+  */
+final case class ScriptedExtendedStatsAggregation(inner: ExtendedStatsAggregation)
+    extends Aggregation {
+
+  require(
+    inner.script.isDefined,
+    "ScriptedExtendedStatsAggregation wraps an extended_stats that carries a script"
+  )
+
+  type T = ScriptedExtendedStatsAggregation
+
+  override def name: String = inner.name
+
+  override def metadata: Map[String, AnyRef] = inner.metadata
+
+  override def subaggs: Seq[AbstractAggregation] = inner.subaggs
+
+  override def subAggregations(aggs: Iterable[AbstractAggregation]): T =
+    copy(inner = inner.subAggregations(aggs))
+
+  override def metadata(map: Map[String, AnyRef]): T = copy(inner = inner.metadata(map))
+}
+
+object ScriptedExtendedStatsAggregation {
+
+  /** True when `aggs`, or any aggregation nested below them, is a
+    * [[ScriptedExtendedStatsAggregation]] -- the shape discriminator behind
+    * `hasTransformExtendedStats` (issue #222). Both binds are covered by construction: the plain
+    * `STDDEV(f(x))` metric and the windowed `STDDEV(f(x)) OVER (PARTITION BY ...)` metric are the
+    * same marker, one at the root and one under a partition bucket.
+    */
+  def existsIn(aggs: Iterable[AbstractAggregation]): Boolean =
+    aggs.exists {
+      case _: ScriptedExtendedStatsAggregation => true
+      case a: Aggregation                      => existsIn(a.subaggs)
+      case _                                   => false
+    }
+}

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

@@ -0,0 +1,74 @@
+/*
+ * 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.sql.bridge
+
+import com.sksamuel.elastic4s.requests.searches.{SearchBodyBuilderFn, SearchRequest}
+
+/** The ONE place a bridge `SearchRequest` becomes a JSON body (issue #222).
+  *
+  * The bridge is a shared template: `copyBridge` copies these sources byte-for-byte into the ES 7,
+  * ES 8 and ES 9 modules, which compile them against elastic4s 7.17.x (one-argument
+  * `SearchBodyBuilderFn`) and 8.x / 9.x (two-argument, with a `customAggregation` handler).
+  * Anything a single major can do therefore lives in that major's CLIENT module and is injected
+  * here: the client module puts an implicit `SearchBodySerializer` in scope of the `SingleSearch`
+  * conversions (`requestToElasticSearchRequest`, `sqlQueryToAggregations`), and every serialisation
+  * door consumes it. Without one, [[SearchBodySerializer.Default]] applies.
+  */
+trait SearchBodySerializer {
+
+  /** The JSON body of `search`, exactly as the transport sends it. */
+  def serialize(search: SearchRequest): String
+}
+
+object SearchBodySerializer {
+
+  /** True when the request carries an `extended_stats` over a transformed expression -- any
+    * [[ScriptedExtendedStatsAggregation]] anywhere in its aggregation tree (plain or windowed
+    * bind).
+    */
+  def hasTransformExtendedStats(search: SearchRequest): Boolean =
+    ScriptedExtendedStatsAggregation.existsIn(search.aggs)
+
+  /** The version-agnostic refusal the Default serializer raises (issue #222). */
+  val TransformExtendedStatsUnsupported: String =
+    "STDDEV/VARIANCE over a transformed expression cannot be serialised by the default " +
+    "Elasticsearch body builder: the underlying elastic4s builder drops the aggregation script " +
+    "(elastic4s#4100), so the statistic would silently be computed over the raw field. " +
+    "Aggregate over a raw field, or use Elasticsearch 8+."
+
+  /** The refusal an ES-major-aware client module raises for the same shape, naming its major. */
+  def transformExtendedStatsUnsupportedOn(major: Int): String =
+    s"STDDEV/VARIANCE over a transformed expression is not supported on Elasticsearch $major: " +
+    "the underlying elastic4s builder drops the aggregation script (elastic4s#4100), so the " +
+    "statistic would silently be computed over the raw field. Aggregate over a raw field, or use " +
+    "Elasticsearch 8+."
+
+  /** Today's behaviour -- the one-argument `SearchBodyBuilderFn` every elastic4s line offers --
+    * guarded against the one shape it cannot render honestly. A transform-bearing extended_stats
+    * has no script-emitting builder on this path, so the request is REFUSED, loudly and by name,
+    * instead of leaving as the statistic of the wrong field. (Left to elastic4s, the marker would
+    * still fail -- `AggregationBuilderFn`'s `case ni => throw new NotImplementedError(...)` -- but
+    * with a message that says nothing about the statistic.)
+    */
+  object Default extends SearchBodySerializer {
+    override def serialize(search: SearchRequest): String = {
+      if (hasTransformExtendedStats(search))
+        throw new UnsupportedOperationException(TransformExtendedStatsUnsupported)
+      SearchBodyBuilderFn(search).string
+    }
+  }
+}

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

@@ -438,9 +438,14 @@ package object bridge {
     }
   }
 
+  // `serializer` is the seam of issue #222: an ES-major-aware client module puts its own
+  // SearchBodySerializer in implicit scope of this conversion; with none in scope the default
+  // argument applies (the one-argument elastic4s builder, refusing a transform-bearing
+  // extended_stats). Same defaulted-implicit pattern as `contextType`.
   implicit def requestToElasticSearchRequest(request: SingleSearch)(implicit
     timestamp: Long,
-    contextType: PainlessContextType = PainlessContextType.Query
+    contextType: PainlessContextType = PainlessContextType.Query,
+    serializer: SearchBodySerializer = SearchBodySerializer.Default
   ): ElasticSearchRequest =
     ElasticSearchRequest(
       request.sql,
@@ -453,7 +458,8 @@ package object bridge {
       request,
       request.buckets,
       request.having.flatMap(_.criteria),
-      request.orderBy.map(_.sorts).getOrElse(Seq.empty)
+      request.orderBy.map(_.sorts).getOrElse(Seq.empty),
+      serializer = serializer
     ).minScore(request.score)
 
   /** Merge percentile ElasticAggregations that share a value column / `cont` flag / partition into
@@ -1087,6 +1093,8 @@ package object bridge {
   implicit def queryToJson(
     query: Query
   ): JsonNode = {
+    // Query-only body (no aggregations): audited exempt from the SearchBodySerializer seam
+    // (issue #222) -- there is no extended_stats here for a serializer to render or refuse.
     JacksonBuilder.toNode(
       SearchBodyBuilderFn(
         ElasticApi.search("") query {
@@ -1124,11 +1132,14 @@ package object bridge {
     ElasticBridge(filter)
   }
 
+  // The second serialisation door (issue #222): each aggregation's own single-aggregation body is
+  // rendered through the same injected SearchBodySerializer as ElasticSearchRequest.query.
   implicit def sqlQueryToAggregations(
     query: SelectStatement
   )(implicit
     timestamp: Long,
-    contextType: PainlessContextType = PainlessContextType.Query
+    contextType: PainlessContextType = PainlessContextType.Query,
+    serializer: SearchBodySerializer = SearchBodySerializer.Default
   ): Seq[ElasticAggregation] = {
     import query._
     statement
@@ -1143,35 +1154,32 @@ package object bridge {
                   .flatMap(_.criteria.map(ElasticCriteria(_).asQuery()))
                   .getOrElse(matchAllQuery())
 
+              val body: SearchRequest =
+                aggregation.aggType match {
+                  case COUNT if aggregation.sourceField.equalsIgnoreCase("_id") =>
+                    ElasticApi.search("") query {
+                      queryFiltered
+                    }
+                  case _ =>
+                    ElasticApi.search("") query {
+                      queryFiltered
+                    } aggregations {
+                      val filtered =
+                        filteredAgg match {
+                          case Some(filtered) => filtered.subAggregations(aggregation.agg)
+                          case _              => aggregation.agg
+                        }
+                      aggregation.nestedAgg match {
+                        case Some(nested) => nested.subAggregations(filtered)
+                        case _            => filtered
+                      }
+                    } size 0
+                }
+
               aggregation.copy(
                 sources = l.sources,
                 query = Some(
-                  (aggregation.aggType match {
-                    case COUNT if aggregation.sourceField.equalsIgnoreCase("_id") =>
-                      SearchBodyBuilderFn(
-                        ElasticApi.search("") query {
-                          queryFiltered
-                        }
-                      )
-                    case _ =>
-                      SearchBodyBuilderFn(
-                        ElasticApi.search("") query {
-                          queryFiltered
-                        }
-                        aggregations {
-                          val filtered =
-                            filteredAgg match {
-                              case Some(filtered) => filtered.subAggregations(aggregation.agg)
-                              case _              => aggregation.agg
-                            }
-                          aggregation.nestedAgg match {
-                            case Some(nested) => nested.subAggregations(filtered)
-                            case _            => filtered
-                          }
-                        }
-                        size 0
-                      )
-                  }).string.replace("\"version\":true,", "") /*FIXME*/
+                  serializer.serialize(body).replace("\"version\":true,", "") /*FIXME*/
                 )
               )
             })