Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand All @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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*/
}
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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 {
Expand All @@ -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*/
}
Original file line number Diff line number Diff line change
@@ -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
}
}
Original file line number Diff line number Diff line change
@@ -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
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand All @@ -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*/
)
)
})
Expand Down
Loading
Loading