fix(sql,core): resolve temporal literals against the mapped date format before rendering the query (#276)

bridge/src/test/scala/app/softnetwork/elastic/sql/TemporalLiteralQuerySpec.scala

@@ -0,0 +1,119 @@
+/*
+ * 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
+
+import app.softnetwork.elastic.sql.bridge._
+import app.softnetwork.elastic.sql.parser.Parser
+import app.softnetwork.elastic.sql.query.{SingleSearch, TemporalLiterals}
+import app.softnetwork.elastic.sql.schema.{Column, Table => SchemaTable}
+import app.softnetwork.elastic.sql.`type`.SQLTypes
+import com.sksamuel.elastic4s.requests.searches.SearchBodyBuilderFn
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.time.ZonedDateTime
+import scala.collection.immutable.ListMap
+
+/** Issue #276 -- the GENERATED Elasticsearch query once temporal literals have been resolved
+  * against the mapped `date` columns. A parse-level assertion cannot see what reaches `rangeQuery`
+  * / `termQuery`; this spec does. The keyword negative control (AC 3) is a byte-identical
+  * comparison of the emitted JSON with and without the schema.
+  */
+class TemporalLiteralQuerySpec extends AnyFlatSpec with Matchers {
+
+  implicit def timestamp: Long = ZonedDateTime.parse("2025-12-31T00:00:00Z").toInstant.toEpochMilli
+
+  private val schema: SchemaTable = SchemaTable(
+    "events",
+    columns = List(
+      Column("id", SQLTypes.Keyword),
+      Column("event_ts", SQLTypes.Date),
+      Column("label", SQLTypes.Keyword),
+      Column(
+        "fmt_ts",
+        SQLTypes.Date,
+        options = ListMap("format" -> StringValue("yyyy-MM-dd HH:mm:ss"))
+      )
+    )
+  )
+
+  private def single(sql: String): SingleSearch = Parser(sql) match {
+    case Right(s: SingleSearch) => s
+    case other                  => fail(s"$sql did not parse as a SingleSearch: $other")
+  }
+
+  private def json(search: SingleSearch): String =
+    SearchBodyBuilderFn(requestToElasticSearchRequest(search).search).string
+
+  private def resolved(sql: String): SingleSearch =
+    TemporalLiterals(single(sql), schema) match {
+      case Right(s)     => s
+      case Left(reason) => fail(s"$sql was rejected: $reason")
+    }
+
+  "a range comparison against a date column" should "emit the T-separated literal" in {
+    json(resolved("SELECT id FROM events WHERE event_ts >= '2026-06-04 00:00:00.000000'")) shouldBe
+    """{"query":{"bool":{"filter":[{"range":{"event_ts":{"gte":"2026-06-04T00:00:00.000000"}}}]}},"_source":{"includes":["id"]}}"""
+  }
+
+  "equality, BETWEEN and IN against a date column" should "emit normalised term, range and terms values" in {
+    json(resolved("SELECT id FROM events WHERE event_ts = '2026-06-04 00:00:00'")) shouldBe
+    """{"query":{"bool":{"filter":[{"term":{"event_ts":{"value":"2026-06-04T00:00:00"}}}]}},"_source":{"includes":["id"]}}"""
+
+    json(
+      resolved(
+        "SELECT id FROM events WHERE event_ts BETWEEN '2026-06-01 00:00:00' AND '2026-06-30 23:59:59'"
+      )
+    ) shouldBe
+    """{"query":{"bool":{"filter":[{"range":{"event_ts":{"gte":"2026-06-01T00:00:00","lte":"2026-06-30T23:59:59"}}}]}},"_source":{"includes":["id"]}}"""
+
+    json(
+      resolved(
+        "SELECT id FROM events WHERE event_ts IN ('2026-06-04 00:00:00', '2026-07-01 00:00:00')"
+      )
+    ) shouldBe
+    """{"query":{"bool":{"filter":[{"terms":{"event_ts":["2026-06-04T00:00:00","2026-07-01T00:00:00"]}}]}},"_source":{"includes":["id"]}}"""
+  }
+
+  "a keyword column compared to a date-shaped string" should "emit byte-identical JSON with and without the schema" in {
+    val parsed = single("SELECT id FROM events WHERE label = '2026-06-04 00:00:00'")
+    val withSchema = TemporalLiterals(parsed, schema) match {
+      case Right(s)     => s
+      case Left(reason) => fail(reason)
+    }
+    withSchema should be theSameInstanceAs parsed
+    json(withSchema) shouldBe json(parsed)
+    json(parsed) shouldBe
+    """{"query":{"bool":{"filter":[{"term":{"label":{"value":"2026-06-04 00:00:00"}}}]}},"_source":{"includes":["id"]}}"""
+  }
+
+  "an epoch-millis literal against a date column" should "be forwarded untouched" in {
+    json(resolved("SELECT id FROM events WHERE event_ts >= '1780531200000'")) shouldBe
+    """{"query":{"bool":{"filter":[{"range":{"event_ts":{"gte":"1780531200000"}}}]}},"_source":{"includes":["id"]}}"""
+  }
+
+  "a custom-format date column" should "keep the space form its format already parses" in {
+    json(resolved("SELECT id FROM events WHERE fmt_ts >= '2026-06-04 00:00:00'")) shouldBe
+    """{"query":{"bool":{"filter":[{"range":{"fmt_ts":{"gte":"2026-06-04 00:00:00"}}}]}},"_source":{"includes":["id"]}}"""
+  }
+
+  "an already ISO literal" should "emit the same JSON as before" in {
+    val parsed =
+      single("SELECT id FROM events WHERE event_ts < '2026-06-04T00:00:00' AND label = 'x'")
+    json(resolved(parsed.sql)) shouldBe json(parsed)
+  }
+}

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

@@ -60,6 +60,12 @@ trait IndicesApi extends ElasticClientHelpers {
 
   private val schemaCache = new ConcurrentHashMap[String, (Schema, Long)]()
 
+  /** Alias -> the CONCRETE index its cached schema was resolved from (issue #276). Without it an
+    * `ALTER TABLE <index>` would leave a stale mapping -- and therefore a stale date `format` --
+    * reachable through the alias for the rest of the TTL (review R4-21).
+    */
+  private val schemaAliasTargets = new ConcurrentHashMap[String, String]()
+
   // ========================================================================
   // PUBLIC METHODS
   // ========================================================================
@@ -231,28 +237,51 @@ trait IndicesApi extends ElasticClientHelpers {
     }
   }
 
+  /** Drop every cached ALIAS schema resolved from `index` (#276 / review R4-21). */
+  private def invalidateAliasesOf(index: String): Unit =
+    schemaAliasTargets
+      .entrySet()
+      .asScala
+      .collect { case e if e.getValue == index => e.getKey }
+      .toList
+      .foreach { alias =>
+        schemaCache.remove(alias)
+        schemaAliasTargets.remove(alias)
+        logger.debug(s"📦 Schema cache invalidated for alias '$alias' (target '$index')")
+      }
+
   def updateSchema(index: String, schema: Schema): Unit = {
     schemaCache.put(index, (schema, System.currentTimeMillis()))
+    invalidateAliasesOf(index)
     // #238 — ALTER TABLE may have reindexed into a different shard count
     invalidateShardCounts(Some(index))
     logger.debug(s"📦 Schema cache updated for '$index'")
   }
 
   def invalidateSchema(index: String): Unit = {
     schemaCache.remove(index)
+    val _ = schemaAliasTargets.remove(index)
+    invalidateAliasesOf(index)
     invalidateShardCounts(Some(index)) // #238 — the sliced-paging shard counts follow the schema
     logger.info(s"🗑️ Schema cache invalidated for '$index'")
   }
 
   def invalidateAllSchemas(): Unit = {
     schemaCache.clear()
+    schemaAliasTargets.clear()
     invalidateShardCounts()
     logger.info("🗑️ All schema caches invalidated")
   }
 
   private def fetchSchemaFromES(index: String): ElasticResult[Schema] = {
     getIndex(index) match {
       case ElasticSuccess(Some(idx)) =>
+        // #276 -- remember which concrete index an ALIAS resolved to, so invalidating that index
+        // also drops the alias entry.
+        idx.resolvedFrom.filter(_ != index) match {
+          case Some(concrete) => val _ = schemaAliasTargets.put(index, concrete)
+          case None           => val _ = schemaAliasTargets.remove(index)
+        }
         ElasticSuccess(idx.schema)
       case ElasticSuccess(None) =>
         logger.warn(s"Index '$index' not found for schema loading")
@@ -357,7 +386,20 @@ trait IndicesApi extends ElasticClientHelpers {
   }
 
   private def loadIndexAsSchema(index: String, json: String): ElasticResult[Option[Index]] = {
-    var tempIndex = Index(index, json)
+    val root = mapper.readTree(json)
+    // Issue #276 -- `index` may be an ALIAS. `Index.apply` resolves an alias over exactly ONE
+    // index; an alias over several is ambiguous (which mapping?) and is reported as NOT FOUND
+    // rather than as an empty schema, on every client alike.
+    if (!root.has(index)) {
+      val members = Index.indexDocuments(root)
+      if (members.size > 1) {
+        logger.warn(
+          s"⚠️ '$index' is an alias over ${members.size} indices (${members.map(_._1).mkString(", ")}); its schema cannot be resolved"
+        )
+        return ElasticSuccess(None)
+      }
+    }
+    var tempIndex = Index(index, root)
     tempIndex.defaultIngestPipelineName match {
       case Some(pipeline) =>
         logger.info(
@@ -794,28 +836,51 @@ trait IndicesApi extends ElasticClientHelpers {
             )
           )
         else Right(())
-      jsonQuery = delete.where match {
+      jsonQuery <- (delete.where match {
         case None =>
           logger.info(
             s"SQL delete query has no WHERE clause, deleting all documents from index '$index'"
           )
-          """{"query": {"match_all": {}}}"""
+          Right("""{"query": {"match_all": {}}}""")
         case Some(where) =>
           implicit val timestamp: Long = System.currentTimeMillis()
-          val search: String =
+          // #276 -- the same temporal-literal resolution the equivalent SELECT gets
+          resolveDmlTemporalLiterals(
             SingleSearch(
               from = From(tables = Seq(delete.table)),
               where = Some(where),
               deleteByQuery = true
-            )
-          logger.info(s"✅ Converted SQL delete query to search for deleteByQuery: $search")
-          search
-      }
+            ),
+            "deleteByQuery"
+          ).map { resolved =>
+            val search: String = resolved
+            logger.info(s"✅ Converted SQL delete query to search for deleteByQuery: $search")
+            search
+          }
+      }): Either[ElasticError, String]
       deleted <- runDeleteByQuery(index, jsonQuery, refresh)
     } yield deleted
     finalizeDeleteByQuery(index, result)
   }
 
+  /** Issue #276 -- resolve the WHERE clause's temporal literals for a DELETE / UPDATE by-query
+    * search body exactly as [[SearchApi.resolveTemporalLiterals]] does for a SELECT (same rule,
+    * same schema-absent boundaries), so `DELETE FROM t WHERE ts >= '2026-06-04 00:00:00'` affects
+    * the rows the equivalent SELECT matches instead of failing with a raw shard error.
+    */
+  private def resolveDmlTemporalLiterals(
+    single: SingleSearch,
+    operation: String
+  ): Either[ElasticError, SingleSearch] =
+    this match {
+      case api: SearchApi =>
+        api.resolveTemporalLiterals(single) match {
+          case ElasticSuccess(resolved) => Right(resolved)
+          case ElasticFailure(error)    => Left(error.copy(operation = Some(operation)))
+        }
+      case _ => Right(single)
+    }
+
   private def validateDeleteIndex(index: String): Either[ElasticError, Unit] =
     validateIndexName(index)
       .toLeft(())
@@ -979,30 +1044,35 @@ trait IndicesApi extends ElasticClientHelpers {
       })
 
       // 4. Build the JSON query to execute
-      jsonQuery <- Right(parsed match {
+      jsonQuery <- (parsed match {
         case Left(u: Update) =>
           u.where match {
             case None =>
               logger.info(
                 s"SQL update query has no WHERE clause, updating all documents from index '$index'"
               )
-              """{"query": {"match_all": {}}}"""
+              Right("""{"query": {"match_all": {}}}""")
 
             case Some(where) =>
               implicit val timestamp: Long = System.currentTimeMillis()
-              val search: String =
+              // #276 -- the same temporal-literal resolution the equivalent SELECT gets
+              resolveDmlTemporalLiterals(
                 SingleSearch(
                   from = From(tables = Seq(Table(u.table))),
                   where = Some(where),
                   updateByQuery = true
-                )
-              logger.info(s"✅ Converted SQL update query to search for updateByQuery: $search")
-              search
+                ),
+                "updateByQuery"
+              ).map { resolved =>
+                val search: String = resolved
+                logger.info(s"✅ Converted SQL update query to search for updateByQuery: $search")
+                search
+              }
           }
 
         case Right(jsonOrConverted) =>
-          jsonOrConverted
-      })
+          Right(jsonOrConverted)
+      }): Either[ElasticError, String]
 
       // 5. Load user pipeline if provided
       userPipeline <- pipelineId match {
@@ -1529,9 +1599,16 @@ trait IndicesApi extends ElasticClientHelpers {
                 )
               else {
                 implicit val timestamp: Long = System.currentTimeMillis()
-                val query: String = search.copy(deleteByQuery = false)
-                logger.info(s"✅ Converted SQL search query to JSON for updateByQuery: $query")
-                ElasticSuccess(Right(query))
+                resolveDmlTemporalLiterals(
+                  search.copy(deleteByQuery = false),
+                  "updateByQuery"
+                ) match {
+                  case Right(resolved) =>
+                    val query: String = resolved
+                    logger.info(s"✅ Converted SQL search query to JSON for updateByQuery: $query")
+                    ElasticSuccess(Right(query))
+                  case Left(error) => ElasticFailure(error)
+                }
               }
 
             case _ =>
@@ -1649,14 +1726,20 @@ trait IndicesApi extends ElasticClientHelpers {
 
                 case Some(where) =>
                   implicit val timestamp: Long = System.currentTimeMillis()
-                  val search: String =
+                  resolveDmlTemporalLiterals(
                     SingleSearch(
                       from = From(tables = Seq(deleteStmt.table)),
                       where = Some(where),
                       deleteByQuery = true
+                    ),
+                    "deleteByQuery"
+                  ).map { resolved =>
+                    val search: String = resolved
+                    logger.info(
+                      s"✅ Converted SQL delete query to search for deleteByQuery: $search"
                     )
-                  logger.info(s"✅ Converted SQL delete query to search for deleteByQuery: $search")
-                  Right(search)
+                    search
+                  }
               }
 
           case search: SingleSearch =>
@@ -1672,9 +1755,12 @@ trait IndicesApi extends ElasticClientHelpers {
               )
             else {
               implicit val timestamp: Long = System.currentTimeMillis()
-              val query: String = search.copy(deleteByQuery = true)
-              logger.info(s"✅ Converted SQL search query to search for deleteByQuery: $query")
-              Right(query)
+              resolveDmlTemporalLiterals(search.copy(deleteByQuery = true), "deleteByQuery").map {
+                resolved =>
+                  val query: String = resolved
+                  logger.info(s"✅ Converted SQL search query to search for deleteByQuery: $query")
+                  query
+              }
             }
 
           case _ =>

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

@@ -274,7 +274,13 @@ trait ScrollApi extends ElasticClientHelpers {
         }
 
       // Single search
-      case single: SingleSearch =>
+      case parsed: SingleSearch =>
+        // #276 -- resolve temporal literals against the mapped `date` columns before ANY branch
+        // renders the query (the window-enrichment branch derives its queries from `single`).
+        val single = resolveTemporalLiterals(parsed) match {
+          case ElasticSuccess(resolved) => resolved
+          case ElasticFailure(error)    => return Source.failed(error)
+        }
         // #238 — an explicit LIMIT keeps the sequential PIT path on EVERY branch, including the
         // window-enrichment branch below (createBaseQuery keeps the LIMIT — AC 6).
         val config0 = if (single.limit.isDefined) config.copy(maxSlices = Some(1)) else config