feat(21.8): an index carries its own schema-cache TTL (Part D)

core/src/main/resources/help/commands/ddl/alter_table.json

@@ -37,7 +37,9 @@
     "SET SETTING setting = value",
     "DROP SETTING setting",
     "SET ALIAS alias = value",
-    "DROP ALIAS alias"
+    "DROP ALIAS alias",
+    "SET SCHEMA CACHE TTL [=] 'duration'",
+    "DROP SCHEMA CACHE TTL"
   ],
   "description": "Modify an existing Elasticsearch index by adding, removing, or modifying columns. Some operations require index reindexing (like changing data types).",
   "clauses": [
@@ -150,6 +152,11 @@
       "name": "SET ALIAS / DROP ALIAS",
       "description": "Add, update, or remove index aliases",
       "optional": true
+    },
+    {
+      "name": "SET SCHEMA CACHE TTL / DROP SCHEMA CACHE TTL",
+      "description": "Set or remove how long clients may cache this table's schema before reading it from Elasticsearch again. Overrides the client's elastic.schema-cache.ttl for this table only; stored in the index metadata, so every client sees it",
+      "optional": true
     }
   ],
   "examples": [
@@ -192,13 +199,20 @@
       "title": "Multiple alterations",
       "description": "Apply several changes in one statement",
       "sql": "ALTER TABLE users (\n  ADD COLUMN IF NOT EXISTS age INT DEFAULT 0,\n  RENAME COLUMN name TO full_name,\n  ALTER COLUMN status SET DEFAULT 'active',\n  ALTER COLUMN profile SET FIELDS (\n    bio VARCHAR DEFAULT 'N/A',\n    verified BOOLEAN DEFAULT false\n  )\n)"
+    },
+    {
+      "title": "Cache this table's schema for an hour",
+      "description": "A mapping that never moves; DROP SCHEMA CACHE TTL returns the table to the client default",
+      "sql": "ALTER TABLE orders SET SCHEMA CACHE TTL = '1h'"
     }
   ],
   "notes": [
     "DROP COLUMN only removes the mapping, existing data is not deleted",
     "RENAME COLUMN and SET DATA TYPE operations trigger a reindex",
     "Adding columns with DEFAULT creates/updates the ingest pipeline",
-    "Use IF EXISTS/IF NOT EXISTS to make scripts idempotent"
+    "Use IF EXISTS/IF NOT EXISTS to make scripts idempotent",
+    "SET SCHEMA CACHE TTL writes _meta.schema_cache_ttl, so it is equivalent to SET MAPPING _meta.schema_cache_ttl = 'duration'; durations use the HOCON spellings (30s, 10m, 1h, or milliseconds) and anything else is rejected when the statement is parsed",
+    "A shortened schema cache TTL is noticed by another client only when ITS current entry expires, so the change takes effect within one OLD period"
   ],
   "limitations": [
     "Cannot change PRIMARY KEY after creation",

core/src/main/resources/softnetwork-elastic.conf

@@ -45,6 +45,16 @@ elastic {
     max-slices = ${?ELASTIC_SCROLL_MAX_SLICES}
   }
 
+  # How long a table's schema -- and the primary shard count that follows it -- may be cached by a
+  # client before Elasticsearch is read again. This is the DEFAULT: an index that declares its own
+  # (ALTER TABLE <t> SET SCHEMA CACHE TTL = '10m') overrides it for itself. Since issue #306 every
+  # executed query reads the cached schema, so a stale entry means wrong emitted Painless, not just
+  # a stale column list: shorten this for mappings that change under a running client.
+  schema-cache {
+    ttl = 5m
+    ttl = ${?ELASTIC_SCHEMA_CACHE_TTL}
+  }
+
   # When enabled, result rows surface the Elasticsearch document id as an `_id` column.
   # Disabled by default: SQL results carry only the selected columns.
   include-document-id = false

core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala

@@ -45,6 +45,10 @@ import java.time.Duration
   * @param scroll
   *   Paged row extraction settings (`elastic.scroll`: page size and the ceiling on concurrent PIT
   *   slices, #238)
+  * @param schemaCache
+  *   How long a table's schema (and the shard count that follows it) may be cached by this client
+  *   (`elastic.schema-cache.ttl`, story 21.8 Part D). An index may shorten or lengthen its own
+  *   through `ALTER TABLE … SET SCHEMA CACHE TTL`
   */
 case class ElasticConfig(
   credentials: ElasticCredentials = ElasticCredentials(),
@@ -55,7 +59,8 @@ case class ElasticConfig(
   metrics: MetricsConfig,
   watcher: ElasticCredentials,
   includeDocumentId: Boolean = false,
-  scroll: ScrollSettings = ScrollSettings()
+  scroll: ScrollSettings = ScrollSettings(),
+  schemaCache: SchemaCacheSettings = SchemaCacheSettings()
 )
 
 object ElasticConfig extends StrictLogging {

core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala

@@ -45,6 +45,10 @@ import java.time.Duration
   * @param scroll
   *   Paged row extraction settings (`elastic.scroll`: page size and the ceiling on concurrent PIT
   *   slices, #238)
+  * @param schemaCache
+  *   How long a table's schema (and the shard count that follows it) may be cached by this client
+  *   (`elastic.schema-cache.ttl`, story 21.8 Part D). An index may shorten or lengthen its own
+  *   through `ALTER TABLE … SET SCHEMA CACHE TTL`
   */
 case class ElasticConfig(
   credentials: ElasticCredentials = ElasticCredentials(),
@@ -55,7 +59,8 @@ case class ElasticConfig(
   metrics: MetricsConfig,
   watcher: ElasticCredentials,
   includeDocumentId: Boolean = false,
-  scroll: ScrollSettings = ScrollSettings()
+  scroll: ScrollSettings = ScrollSettings(),
+  schemaCache: SchemaCacheSettings = SchemaCacheSettings()
 )
 
 object ElasticConfig extends StrictLogging {

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

@@ -76,4 +76,11 @@ trait ElasticClientApi
     ScrollConfig(scrollSize = elasticConfig.scroll.size)
 
   override protected def configuredMaxSlices: Int = elasticConfig.scroll.maxSlices
+
+  /** The DEFAULT schema-cache TTL (`elastic.schema-cache.ttl`, `ELASTIC_SCHEMA_CACHE_TTL`, story
+    * 21.8 Part D). An index that declares its own through `ALTER TABLE … SET SCHEMA CACHE TTL`
+    * overrides it for itself; the shard-count cache and the 404 negative cache follow the same
+    * value.
+    */
+  override protected def schemaCacheTtlMs: Long = elasticConfig.schemaCache.ttlMs
 }

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

@@ -33,7 +33,13 @@ import app.softnetwork.elastic.sql.query.{
   Unknown,
   Update
 }
-import app.softnetwork.elastic.sql.schema.{GenericProcessor, IngestPipeline, Schema, TableAlias}
+import app.softnetwork.elastic.sql.schema.{
+  GenericProcessor,
+  IngestPipeline,
+  Schema,
+  SchemaCacheTtl,
+  TableAlias
+}
 import app.softnetwork.elastic.sql.serialization._
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.node.{ArrayNode, ObjectNode}
@@ -52,13 +58,52 @@ import scala.jdk.CollectionConverters._
   *   - Parameter validation
   *   - Automatic retry for transient errors
   */
-trait IndicesApi extends ElasticClientHelpers {
+trait IndicesApi extends ElasticClientHelpers with SchemaCacheTtlApi {
   _: RefreshApi with PipelineApi with BulkApi with ScrollApi with VersionApi with TemplateApi =>
 
-  // Schema cache TTL in milliseconds. Override in subclass to change (default: 5 minutes).
-  protected def schemaCacheTtlMs: Long = 5 * 60 * 1000L
+  private val schemaCache = new ConcurrentHashMap[String, CachedSchema]()
+
+  /** Above this many entries a cache miss also purges the expired ones. Keyed by index name, this
+    * map was unbounded while it was cold; since #306 every executed query writes to it, and a
+    * deployment that mints dated indices (`logs-2025.03`) would otherwise grow without bound. Same
+    * threshold and same precedent as #238's `shardCountCache` and #276's `schemaMisses`.
+    */
+  private val schemaCachePurgeThreshold = 256
+
+  /** The resolved TTL for `index`: what the index itself declares in its mapping metadata (`ALTER
+    * TABLE … SET SCHEMA CACHE TTL`), else the client default.
+    *
+    * Read off the cache ENTRY, which was stamped when the schema was fetched — answering costs no
+    * round trip, so `ScrollApi` may call it per extraction. An index whose schema is not cached
+    * (yet, or any more) resolves to the default; it will be re-resolved the moment its schema is.
+    *
+    * 🔴 Self-referential, and bounded because of it: change an index's TTL and another client
+    * notices only when ITS current entry expires, i.e. after at most one OLD period. Shortening a
+    * TTL therefore takes effect one old period later, never sooner.
+    */
+  override protected def schemaCacheTtlMsFor(index: String): Long =
+    schemaCache.get(index) match {
+      case null  => schemaCacheTtlMs
+      case entry => entry.ttlMs
+    }
 
-  private val schemaCache = new ConcurrentHashMap[String, (Schema, Long)]()
+  /** The TTL an index declares, or the default — and a WARN, once per fetch, for a declared value
+    * that is not a duration (the DDL sugar validates it, but `SET MAPPING` and a hand-written
+    * `_meta` do not).
+    */
+  private def resolveTtlMs(index: String, schema: Schema): Long =
+    SchemaCacheTtl.of(schema) match {
+      case Some(Right(ms))    => ms
+      case Some(Left(reason)) =>
+        // Called from inside `schemaCache.compute`, so this log holds a bin lock — deliberately
+        // acceptable here: the blocking `GET <index>` that produced this schema runs there too
+        // (see `loadSchema`), so the lock is already held across far more than a log line.
+        logger.warn(
+          s"⚠️ Index '$index' declares an invalid ${SchemaCacheTtl.MetadataPath}: $reason — using the default TTL"
+        )
+        schemaCacheTtlMs
+      case None => schemaCacheTtlMs
+    }
 
   /** 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` --
@@ -234,28 +279,75 @@ trait IndicesApi extends ElasticClientHelpers {
     // Only a SUCCESS is cached, exactly as before, so a failure is retried on the next statement
     // rather than being remembered here (the 404 negative cache lives in `SearchApi`).
     var result: ElasticResult[Schema] = null
+    var fetched = false
     schemaCache.compute(
       index,
       (_, existing) =>
         existing match {
-          case (schema, cachedAt) if now - cachedAt < schemaCacheTtlMs =>
+          case entry: CachedSchema if !entry.isExpired(now) =>
             logger.debug(s"📦 Schema cache hit for '$index'")
-            result = ElasticSuccess(schema)
-            existing
+            result = ElasticSuccess(entry.schema)
+            entry
           case _ =>
+            fetched = true
             fetchSchemaFromES(index) match {
               case success @ ElasticSuccess(schema) =>
                 result = success
-                (schema, now)
+                CachedSchema(schema, now, resolveTtlMs(index, schema))
               case failure =>
                 result = failure
                 existing // null when absent -> the entry stays absent
             }
         }
     )
+    if (fetched && schemaCache.size() > schemaCachePurgeThreshold) purgeExpiredSchemas(now)
     result
   }
 
+  /** Drop every expired entry — each on ITS own clock, through the single `isExpired` rule — plus
+    * the alias->target rows for exactly those entries.
+    *
+    * 🔴 The alias rows are keyed by what this sweep ACTUALLY removed, never by "no schema is cached
+    * under this alias". `fetchSchemaFromES` writes `schemaAliasTargets` from INSIDE the
+    * `schemaCache.compute` lambda, i.e. before `compute` installs the entry, so a concurrent sweep
+    * that asked `schemaCache.containsKey(alias)` would delete the row a fetch had just written —
+    * and an alias whose target is later altered would then keep serving a stale mapping (a stale
+    * date `format`, and a stale `baseType` for #306's emission) for a full TTL, silently. That is
+    * the defect the row exists to prevent (#276 / review R4-21).
+    *
+    * ⚠️ NOT covered by a test, and it cannot be from outside: the divergent state exists only
+    * between that `put` and `compute` installing the entry, and every seam a test can override
+    * (`executeGetIndex`, `getIndex`, `getTemplate`, `getPipeline`) runs BEFORE the `put`. Adding a
+    * hook that exists only to be parked in would be worse than the finding. The structural cure is
+    * to carry the target ON the cache entry so there is no second map to keep in sync — worth
+    * doing, deliberately not done here (it rewrites #276's invalidation path).
+    *
+    * Above four times the threshold the whole cache is dropped instead: values here are entire
+    * `Table`s, not the `Int` of #238's shard counts, and expiry alone bounds nothing when an index
+    * asks for a long TTL (`SET SCHEMA CACHE TTL = '30d'`) and the names keep changing — dated
+    * indices, per-tenant indices. The worst case of dropping is today's cost, one re-fetch.
+    */
+  private def purgeExpiredSchemas(now: Long): Unit = {
+    val expired = scala.collection.mutable.Set.empty[String]
+    schemaCache
+      .entrySet()
+      .removeIf((e: java.util.Map.Entry[String, CachedSchema]) =>
+        e.getValue.isExpired(now) && { expired += e.getKey; true }
+      )
+    schemaAliasTargets.keySet().removeIf((alias: String) => expired.contains(alias))
+    if (schemaCache.size() > schemaCachePurgeThreshold * 4) {
+      logger.info(
+        s"🗑️ Schema cache above ${schemaCachePurgeThreshold * 4} live entries; dropping it (the next statement per index re-reads its mapping)"
+      )
+      schemaCache.clear()
+      schemaAliasTargets.clear()
+    }
+    ()
+  }
+
+  /** Current size of the schema cache (tests). */
+  private[client] def schemaCacheSize: Int = schemaCache.size()
+
   /** Drop every cached ALIAS schema resolved from `index` (#276 / review R4-21). */
   private def invalidateAliasesOf(index: String): Unit =
     schemaAliasTargets
@@ -270,7 +362,8 @@ trait IndicesApi extends ElasticClientHelpers {
       }
 
   def updateSchema(index: String, schema: Schema): Unit = {
-    schemaCache.put(index, (schema, System.currentTimeMillis()))
+    val now = System.currentTimeMillis()
+    schemaCache.put(index, CachedSchema(schema, now, resolveTtlMs(index, schema)))
     invalidateAliasesOf(index)
     // #238 — ALTER TABLE may have reindexed into a different shard count
     invalidateShardCounts(Some(index))

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

@@ -0,0 +1,48 @@
+/*
+ * 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 java.time.Duration
+
+/** Schema-cache settings (`elastic.schema-cache` in HOCON, story 21.8 Part D).
+  *
+  * @param ttl
+  *   how long a table's schema may be cached by this client before it is read from Elasticsearch
+  *   again (`elastic.schema-cache.ttl`, `ELASTIC_SCHEMA_CACHE_TTL`). The DEFAULT: an index that
+  *   declares its own (`ALTER TABLE … SET SCHEMA CACHE TTL = '10m'`) overrides it for itself. Since
+  *   #306 every executed query reads the cached schema, so this is a correctness-latency knob:
+  *   shorten it for a mapping that changes under a running client, lengthen it to spare the
+  *   re-parse of a wide mapping. Must be positive.
+  */
+case class SchemaCacheSettings(
+  ttl: Duration = Duration.ofMillis(SchemaCacheSettings.DefaultTtlMs)
+) {
+  require(
+    !ttl.isZero && !ttl.isNegative,
+    s"elastic.schema-cache.ttl must be positive (ELASTIC_SCHEMA_CACHE_TTL), got $ttl"
+  )
+
+  def ttlMs: Long = ttl.toMillis
+}
+
+object SchemaCacheSettings {
+
+  /** Five minutes -- what `IndicesApi`, `ScrollApi` and `SearchApi` each hard-coded before this
+    * setting existed.
+    */
+  val DefaultTtlMs: Long = 5 * 60 * 1000L
+}