fix(sql): date arithmetic follows the SQL engines; type checks wait for the column types

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

@@ -0,0 +1,92 @@
+/*
+ * 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.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+import com.fasterxml.jackson.databind.node.{ArrayNode, ObjectNode}
+import com.sksamuel.elastic4s.requests.searches.SearchRequest
+import com.sksamuel.elastic4s.requests.searches.aggs.{AbstractAggregation, Aggregation}
+import com.sksamuel.elastic4s.requests.searches.aggs.pipeline.BucketScriptPipelineAgg
+
+import scala.jdk.CollectionConverters._
+
+/** The PARAMS of a `bucket_script`'s script, which the elastic4s body builder drops.
+  *
+  * 🔴 elastic4s writes a `bucket_script` as `"script": "<source>"` -- the bare source string
+  * (`BucketScriptPipelineAggBuilder`, the same in elastic4s 6.7, 7.17, 8.18 and 9.0) -- while a
+  * `bucket_selector` gets the whole script object. So a per-group calculation reading the request
+  * clock, `CURRENT_DATE - MAX(d)` or `NOW() - MAX(ts)`, was sent WITHOUT the `__now__` its script
+  * reads, and Elasticsearch failed it; the same clock in a HAVING worked. MEASURED on 8.18.3.
+  *
+  * The script is sent as the object Elasticsearch documents -- `source`, `lang`, `params` -- for
+  * exactly the `bucket_script`s that carry params, so the source stays the SAME text from one
+  * request to the next and Elasticsearch compiles it once: inlining the clock instead would compile
+  * a new script on every request, against the cluster's compilation rate limit. Every other body is
+  * returned untouched, unparsed.
+  */
+object BucketScriptParams {
+
+  private val mapper = new ObjectMapper()
+
+  /** The params of every `bucket_script` of `aggs` that carries some, by aggregation name, at any
+    * depth.
+    */
+  private def withParams(aggs: Seq[AbstractAggregation]): Map[String, Map[String, Any]] =
+    aggs.flatMap {
+      case script: BucketScriptPipelineAgg if script.script.params.nonEmpty =>
+        Seq(script.name -> script.script.params)
+      case parent: Aggregation => withParams(parent.subaggs).toSeq
+      case _                   => Nil
+    }.toMap
+
+  /** `body`, the JSON of `search`, with the params of its `bucket_script`s bound. */
+  def bind(body: String, search: SearchRequest): String = {
+    val params = withParams(search.aggs)
+    if (params.isEmpty) body
+    else {
+      val root = mapper.readTree(body)
+      bind(root, params)
+      mapper.writeValueAsString(root)
+    }
+  }
+
+  private def bind(node: JsonNode, params: Map[String, Map[String, Any]]): Unit =
+    node match {
+      case obj: ObjectNode =>
+        obj.fields().asScala.toList.foreach { entry =>
+          (params.get(entry.getKey), entry.getValue) match {
+            case (Some(values), agg: ObjectNode) if agg.has("bucket_script") =>
+              agg.get("bucket_script") match {
+                case bucketScript: ObjectNode if bucketScript.path("script").isTextual =>
+                  val script = mapper.createObjectNode()
+                  script.put("source", bucketScript.get("script").asText())
+                  script.put("lang", "painless")
+                  val bound = script.putObject("params")
+                  values.foreach { case (name, value) =>
+                    bound.set[JsonNode](name, mapper.valueToTree[JsonNode](value))
+                  }
+                  bucketScript.set[JsonNode]("script", script)
+                case _ =>
+              }
+            case _ =>
+          }
+          bind(entry.getValue, params)
+        }
+      case array: ArrayNode => array.elements().asScala.foreach(bind(_, params))
+      case _                =>
+    }
+}

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

@@ -16,7 +16,7 @@
 
 package app.softnetwork.elastic.sql.bridge
 
-import app.softnetwork.elastic.sql.{PainlessContext, PainlessContextType}
+import app.softnetwork.elastic.sql.{PainlessContext, PainlessContextType, PainlessTarget}
 import app.softnetwork.elastic.sql.`type`.SQLTemporal
 import app.softnetwork.elastic.sql.query.{
   Asc,
@@ -162,7 +162,8 @@ object ElasticAggregation {
     allAggregations: Map[String, SQLAggregation]
   )(implicit
     timestamp: Long,
-    contextType: PainlessContextType
+    contextType: PainlessContextType,
+    target: PainlessTarget
   ): ElasticAggregation = {
     import sqlAgg._
     val sourceField = identifier.path
@@ -218,14 +219,14 @@ object ElasticAggregation {
       buildScript: (String, Script) => Aggregation
     ): Aggregation = {
       if (transformFuncs.nonEmpty) {
-        val context = PainlessContext(context = contextType)
+        val context = PainlessContext(context = contextType, target = target)
         val scriptSrc = identifier.painless(Some(context))
         val script = now(Script(s"$context$scriptSrc").lang("painless"))
         buildScript(aggName, script)
       } else {
         aggType match {
           case th: WindowFunction if th.shouldBeScripted =>
-            val context = PainlessContext(context = contextType)
+            val context = PainlessContext(context = contextType, target = target)
             val scriptSrc = th.identifier.painless(Some(context))
             val script = now(Script(s"$context$scriptSrc").lang("painless"))
             buildScript(aggName, script)
@@ -475,7 +476,8 @@ object ElasticAggregation {
     allElasticAggregations: Seq[ElasticAggregation]
   )(implicit
     timestamp: Long,
-    contextType: PainlessContextType = PainlessContextType.Query
+    contextType: PainlessContextType = PainlessContextType.Query,
+    target: PainlessTarget = PainlessTarget.Default
   ): Seq[Aggregation] = {
     for (tree <- buckets) yield {
       val treeNodes =
@@ -492,7 +494,7 @@ object ElasticAggregation {
 
           val aggScript =
             if (!bucket.isBucketScript && bucket.shouldBeScripted) {
-              val context = PainlessContext(context = contextType)
+              val context = PainlessContext(context = contextType, target = target)
               val painless = bucket.painless(Some(context))
               Some(now(Script(s"$context$painless").lang("painless")))
             } else {
@@ -720,7 +722,8 @@ object ElasticAggregation {
     references: Seq[ElasticAggregation]
   )(implicit
     timestamp: Long,
-    contextType: PainlessContextType
+    contextType: PainlessContextType,
+    target: PainlessTarget
   ): Option[Aggregation] =
     request.having.flatMap(_.criteria) match {
       case Some(criteria) if request.wholeTableHaving && aggs.nonEmpty =>

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

@@ -16,7 +16,7 @@
 
 package app.softnetwork.elastic.sql.bridge
 
-import app.softnetwork.elastic.sql.PainlessContextType
+import app.softnetwork.elastic.sql.{PainlessContextType, PainlessTarget}
 import app.softnetwork.elastic.sql.operator.AND
 import app.softnetwork.elastic.sql.query.{
   BetweenExpr,
@@ -48,7 +48,8 @@ case class ElasticBridge(filter: ElasticFilter) {
     currentQuery: Option[ElasticBoolQuery]
   )(implicit
     timestamp: Long,
-    contextType: PainlessContextType = PainlessContextType.Query
+    contextType: PainlessContextType = PainlessContextType.Query,
+    target: PainlessTarget = PainlessTarget.Default
   ): Query = {
     filter match {
       case boolQuery: ElasticBoolQuery =>

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

@@ -16,15 +16,16 @@
 
 package app.softnetwork.elastic.sql.bridge
 
-import app.softnetwork.elastic.sql.PainlessContextType
+import app.softnetwork.elastic.sql.{PainlessContextType, PainlessTarget}
 import app.softnetwork.elastic.sql.query.Criteria
 import com.sksamuel.elastic4s.requests.searches.queries.Query
 
 case class ElasticCriteria(criteria: Criteria) {
 
   def asQuery(group: Boolean = true, innerHitsNames: Set[String] = Set.empty)(implicit
     timestamp: Long,
-    contextType: PainlessContextType = PainlessContextType.Query
+    contextType: PainlessContextType = PainlessContextType.Query,
+    target: PainlessTarget = PainlessTarget.Default
   ): Query = {
     val query = criteria.boolQuery.copy(group = group)
     query

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

@@ -49,5 +49,10 @@ case class ElasticSearchRequest(
   def hasTransformExtendedStats: Boolean = SearchBodySerializer.hasTransformExtendedStats(search)
 
   def query: String =
-    serializer.serialize(search).replace("\"version\":true,", "") /*FIXME*/
+    // the params of a per-group calculation's script, which the body builder drops: the request
+    // clock `CURRENT_DATE - MAX(d)` reads (`BucketScriptParams`)
+    BucketScriptParams.bind(
+      serializer.serialize(search).replace("\"version\":true,", ""), /*FIXME*/
+      search
+    )
 }