diff --git a/core/src/main/resources/help/commands/ddl/alter_table.json b/core/src/main/resources/help/commands/ddl/alter_table.json index 51bf7758..a60543e6 100644 --- a/core/src/main/resources/help/commands/ddl/alter_table.json +++ b/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", diff --git a/core/src/main/resources/softnetwork-elastic.conf b/core/src/main/resources/softnetwork-elastic.conf index f0299512..6123ed5a 100644 --- a/core/src/main/resources/softnetwork-elastic.conf +++ b/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 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 diff --git a/core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala b/core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala index 11b347c4..dd18dfac 100644 --- a/core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala +++ b/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 { diff --git a/core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala b/core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala index 8144b13a..fdfeb5c4 100644 --- a/core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala +++ b/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 { diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientApi.scala index 8812a19a..bd65d75c 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientApi.scala +++ b/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 } diff --git a/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala index abe61359..58fc672c 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala +++ b/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 ` 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 ` 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)) diff --git a/core/src/main/scala/app/softnetwork/elastic/client/SchemaCacheSettings.scala b/core/src/main/scala/app/softnetwork/elastic/client/SchemaCacheSettings.scala new file mode 100644 index 00000000..1477a037 --- /dev/null +++ b/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 +} diff --git a/core/src/main/scala/app/softnetwork/elastic/client/SchemaCacheTtlApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/SchemaCacheTtlApi.scala new file mode 100644 index 00000000..a3a5305c --- /dev/null +++ b/core/src/main/scala/app/softnetwork/elastic/client/SchemaCacheTtlApi.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.client + +import app.softnetwork.elastic.client.result.ElasticResult +import app.softnetwork.elastic.sql.schema.Schema + +/** The ONE schema-cache TTL derivation, shared by every cache that keys on an index (story 21.8 + * Part D). + * + * Precedence, matching the house `flag > env > file > default` convention: the index's own + * metadata (`ALTER TABLE … SET SCHEMA CACHE TTL`, read from the mapping when the schema is + * fetched) beats `elastic.schema-cache.ttl` (HOCON / `ELASTIC_SCHEMA_CACHE_TTL`), which beats the + * built-in five minutes. + * + * 🔴 Three caches used to hard-code five minutes in three files -- the schema cache + * ([[IndicesApi]]), the primary-shard-count cache ([[ScrollApi]], #238) and the 404 negative cache + * ([[SearchApi]], #276). They are already coupled (every schema-cache write or invalidation drops + * the shard counts naming that index), and a coupled pair that drifts is the failure nobody + * debugs, because each half reads as correct in isolation. Hence one trait, mixed into all three. + * + * The two members are deliberately different questions: + * - [[schemaCacheTtlMs]] — the DEFAULT, which is all a cache keyed by anything other than a + * concrete index can use; + * - [[schemaCacheTtlMsFor]] — the resolved TTL for one index, overridden by [[IndicesApi]] (the + * only trait that holds schemas) to consult what the index itself declares. + */ +trait SchemaCacheTtlApi { + + /** The default TTL for every cache keyed by an index. Overridden from `elastic.schema-cache.ttl` + * by [[ElasticClientApi]]; still a `protected def` so a subclass can pin it in a test. + */ + protected def schemaCacheTtlMs: Long = SchemaCacheSettings.DefaultTtlMs + + /** The resolved TTL for one index: what it declares, else [[schemaCacheTtlMs]]. + * + * 🔴 Answering must never cost a round trip — this is called on the paging path. [[IndicesApi]] + * answers it from the cache entry it already holds, so an index whose schema is not cached (or + * an expression that is not one index, such as `orders*`) resolves to the default. + */ + protected def schemaCacheTtlMsFor(index: String): Long = schemaCacheTtlMs +} + +/** One expiry rule, for every cache that stamps an entry with the TTL that governed it. + * + * 🔴 The rule is `final` and the entries carry their own `ttlMs` on purpose: the TTL is per index + * now, so re-spelling `now - cachedAt < ttl` at each call site would let two caches disagree about + * when the same index goes stale (story 21.8 D.2). + */ +private[client] trait CacheEntry { + def cachedAt: Long + def ttlMs: Long + final def isExpired(now: Long): Boolean = now - cachedAt >= ttlMs +} + +/** A cached schema and the TTL resolved for it when it was fetched — see + * [[SchemaCacheTtlApi.schemaCacheTtlMsFor]]. + */ +private[client] case class CachedSchema(schema: Schema, cachedAt: Long, ttlMs: Long) + extends CacheEntry + +/** A cached primary-shard count (#238) on the schema's clock: the shortest TTL among the indices + * the cache key names. + */ +private[client] case class CachedShardCount( + count: ElasticResult[Int], + cachedAt: Long, + ttlMs: Long +) extends CacheEntry diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala index 1dc80fcb..3a7de746 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala @@ -130,7 +130,7 @@ import scala.util.{Failure, Success} * @see * [[https://www.elastic.co/guide/en/elasticsearch/reference/7.10/point-in-time-api.html PIT API Documentation]] */ -trait ScrollApi extends ElasticClientHelpers { +trait ScrollApi extends ElasticClientHelpers with SchemaCacheTtlApi { _: VersionApi with SearchApi with SettingsApi => // ======================================================================== @@ -150,8 +150,10 @@ trait ScrollApi extends ElasticClientHelpers { */ protected def configuredMaxSlices: Int = ScrollConfig.DefaultMaxSlices - /** TTL, in milliseconds, of the primary-shard-count cache consulted by sliced PIT paging (#238). - * Override in a subclass to change (default: 5 minutes, the schema cache's TTL). One `_settings` + /** The primary-shard-count cache consulted by sliced PIT paging (#238), on the schema cache's + * clock: its TTL is [[SchemaCacheTtlApi.schemaCacheTtlMsFor]], the shortest among the indices a + * key names (story 21.8 Part D — a schema cached for an hour and a shard count re-probed every + * five minutes was incoherent, and two hard-coded constants in two files drift). One `_settings` * round-trip per distinct index set per TTL instead of one per un-LIMITed row query. What is * cached: a positive count, and a **privilege** failure (HTTP 401 / 403 — the one failure class * that is deterministic), so an under-privileged user sees ONE WARN per TTL instead of one per @@ -164,10 +166,8 @@ trait ScrollApi extends ElasticClientHelpers { * `invalidateAllSchemas` drops all of them; DDL issued through another client, and an entry * whose expression the index does not prefix, are seen after the TTL. */ - protected def shardCountCacheTtlMs: Long = 5 * 60 * 1000L - private val shardCountCache = - new java.util.concurrent.ConcurrentHashMap[String, (ElasticResult[Int], Long)]() + new java.util.concurrent.ConcurrentHashMap[String, CachedShardCount]() /** Above this many entries a cache miss also purges the expired ones (keys are index SETS, so a * long-lived server over date-suffixed or per-tenant indices would otherwise grow without @@ -175,6 +175,21 @@ trait ScrollApi extends ElasticClientHelpers { */ private val shardCountCachePurgeThreshold = 256 + /** The TTL a shard-count entry for these indices is stamped with — and the ONE derivation of it. + * + * The key names a SET of expressions, so the shortest TTL among them governs: no member may be + * remembered longer than it asked to be. A wildcard, or an index whose schema is not cached, + * resolves to the default, so this is the default until the schemas that shorten it are + * themselves loaded. + * + * 🔴 A `def`, called by BOTH the stamp and the message that tells an operator how long a + * privilege failure is remembered. Re-deriving it at the message would put the two halves of a + * coupled pair in two places again — which is the defect story 21.8 D.4 exists to close, and + * exactly the half nobody would notice was lying. + */ + private def shardCountTtlMs(indices: Seq[String]): Long = + indices.distinct.map(schemaCacheTtlMsFor).reduceOption(_ min _).getOrElse(schemaCacheTtlMs) + private def shardCountCacheable(result: ElasticResult[Int]): Boolean = result match { case ElasticSuccess(shards) => shards > 0 case ElasticFailure(err) => err.statusCode.exists(s => s == 401 || s == 403) @@ -189,15 +204,16 @@ trait ScrollApi extends ElasticClientHelpers { private[client] def cachedPrimaryShardCount( indices: Seq[String] ): (ElasticResult[Int], Boolean) = { - val key = indices.distinct.sorted.mkString(",") - val ttl = shardCountCacheTtlMs + val members = indices.distinct.sorted + val key = members.mkString(",") + val ttl = shardCountTtlMs(members) var result: ElasticResult[Int] = null var fromCache = true shardCountCache.compute( key, - (_: String, entry: (ElasticResult[Int], Long)) => { - if (entry != null && System.currentTimeMillis() - entry._2 < ttl) { - result = entry._1 + (_: String, entry: CachedShardCount) => { + if (entry != null && !entry.isExpired(System.currentTimeMillis())) { + result = entry.count entry } else { fromCache = false @@ -209,7 +225,7 @@ trait ScrollApi extends ElasticClientHelpers { ElasticFailure(err.copy(cause = None)) // never pin a stack case success => success } - (stored, System.currentTimeMillis()) + CachedShardCount(stored, System.currentTimeMillis(), ttl) } else null // not cached: a null mapping removes the (expired) entry } } @@ -218,9 +234,7 @@ trait ScrollApi extends ElasticClientHelpers { val now = System.currentTimeMillis() shardCountCache .entrySet() - .removeIf((e: java.util.Map.Entry[String, (ElasticResult[Int], Long)]) => - now - e.getValue._2 >= ttl - ) + .removeIf((e: java.util.Map.Entry[String, CachedShardCount]) => e.getValue.isExpired(now)) } (result, fromCache) } @@ -596,7 +610,7 @@ trait ScrollApi extends ElasticClientHelpers { * per primary shard of the resolved indices, capped by the ceiling, never above the shard count. * The guard order is load-bearing: no `_settings` round-trip on ORDER BY / LIMIT / opt-out / ES6 * / classic-scroll paths. The shard count comes from [[cachedPrimaryShardCount]] (one - * `_settings` round-trip per index set per [[shardCountCacheTtlMs]]); a lookup failure degrades + * `_settings` round-trip per index set per resolved schema-cache TTL); a lookup failure degrades * to sequential with a WARN — once per TTL for a privilege failure (DEBUG while the cached * failure is replayed), on every extraction for a transient one. */ @@ -635,11 +649,11 @@ trait ScrollApi extends ElasticClientHelpers { ) 1 case ElasticFailure(err) => - // only a privilege failure is remembered (see shardCountCacheTtlMs); anything else + // only a privilege failure is remembered (see cachedPrimaryShardCount); anything else // is probed again by the next extraction, and says so val why = err.statusCode match { case Some(s) if s == 401 || s == 403 => - s"the lookup needs the view_index_metadata privilege on the indices (HTTP $s) — remembered for ${shardCountCacheTtlMs / 1000} s" + s"the lookup needs the view_index_metadata privilege on the indices (HTTP $s) — remembered for ${shardCountTtlMs(elasticQuery.indices) / 1000} s" case _ => "retried on the next extraction" } diff --git a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala index ebbfc7e5..104db3d4 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala @@ -63,7 +63,7 @@ import scala.util.{Failure, Success, Try} * }}} */ //format:on -trait SearchApi extends ElasticConversion with ElasticClientHelpers { +trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaCacheTtlApi { /** Extract output field names from a SingleSearch in SQL SELECT order. For each field, uses the * alias if present, otherwise the source field name. Returns empty Seq for SELECT * queries. @@ -217,8 +217,23 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { private val schemaMissPurgeThreshold = 256 - /** How long a failed schema lookup is remembered -- the schema cache's own default TTL. */ - protected def schemaMissTtlMs: Long = 5 * 60 * 1000L + /** How long a failed schema lookup is remembered: the schema cache's DEFAULT TTL + * (`elastic.schema-cache.ttl`), never longer than the built-in five minutes. + * + * 🔴 Deliberately the default and never a per-index value (story 21.8 D.3.2): this map records a + * MISS. There is no metadata to read a TTL from, because there was no schema — an index that + * does not exist cannot ask to be forgotten sooner. + * + * 🔴 And deliberately CAPPED, which the two positive caches are not. Nothing invalidates a miss: + * it is cleared only by a later successful load, and `resolveWithSchema` does not even attempt + * the lookup while one stands. So a miss remembered for an operator-chosen hour would mean that + * querying a table before it exists, then creating it, leaves every statement against it running + * with NO schema attached for that hour — no `SQLTypeUtils.coerce`, no temporal-literal + * resolution, silently (#306's skip conditions). Shortening the TTL still shortens this; making + * the knob able to LENGTHEN it would turn a five-minute nuisance into an unbounded one. + */ + protected def schemaMissTtlMs: Long = + math.min(schemaCacheTtlMs, SchemaCacheSettings.DefaultTtlMs) /** Current size of the negative cache (tests). */ private[client] def schemaMissCount: Int = schemaMisses.size() diff --git a/core/src/test/scala/app/softnetwork/elastic/client/SchemaCacheTtlApiSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/SchemaCacheTtlApiSpec.scala new file mode 100644 index 00000000..f7c6f199 --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/client/SchemaCacheTtlApiSpec.scala @@ -0,0 +1,283 @@ +package app.softnetwork.elastic.client + +import app.softnetwork.elastic.client.result._ +import app.softnetwork.elastic.sql.schema.SchemaCacheTtl +import com.typesafe.config.{Config, ConfigFactory} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.slf4j.{Logger, LoggerFactory} + +import java.util.concurrent.atomic.AtomicInteger + +/** The per-index schema-cache TTL (story 21.8 Part D). + * + * Since #306 every executed query reads the cached schema, so a stale entry no longer means a + * stale column list — it means wrong emitted Painless. The TTL therefore stopped being a + * performance knob, and this spec pins the three things that makes necessary: it is configurable, + * an index may override it for itself, and the caches that key on an index agree about it. + * + * 🔴 Everything here is asserted through OBSERVED FETCHES, never through the resolver's return + * value alone: a resolver that answers `1h` while the cache re-reads every five minutes would pass + * a value-only test and fail every user. + */ +class SchemaCacheTtlApiSpec extends AnyFlatSpec with Matchers { + + private val testLogger: Logger = LoggerFactory.getLogger(getClass) + + private def mappingJson(ttl: Option[String]): String = { + val meta = ttl match { + case Some(v) => s""""_meta":{"${SchemaCacheTtl.MetadataKey}":"$v"},""" + case None => "" + } + s"""{"mappings":{$meta"properties":{"id":{"type":"keyword"}}}, + | "settings":{"index":{"number_of_shards":"3"}}}""".stripMargin + } + + /** @param declared + * index -> the TTL that index declares in its own `_meta` (None = declares none) + * @param aliasOf + * alias -> the concrete index `GET ` answers with + */ + private class TtlClient( + declared: Map[String, Option[String]] = Map.empty, + hocon: String = "", + aliasOf: Map[String, String] = Map.empty + ) extends NopeClientApi { + override protected def logger: Logger = testLogger + override def config: Config = ConfigFactory.parseString(hocon) + + val getIndexCalls = new AtomicInteger(0) + val settingsCalls = new AtomicInteger(0) + + override private[client] def executeGetIndex(index: String): ElasticResult[Option[String]] = { + getIndexCalls.incrementAndGet() + // `GET ` answers with the TARGET's document, so the metadata read is the target's. + val body = mappingJson(declared.getOrElse(aliasOf.getOrElse(index, index), None)) + ElasticSuccess( + Some(aliasOf.get(index).map(target => s"""{"$target":$body}""").getOrElse(body)) + ) + } + + override private[client] def executeLoadSettings(index: String): ElasticResult[String] = { + settingsCalls.incrementAndGet() + ElasticSuccess("""{"idx":{"settings":{"index":{"number_of_shards":"3"}}}}""") + } + + // The two resolutions under test, exposed: both are `protected` on the trait. + def defaultTtl: Long = schemaCacheTtlMs + def ttlFor(index: String): Long = schemaCacheTtlMsFor(index) + def missTtl: Long = schemaMissTtlMs + } + + private def load(client: TtlClient, index: String): Unit = + client.loadSchema(index) match { + case ElasticSuccess(_) => () + case failure => fail(s"loadSchema($index) failed: $failure") + } + + // -- the default ------------------------------------------------------------- + + "the default TTL" should "be five minutes when nothing configures it" in { + val client = new TtlClient() + client.defaultTtl shouldBe SchemaCacheSettings.DefaultTtlMs + load(client, "orders") + load(client, "orders") + client.getIndexCalls.get() shouldBe 1 // the second read is a cache hit + } + + it should "come from elastic.schema-cache.ttl" in { + val client = new TtlClient(hocon = "elastic.schema-cache.ttl = 42m") + client.defaultTtl shouldBe 42 * 60 * 1000L + } + + it should "be honoured, not merely reported" in { + val client = new TtlClient(hocon = "elastic.schema-cache.ttl = 1ms") + load(client, "orders") + Thread.sleep(20) + load(client, "orders") + client.getIndexCalls.get() shouldBe 2 // expired, so it was read again + } + + it should "refuse a non-positive value rather than cache forever" in { + // `ElasticConfig` reports a rejected setting as a ConfigException naming the key — the require + // in `SchemaCacheSettings` is what produces the message. + val failure = + the[Exception] thrownBy new TtlClient(hocon = "elastic.schema-cache.ttl = 0s").defaultTtl + failure.getMessage should include("elastic.schema-cache.ttl must be positive") + } + + // -- the per-index override -------------------------------------------------- + + "an index that declares its own TTL" should "override the configured default" in { + // The default has already expired; the index asked to be remembered for an hour. + val client = + new TtlClient( + declared = Map("orders" -> Some("1h")), + hocon = "elastic.schema-cache.ttl = 1ms" + ) + load(client, "orders") + client.ttlFor("orders") shouldBe 3600000L + Thread.sleep(20) + load(client, "orders") + client.getIndexCalls.get() shouldBe 1 // still a hit: the index's own clock governs + } + + it should "be able to SHORTEN a long default" in { + val client = + new TtlClient(declared = Map("logs" -> Some("1ms")), hocon = "elastic.schema-cache.ttl = 1h") + load(client, "logs") + client.ttlFor("logs") shouldBe 1L + Thread.sleep(20) + load(client, "logs") + client.getIndexCalls.get() shouldBe 2 + } + + it should "leave every OTHER index on the default" in { + val client = new TtlClient( + declared = Map("orders" -> Some("1h")), + hocon = "elastic.schema-cache.ttl = 30m" + ) + load(client, "orders") + load(client, "logs") + client.ttlFor("orders") shouldBe 3600000L + client.ttlFor("logs") shouldBe 30 * 60 * 1000L + load(client, "orders") + load(client, "logs") + client.getIndexCalls.get() shouldBe 2 // both still cached, neither re-read + } + + it should "fall back to the default, with a WARN, when what it declares is not a duration" in { + val client = new TtlClient(declared = Map("orders" -> Some("soon"))) + load(client, "orders") + client.ttlFor("orders") shouldBe SchemaCacheSettings.DefaultTtlMs + } + + "an index whose schema is not cached" should "resolve to the default" in { + // 🔴 Answering must never cost a round trip — this is called on the paging path. + val client = new TtlClient(declared = Map("orders" -> Some("1h"))) + client.ttlFor("orders") shouldBe SchemaCacheSettings.DefaultTtlMs + client.getIndexCalls.get() shouldBe 0 + } + + "an ALIAS" should "inherit the TTL of the index it resolves to" in { + // D.3.3: the alias entry caches the TARGET's schema, so it carries the target's `_meta`. + val client = new TtlClient( + declared = Map("orders_v2" -> Some("1h")), + hocon = "elastic.schema-cache.ttl = 1ms", + aliasOf = Map("orders" -> "orders_v2") + ) + load(client, "orders") + client.ttlFor("orders") shouldBe 3600000L + // …and observed, not merely reported: the default has long expired, so a re-read here would + // mean the alias had NOT inherited its target's clock. + Thread.sleep(20) + load(client, "orders") + client.getIndexCalls.get() shouldBe 1 + } + + // -- the coupled caches ------------------------------------------------------ + + "the shard-count cache" should "follow the TTL the index declares" in { + val client = + new TtlClient( + declared = Map("orders" -> Some("1h")), + hocon = "elastic.schema-cache.ttl = 1ms" + ) + load(client, "orders") + client.cachedPrimaryShardCount(Seq("orders")) + Thread.sleep(20) + client.cachedPrimaryShardCount(Seq("orders")) + client.settingsCalls.get() shouldBe 1 // the index asked for an hour; both caches obey it + } + + it should "expire with a shortened index TTL, not on its own five minutes" in { + val client = + new TtlClient(declared = Map("logs" -> Some("1ms")), hocon = "elastic.schema-cache.ttl = 1h") + load(client, "logs") + client.cachedPrimaryShardCount(Seq("logs")) + Thread.sleep(20) + client.cachedPrimaryShardCount(Seq("logs")) + client.settingsCalls.get() shouldBe 2 + } + + it should "take the SHORTEST TTL among the indices a key names" in { + // The key is an index SET: no member may be remembered longer than it asked to be. + val client = new TtlClient( + declared = Map("orders" -> Some("1h"), "logs" -> Some("1ms")), + hocon = "elastic.schema-cache.ttl = 1h" + ) + load(client, "orders") + load(client, "logs") + client.cachedPrimaryShardCount(Seq("orders", "logs")) + Thread.sleep(20) + client.cachedPrimaryShardCount(Seq("orders", "logs")) + // `primaryShardCount` probes `_settings` once per index expression, so a re-probe of this + // 2-index key is 4 calls in total; the control below is what makes that number mean something. + client.settingsCalls.get() shouldBe 4 + } + + it should "keep a set whose members ALL asked to be remembered" in { + val client = new TtlClient( + declared = Map("orders" -> Some("1h"), "logs" -> Some("1h")), + hocon = "elastic.schema-cache.ttl = 1ms" + ) + load(client, "orders") + load(client, "logs") + client.cachedPrimaryShardCount(Seq("orders", "logs")) + Thread.sleep(20) + client.cachedPrimaryShardCount(Seq("orders", "logs")) + client.settingsCalls.get() shouldBe 2 // one probe, served from the cache the second time + } + + "the 404 negative cache" should "use the DEFAULT, never a per-index value" in { + // D.3.2: it records a MISS — there was no schema, so there is no metadata to read a TTL from. + val client = + new TtlClient(declared = Map("orders" -> Some("1h")), hocon = "elastic.schema-cache.ttl = 3m") + load(client, "orders") + client.missTtl shouldBe 3 * 60 * 1000L + } + + it should "follow a SHORTENED default but never be lengthened past five minutes" in { + // Nothing invalidates a miss and no lookup is attempted while one stands, so an + // operator-lengthened TTL would leave a table created after a failed probe running with NO + // schema attached — no conversions, no temporal resolution — for that whole period. + new TtlClient(hocon = "elastic.schema-cache.ttl = 30s").missTtl shouldBe 30000L + new TtlClient(hocon = "elastic.schema-cache.ttl = 6h").missTtl shouldBe + SchemaCacheSettings.DefaultTtlMs + } + + // -- the bound --------------------------------------------------------------- + + "the schema cache" should "purge expired entries once it grows past its threshold" in { + // Unbounded while it was cold; since #306 every query writes to it, and a deployment that mints + // dated indices would grow without bound. + val client = new TtlClient(hocon = "elastic.schema-cache.ttl = 1ms") + (1 to 300).foreach(i => load(client, s"logs-$i")) + Thread.sleep(20) + load(client, "logs-301") + client.schemaCacheSize should be < 300 + } + + it should "keep entries that have NOT expired, whatever the map's size" in { + val client = new TtlClient(hocon = "elastic.schema-cache.ttl = 1h") + (1 to 300).foreach(i => load(client, s"logs-$i")) + load(client, "logs-301") + client.schemaCacheSize shouldBe 301 + } + + it should "drop the cache outright rather than grow without limit on long TTLs" in { + // Expiry alone bounds nothing when an index asks for a long TTL and the names keep changing + // (dated or per-tenant indices), and the values here are whole schemas. + val client = new TtlClient(hocon = "elastic.schema-cache.ttl = 30d") + (1 to 1100).foreach(i => load(client, s"logs-$i")) + client.schemaCacheSize should be < 1100 + } + + // -- the expiry rule --------------------------------------------------------- + + "the expiry rule" should "be shared, and treat exactly-the-TTL as expired" in { + val entry = CachedShardCount(ElasticSuccess(3), cachedAt = 1000L, ttlMs = 100L) + entry.isExpired(1099L) shouldBe false + entry.isExpired(1100L) shouldBe true + } +} diff --git a/core/src/test/scala/app/softnetwork/elastic/client/ScrollSlicingSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/ScrollSlicingSpec.scala index 1b4d6a16..a54a0d5b 100644 --- a/core/src/test/scala/app/softnetwork/elastic/client/ScrollSlicingSpec.scala +++ b/core/src/test/scala/app/softnetwork/elastic/client/ScrollSlicingSpec.scala @@ -94,7 +94,9 @@ class ScrollSlicingSpec override protected def logger: Logger = mockLogger override def version: ElasticResult[String] = ElasticSuccess(esVersion) override protected def configuredMaxSlices: Int = ceiling - override protected def shardCountCacheTtlMs: Long = ttlMs + // Story 21.8 Part D: the shard-count cache follows the schema cache's TTL; with no schema + // cached for these indices, that is this default. + override protected def schemaCacheTtlMs: Long = ttlMs val pitConfig = new AtomicReference[ScrollConfig]() val classicConfig = new AtomicReference[ScrollConfig]() diff --git a/documentation/client/common_principles.md b/documentation/client/common_principles.md index 4027a254..c85d2e91 100644 --- a/documentation/client/common_principles.md +++ b/documentation/client/common_principles.md @@ -520,6 +520,15 @@ elastic { max-slices = 8 } + # How long a table's schema -- and the primary shard count that follows it -- may be cached + # before Elasticsearch is read again. This is the DEFAULT: an index that declares its own + # (ALTER TABLE SET SCHEMA CACHE TTL = '10m') overrides it for itself. Since every executed + # statement reads the cached schema, a stale entry means Painless emitted for the previous + # mapping -- shorten this for mappings that change under a running client. + schema-cache { + ttl = 5m + } + # Cluster discovery discovery { enabled = false diff --git a/documentation/client/scroll.md b/documentation/client/scroll.md index 59740ae5..14dc2fab 100644 --- a/documentation/client/scroll.md +++ b/documentation/client/scroll.md @@ -162,8 +162,9 @@ clock improve when shards and nodes are added. | Otherwise | `max(1, min(Σ number_of_shards of the resolved indices, ceiling))` | The shard count comes from `GET /_settings` (the indices a wildcard, alias or data stream -resolves to are summed and deduplicated) and is **cached per index set for 5 minutes** — the same -TTL as the schema cache (`shardCountCacheTtlMs`, overridable in a subclass) — so a workload of many +resolves to are summed and deduplicated) and is **cached per index set on the schema cache's own +clock** — `elastic.schema-cache.ttl` (5 minutes by default), or the shortest TTL the indices in the +key declare for themselves via `ALTER TABLE … SET SCHEMA CACHE TTL` — so a workload of many small un-LIMITed queries pays one round-trip per table per TTL, not one per query (concurrent cold extractions of the same set share one lookup). What is remembered: a positive count, and a **privilege** failure (HTTP 401/403 — the credentials lack `view_index_metadata`), for which the diff --git a/documentation/client/search.md b/documentation/client/search.md index 484906ee..94ea38f0 100644 --- a/documentation/client/search.md +++ b/documentation/client/search.md @@ -1518,10 +1518,11 @@ local numbers; treat them as orders of magnitude. Two distinct costs, with different profiles: -- **The load** is amortised by the schema cache (5-minute TTL, `schemaCacheTtlMs`). A hit is a +- **The load** is amortised by the schema cache (`elastic.schema-cache.ttl`, 5 minutes by + default, and an index may set its own — see [Tuning](#tuning)). A hit is a concurrent-map lookup and a timestamp compare — 0.1 µs, i.e. nothing. A miss additionally costs one `GET ` round trip to Elasticsearch plus the parse above. **Parse cost scales with - mapping width**, roughly linearly: a 300-field index costs ~1 ms to parse, once per five minutes. + mapping width**, roughly linearly: a 300-field index costs ~1 ms to parse, once per TTL. - **The attach** never amortises — it rebuilds the AST on every statement. It scales with **statement size**, not mapping width: a 300-field mapping costs no more than a 5-field one for a trivial statement. It runs **once per statement**, or **twice** for an un-`LIMIT`ed row query, @@ -1537,15 +1538,56 @@ of milliseconds to tens of milliseconds — well under 0.1%. The one case worth The mapping parse is the only cost that a longer TTL amortises, and it is the one that grows with your mapping: at ~1 ms for a 300-field index, a workload spread across many wide indices pays that -once per index per TTL. Raising `schemaCacheTtlMs` trades staleness — a mapping changed outside this -client is not seen until the entry expires — for fewer parses. The cache-hit and AST-attach costs -are unaffected by the TTL, so tuning it does nothing for a workload against a single narrow index. +once per index per TTL. The cache-hit and AST-attach costs are unaffected by the TTL, so tuning it +does nothing for a workload against a single narrow index. + +⚠️ **The TTL is a correctness knob, not only a performance one.** Because the schema now reaches +every executed statement, a stale entry no longer means a stale column list — it means Painless +emitted for the *previous* mapping. Retype a column from `keyword` to `long` and, for the rest of +the TTL, this client keeps emitting `Long.parseLong(doc['x'].value)` against a field that is already +numeric. Shorten the TTL for mappings that change under a running client; lengthen it for mappings +that do not. + +**The client default** (all indices): + +```hocon +elastic.schema-cache.ttl = 5m # or the ELASTIC_SCHEMA_CACHE_TTL environment variable +``` + +**Per index** — the volatility of a mapping is a property of the index, so an index can carry its +own, and it wins over the client default: + +```sql +ALTER TABLE orders SET SCHEMA CACHE TTL = '1h'; -- a mapping that never moves +ALTER TABLE feature_flags SET SCHEMA CACHE TTL = '30s'; +ALTER TABLE orders DROP SCHEMA CACHE TTL; -- back to the client default +``` + +The value is stored in the index's own mapping metadata (`_meta.schema_cache_ttl`), so every client +against that cluster picks it up; `ALTER TABLE … SET MAPPING _meta.schema_cache_ttl = '1h'` and +`CREATE TABLE … OPTIONS (mappings = (_meta = (schema_cache_ttl = '1h')))` write exactly the same +thing. Durations are written the way HOCON writes them (`30s`, `10m`, `1h`, or a bare number of +milliseconds); the DDL form rejects anything else at parse time. + +Precedence: **index metadata > `elastic.schema-cache.ttl` > the built-in 5 minutes.** + +⚠️ Changing an index's TTL is **self-referential**: another client learns the new value only when +*its* current entry expires, so the **old** TTL governs how fast the new one is noticed. Shortening +an hour to a minute therefore takes effect within the hour, not within the minute. ### Operational note -Expect one `GET ` per distinct index per 5 minutes from each client instance. Concurrent +Expect one `GET ` per distinct index per TTL from each client instance. Concurrent first-touch queries do **not** stampede: the cache is populated under `ConcurrentHashMap.compute`, -so only one of them fetches. +so only one of them fetches. Past 256 entries a miss also drops the expired ones, each on its own +clock, and past 1024 live entries it drops the cache outright — a long per-index TTL over +ever-changing index names (dated or per-tenant indices) would otherwise grow it without limit. The +cost of a drop is one re-read per index. + +The **primary shard-count cache** used by sliced paging follows the same clock (see +[scroll](scroll.md)). The negative cache that remembers a 404 follows it only downwards: a *miss* +has no metadata to read a per-index TTL from, and nothing invalidates it — so lowering +`elastic.schema-cache.ttl` shortens it, while raising the setting leaves it at five minutes. ## Performance Optimization diff --git a/documentation/sql/ddl_statements.md b/documentation/sql/ddl_statements.md index 4e1788c9..c741ed35 100644 --- a/documentation/sql/ddl_statements.md +++ b/documentation/sql/ddl_statements.md @@ -435,6 +435,8 @@ a good value with `NULL`. - `DROP SETTING key` - `SET|ADD ALIAS alias_name = value` - `DROP ALIAS alias_name` +- `SET SCHEMA CACHE TTL [=] 'duration'` +- `DROP SCHEMA CACHE TTL` ### Table-level clauses take no parentheses @@ -456,6 +458,42 @@ A value may be a scalar (`'1s'`, `true`, `2`), an array (`['a', 'b']`), or a nes written with **parentheses**: `(key = value, key = (nested = value))`. The `{…}` brace form is for `STRUCT` column values in `INSERT`, and is not accepted here. +### SET SCHEMA CACHE TTL + +How long a client may cache this table's schema before reading it from Elasticsearch again. The +volatility of a mapping is a property of the index, so the value lives with the index and overrides +the client's `elastic.schema-cache.ttl` default for this table alone: + +```sql +ALTER TABLE orders SET SCHEMA CACHE TTL = '1h'; +ALTER TABLE orders DROP SCHEMA CACHE TTL; +``` + +The `=` is optional (`SET SCHEMA CACHE TTL '1h'`). Durations use the HOCON spellings — `30s`, +`10m`, `1h`, or a bare number of milliseconds; anything else is rejected when the statement is +parsed, before it reaches the cluster. + +This is sugar: it writes `_meta.schema_cache_ttl`, so it is exactly equivalent to + +```sql +ALTER TABLE orders SET MAPPING _meta.schema_cache_ttl = '1h'; +``` + +and a table can be created with one: + +```sql +CREATE TABLE IF NOT EXISTS orders ( + id INT NOT NULL +) OPTIONS (mappings = (_meta = (schema_cache_ttl = '1h'))); +``` + +Since the schema now reaches every executed statement, a stale entry means Painless emitted for the +*previous* mapping, not merely a stale column list — so shorten this for a table whose mapping +changes under a running client. Note that changing it is self-referential: another client notices +the new value only when its current entry expires, i.e. after at most one **old** period. The +primary shard-count cache used by sliced paging follows the same TTL. See +[client/search](../client/search.md#tuning). + ### Type Changes and Safety When applying `ALTER COLUMN column_name SET DATA TYPE new_type`, the SQL Gateway computes a structural diff between the current schema and the target schema. diff --git a/documentation/sql/dql_statements.md b/documentation/sql/dql_statements.md index 3a90e690..bce12620 100644 --- a/documentation/sql/dql_statements.md +++ b/documentation/sql/dql_statements.md @@ -462,13 +462,15 @@ WHERE event_ts >= '2026-06-04T00:00:00' -- what Elasticsearch's default - `keyword`/`text` columns, `LIKE`/`RLIKE` patterns, function-wrapped columns (`YEAR(event_ts)`), `date_nanos` columns, columns qualified with a `JOIN` alias (the FROM table's own columns are resolved) and `HAVING` conditions are never touched. -- The resolution needs the index mapping, loaded through the schema cache (one lookup per index - every 5 minutes, and only for statements whose `WHERE` compares a string literal to a column). An +- The resolution needs the index mapping, loaded through the schema cache (one lookup per index per + TTL — `elastic.schema-cache.ttl`, 5 minutes by default, and an index may set its own with + `ALTER TABLE … SET SCHEMA CACHE TTL`). An **index alias over exactly one index** resolves to that index's mapping (`SHOW TABLE` / `DESCRIBE` through such an alias resolve the same way); an alias over **several** indices is ambiguous and is treated as unresolvable. When the statement reads several indices or a wildcard, or the mapping cannot be loaded, the literal is forwarded verbatim as in previous releases, and a - failed mapping lookup is remembered for 5 minutes so it is not retried on every statement. + failed mapping lookup is remembered for the DEFAULT TTL (a miss has no index metadata to read a + per-index one from) so it is not retried on every statement. --- diff --git a/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientPerIndexSchemaCacheTtlSpec.scala b/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientPerIndexSchemaCacheTtlSpec.scala new file mode 100644 index 00000000..1bffee4a --- /dev/null +++ b/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientPerIndexSchemaCacheTtlSpec.scala @@ -0,0 +1,27 @@ +/* + * 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 + +/** Pinned to 6.7.2 like every other jest spec that runs DDL (`JestGatewayApiSpec`, + * `JestClientTemplateApiSpec`, …): on 6.8 this client sends a typed mapping in which `_meta` is + * read as the TYPE name, so any `CREATE TABLE` — which always writes `_meta.columns` — is rejected + * with `Root mapping definition has unsupported parameters`. Pre-existing and unrelated to the + * TTL; the ES 6 rest client on 6.8 runs the same DDL fine. + */ +class JestClientPerIndexSchemaCacheTtlSpec extends PerIndexSchemaCacheTtlSpec { + override def elasticVersion: String = "6.7.2" +} diff --git a/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientPerIndexSchemaCacheTtlSpec.scala b/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientPerIndexSchemaCacheTtlSpec.scala new file mode 100644 index 00000000..5390e25d --- /dev/null +++ b/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientPerIndexSchemaCacheTtlSpec.scala @@ -0,0 +1,19 @@ +/* + * 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 + +class RestHighLevelClientPerIndexSchemaCacheTtlSpec extends PerIndexSchemaCacheTtlSpec diff --git a/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientPerIndexSchemaCacheTtlSpec.scala b/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientPerIndexSchemaCacheTtlSpec.scala new file mode 100644 index 00000000..5390e25d --- /dev/null +++ b/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientPerIndexSchemaCacheTtlSpec.scala @@ -0,0 +1,19 @@ +/* + * 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 + +class RestHighLevelClientPerIndexSchemaCacheTtlSpec extends PerIndexSchemaCacheTtlSpec diff --git a/es8/java/src/test/scala/app/softnetwork/elastic/client/JavaClientPerIndexSchemaCacheTtlSpec.scala b/es8/java/src/test/scala/app/softnetwork/elastic/client/JavaClientPerIndexSchemaCacheTtlSpec.scala new file mode 100644 index 00000000..d9de698e --- /dev/null +++ b/es8/java/src/test/scala/app/softnetwork/elastic/client/JavaClientPerIndexSchemaCacheTtlSpec.scala @@ -0,0 +1,19 @@ +/* + * 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 + +class JavaClientPerIndexSchemaCacheTtlSpec extends PerIndexSchemaCacheTtlSpec diff --git a/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientPerIndexSchemaCacheTtlSpec.scala b/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientPerIndexSchemaCacheTtlSpec.scala new file mode 100644 index 00000000..d9de698e --- /dev/null +++ b/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientPerIndexSchemaCacheTtlSpec.scala @@ -0,0 +1,19 @@ +/* + * 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 + +class JavaClientPerIndexSchemaCacheTtlSpec extends PerIndexSchemaCacheTtlSpec diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/SQLKeywords.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/SQLKeywords.scala index 115a1162..b1445856 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/SQLKeywords.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/SQLKeywords.scala @@ -352,6 +352,7 @@ object SQLKeywords { "AS", "AT", "BY", + "CACHE", "CLUSTER", "COLUMN", "COMMENT", @@ -424,6 +425,7 @@ object SQLKeywords { "REPLACE", "RETURNS", "SCHEDULE", + "SCHEMA", "SCRIPT", "STORED", "SECOND", @@ -436,6 +438,7 @@ object SQLKeywords { "TO", "TRUE", "TRUNCATE", + "TTL", "TYPE", "UPDATE", "USING", diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/parser/Parser.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/parser/Parser.scala index 798c2f77..fdd5304a 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/parser/Parser.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/parser/Parser.scala @@ -37,6 +37,7 @@ import app.softnetwork.elastic.sql.schema.{ IngestProcessor, IngestProcessorType, PartitionDate, + SchemaCacheTtl, ScriptProcessor } import app.softnetwork.elastic.sql.time.TimeUnit @@ -740,6 +741,30 @@ object Parser def dropTableSetting: PackratParser[DropTableSetting] = (keyword("DROP") ~ keyword("SETTING")) ~> identName ^^ { m => DropTableSetting(m) } + /** `SET SCHEMA CACHE TTL = '10m'` — sugar over the metadata write it desugars to, NOT a second + * way of storing the same thing (story 21.8 Part D). The TTL lives at + * `SchemaCacheTtl.MetadataPath` whichever spelling wrote it, so `SET MAPPING`, `CREATE TABLE … + * OPTIONS` and this production produce one AST, one diff, one mapping update and one read. + * + * The duration is validated HERE, by the same parse the client applies, so a misspelled TTL is + * refused at parse time rather than silently ignored for the lifetime of the index. `err`, never + * `throw`: `Parser.apply` is typed `Either[ParserError, Statement]` (#250). + */ + def alterTableSchemaCacheTtl: PackratParser[AlterTableMapping] = + ((keyword("SET") ~ keyword("SCHEMA") ~ keyword("CACHE") ~ keyword( + "TTL" + )) ~ "=".? ~ literal) >> { case _ ~ _ ~ ttl => + SchemaCacheTtl.parse(ttl.value) match { + case Right(_) => success(AlterTableMapping(SchemaCacheTtl.MetadataPath, ttl)) + case Left(reason) => err(s"Invalid ${SchemaCacheTtl.Ddl}: $reason") + } + } + + def dropTableSchemaCacheTtl: PackratParser[DropTableMapping] = + (keyword("DROP") ~ keyword("SCHEMA") ~ keyword("CACHE") ~ keyword("TTL")) ^^ { _ => + DropTableMapping(SchemaCacheTtl.MetadataPath) + } + def alterTableAlias: PackratParser[AlterTableAlias] = ((keyword("SET") | keyword("ADD")) ~ keyword("ALIAS")) ~ option ^^ { case _ ~ opt => AlterTableAlias(opt._1, opt._2) @@ -767,6 +792,8 @@ object Parser alterColumnFields | alterColumnField | dropColumnField | + alterTableSchemaCacheTtl | + dropTableSchemaCacheTtl | alterTableMapping | dropTableMapping | alterTableSetting | diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/schema/SchemaCacheTtl.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/schema/SchemaCacheTtl.scala new file mode 100644 index 00000000..359e014b --- /dev/null +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/schema/SchemaCacheTtl.scala @@ -0,0 +1,97 @@ +/* + * 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.schema + +import app.softnetwork.elastic.sql.{ObjectValue, Value} +import com.typesafe.config.{ConfigFactory, ConfigValueFactory} + +import scala.util.{Failure, Success, Try} + +/** How long a table's schema may be cached by a client -- where it is written, how it is spelled, + * and the ONE parse of it (story 21.8 Part D). + * + * The value lives in the index's own mapping metadata, at [[MetadataPath]], because the volatility + * of a mapping is a property of the index. Three surfaces write and read it, and all three go + * through this object so they cannot disagree: + * - `ALTER TABLE SET SCHEMA CACHE TTL = '10m'` (and `CREATE TABLE … OPTIONS (mappings = + * (_meta = (schema_cache_ttl = '10m')))`), which the parser validates with [[parse]] so a + * misspelled duration is refused at parse time rather than silently ignored a page later; + * - `IndicesApi`, which stamps the resolved TTL on the cache entry when the schema is fetched; + * - `ScrollApi`'s shard-count cache, which follows the same clock (a schema cached for an hour + * and a shard count re-probed every five minutes is incoherent). + * + * 🔴 The TTL is a CORRECTNESS-latency knob, not a performance one. Since #306 every executed query + * reads the cached schema, so a stale entry no longer means a stale column list -- it means wrong + * emitted Painless (retype a column `keyword` -> `long` and the cached `keyword` schema keeps + * emitting `Long.parseLong(doc['x'].value)` against a numeric field for the rest of the TTL). + */ +object SchemaCacheTtl { + + /** The `_meta` key. Snake case, like every other key this project writes into `_meta` + * (`primary_key`, `partition_by`, `materialized_views`). + */ + val MetadataKey: String = "schema_cache_ttl" + + /** The dotted path an `ALTER TABLE … SET MAPPING` writes -- `ObjectValue.set` splits on `.`. */ + val MetadataPath: String = s"_meta.$MetadataKey" + + /** The spelling of the dedicated DDL sugar, for error messages. */ + val Ddl: String = "SET SCHEMA CACHE TTL" + + /** Parse a written TTL into milliseconds. + * + * HOCON duration syntax (`10m`, `30 seconds`, `1h`, a bare number of milliseconds) through + * Typesafe Config itself, so the per-index spelling and the `elastic.schema-cache.ttl` default + * accept exactly the same forms -- two duration parsers would be two sets of accepted spellings. + * The raw text is passed as a config VALUE, never interpolated into parsed text, so a `"` or a + * newline in it cannot become syntax. + * + * @return + * the TTL in milliseconds, or the reason it is not one + */ + def parse(raw: String): Either[String, Long] = { + val trimmed = Option(raw).map(_.trim).getOrElse("") + if (trimmed.isEmpty) Left("a schema cache TTL cannot be empty") + else + Try( + ConfigFactory.empty + .withValue("ttl", ConfigValueFactory.fromAnyRef(trimmed)) + .getDuration("ttl") + .toMillis + ) match { + case Success(ms) if ms > 0 => Right(ms) + case Success(ms) => + Left(s"a schema cache TTL must be positive, got '$trimmed' ($ms ms)") + case Failure(_) => + Left( + s"'$trimmed' is not a duration — write it as '30s', '10m', '1h' or a number of milliseconds" + ) + } + } + + /** The TTL written on this table, if any: `None` when the table declares none (the caller's + * default governs), `Some(Left(reason))` when it declares one that is not a duration. + * + * A table read back from Elasticsearch carries `_meta` verbatim in `Table.mappings` + * (`IndexMappings.options`), and `Table.update()` rebuilds only the five keys it owns, so this + * key survives every round trip. + */ + def of(table: Table): Option[Either[String, Long]] = + ObjectValue(table.mappings) + .find(MetadataPath) + .map((v: Value[_]) => parse(String.valueOf(v.value))) +} diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/schema/SchemaCacheTtlSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/schema/SchemaCacheTtlSpec.scala new file mode 100644 index 00000000..b38cd076 --- /dev/null +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/schema/SchemaCacheTtlSpec.scala @@ -0,0 +1,187 @@ +package app.softnetwork.elastic.sql.schema + +import app.softnetwork.elastic.sql.StringValue +import app.softnetwork.elastic.sql.`type`.SQLTypes +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.query.{ + AlterTable, + AlterTableMapping, + AlterTableStatement, + CreateTable, + DropTableMapping +} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +/** The per-index schema-cache TTL (story 21.8 Part D). + * + * Two spellings write ONE thing: the dedicated `SET SCHEMA CACHE TTL` is sugar that desugars to + * the `_meta` write `SET MAPPING` performs, so there is a single AST, a single diff and a single + * read. These tests pin that identity — sugar that produced a second representation would be a + * second place for the value to be missing. + */ +class SchemaCacheTtlSpec extends AnyFlatSpec with Matchers { + + private def statementsOf(sql: String): Seq[AlterTableStatement] = + Parser(sql) match { + case Right(AlterTable(_, _, statements, _)) => statements + case other => fail(s"Expected an AlterTable, got $other") + } + + private val users: Table = + Table( + name = "users", + columns = List(Column("id", SQLTypes.Int), Column("name", SQLTypes.Varchar)), + primaryKey = List("id") + ).update() + + "parse" should "accept the HOCON duration spellings the configuration default accepts" in { + SchemaCacheTtl.parse("10m") shouldBe Right(600000L) + SchemaCacheTtl.parse("30s") shouldBe Right(30000L) + SchemaCacheTtl.parse("1h") shouldBe Right(3600000L) + SchemaCacheTtl.parse("500 milliseconds") shouldBe Right(500L) + SchemaCacheTtl.parse("750") shouldBe Right(750L) // bare number = milliseconds + SchemaCacheTtl.parse(" 10m ") shouldBe Right(600000L) + } + + it should "reject what is not a positive duration, naming the accepted forms" in { + SchemaCacheTtl.parse("soon").left.map(_.contains("not a duration")) shouldBe Left(true) + SchemaCacheTtl.parse("").left.map(_.contains("cannot be empty")) shouldBe Left(true) + SchemaCacheTtl.parse("0s").left.map(_.contains("must be positive")) shouldBe Left(true) + SchemaCacheTtl.parse("-5m").left.map(_.contains("must be positive")) shouldBe Left(true) + } + + "SET SCHEMA CACHE TTL" should "desugar to the same _meta write SET MAPPING performs" in { + val sugar = statementsOf("ALTER TABLE users SET SCHEMA CACHE TTL = '10m'") + val explicit = + statementsOf("ALTER TABLE users SET MAPPING _meta.schema_cache_ttl = '10m'") + sugar shouldBe explicit + sugar shouldBe Seq(AlterTableMapping(SchemaCacheTtl.MetadataPath, StringValue("10m"))) + } + + it should "accept the equals sign as optional" in { + statementsOf("ALTER TABLE users SET SCHEMA CACHE TTL '10m'") shouldBe + statementsOf("ALTER TABLE users SET SCHEMA CACHE TTL = '10m'") + } + + it should "be case-insensitive like every other statement keyword" in { + statementsOf("alter table users set schema cache ttl = '10m'") shouldBe + Seq(AlterTableMapping(SchemaCacheTtl.MetadataPath, StringValue("10m"))) + } + + it should "re-parse from its own render" in { + // The render is the `SET MAPPING` form (that IS the AST); what matters is that running the + // rendered DDL back through the parser yields the same statement — MV deployment renders a + // schema to DDL and executes the text. + val stmts = statementsOf("ALTER TABLE users SET SCHEMA CACHE TTL = '10m'") + val rendered = s"ALTER TABLE users ${stmts.map(_.sql).mkString}" + statementsOf(rendered) shouldBe stmts + } + + it should "reject a TTL that is not a duration, at parse time" in { + val rejection = Parser("ALTER TABLE users SET SCHEMA CACHE TTL = 'soon'") + rejection.isLeft shouldBe true + val reason = rejection.swap.map(_.msg).getOrElse("") + reason should include("SET SCHEMA CACHE TTL") + reason should include("not a duration") + // 21.4: a rejection assertion is unfalsifiable unless it also excludes the boundary catch. + reason should not startWith Parser.InternalParseFailure + } + + it should "reject a non-positive TTL at parse time" in { + val rejection = Parser("ALTER TABLE users SET SCHEMA CACHE TTL = '0s'") + rejection.isLeft shouldBe true + val reason = rejection.swap.map(_.msg).getOrElse("") + reason should include("must be positive") + reason should not startWith Parser.InternalParseFailure + } + + "DROP SCHEMA CACHE TTL" should "desugar to the matching _meta removal" in { + statementsOf("ALTER TABLE users DROP SCHEMA CACHE TTL") shouldBe + statementsOf("ALTER TABLE users DROP MAPPING _meta.schema_cache_ttl") + } + + "merge" should "store the TTL where SchemaCacheTtl.of reads it, and survive update()" in { + val merged = users.merge(statementsOf("ALTER TABLE users SET SCHEMA CACHE TTL = '10m'")) + SchemaCacheTtl.of(merged) shouldBe Some(Right(600000L)) + // `Table.update()` runs on every schema and REBUILDS the five `_meta` keys it owns; a key it + // does not own must survive, or the TTL would vanish on the first round trip. + SchemaCacheTtl.of(merged.update()) shouldBe Some(Right(600000L)) + users.diff(merged).mappings should not be empty + } + + it should "remove it on DROP" in { + val withTtl = users.merge(statementsOf("ALTER TABLE users SET SCHEMA CACHE TTL = '10m'")) + val dropped = withTtl.merge(statementsOf("ALTER TABLE users DROP SCHEMA CACHE TTL")) + SchemaCacheTtl.of(dropped) shouldBe None + withTtl.diff(dropped).mappings should not be empty + } + + it should "report a hand-written TTL that is not a duration rather than throw" in { + // `SET MAPPING` and a hand-edited `_meta` bypass the sugar's parse-time validation. + val merged = + users.merge(statementsOf("ALTER TABLE users SET MAPPING _meta.schema_cache_ttl = 'soon'")) + SchemaCacheTtl.of(merged).map(_.isLeft) shouldBe Some(true) + } + + "a table declaring no TTL" should "read as None, not as a default" in { + SchemaCacheTtl.of(users) shouldBe None + } + + // The three spellings documented in `documentation/sql/ddl_statements.md` are asserted to be + // equivalent, not merely to parse: a doc example that parses can still mean something else. + "CREATE TABLE ... OPTIONS" should "be able to declare the TTL, identically" in { + val created = Parser( + """CREATE TABLE IF NOT EXISTS orders ( + | id INT NOT NULL + |) OPTIONS (mappings = (_meta = (schema_cache_ttl = '1h')))""".stripMargin + ) match { + case Right(create: CreateTable) => create.schema + case other => fail(s"Expected a CreateTable, got $other") + } + SchemaCacheTtl.of(created) shouldBe Some(Right(3600000L)) + // …and identical to what the ALTER sugar writes on the same table. + val altered = created + .copy(mappings = created.mappings - "_meta") + .merge(statementsOf("ALTER TABLE orders SET SCHEMA CACHE TTL = '1h'")) + SchemaCacheTtl.of(altered) shouldBe SchemaCacheTtl.of(created) + } + + "the documented examples" should "parse and mean what the documentation says" in { + statementsOf("ALTER TABLE orders SET SCHEMA CACHE TTL = '1h'") shouldBe + Seq(AlterTableMapping(SchemaCacheTtl.MetadataPath, StringValue("1h"))) + statementsOf("ALTER TABLE feature_flags SET SCHEMA CACHE TTL = '30s'") shouldBe + Seq(AlterTableMapping(SchemaCacheTtl.MetadataPath, StringValue("30s"))) + statementsOf("ALTER TABLE orders SET SCHEMA CACHE TTL '1h'") shouldBe + Seq(AlterTableMapping(SchemaCacheTtl.MetadataPath, StringValue("1h"))) + statementsOf("ALTER TABLE orders DROP SCHEMA CACHE TTL") shouldBe + Seq(DropTableMapping(SchemaCacheTtl.MetadataPath)) + statementsOf("ALTER TABLE orders SET MAPPING _meta.schema_cache_ttl = '1h'") shouldBe + Seq(AlterTableMapping(SchemaCacheTtl.MetadataPath, StringValue("1h"))) + // The single-line OPTIONS spelling published on the documentation site, and the + // semicolon-terminated forms every site example uses. + Parser( + "CREATE TABLE orders (id INT NOT NULL) OPTIONS (mappings = (_meta = (schema_cache_ttl = '1h')));" + ).isRight shouldBe true + Parser("ALTER TABLE orders SET SCHEMA CACHE TTL = '1h';").isRight shouldBe true + Parser("ALTER TABLE orders DROP SCHEMA CACHE TTL;").isRight shouldBe true + } + + // The three new keywords (SCHEMA, CACHE, TTL) are matched only after `SET`/`DROP` at the start of + // an ALTER TABLE statement. They must not become reserved words anywhere else — adding a keyword + // that swallows an identifier is how `COUNT(*) AS count` became a parse error (#302). + "the new keywords" should "not be reserved as identifiers elsewhere" in { + statementsOf("ALTER TABLE users SET MAPPING schema = 'analytics'") shouldBe + Seq(AlterTableMapping("schema", StringValue("analytics"))) + statementsOf("ALTER TABLE users SET SETTING cache = 'x'") should have size 1 + statementsOf("ALTER TABLE users ADD COLUMN cache INT") should have size 1 + statementsOf("ALTER TABLE users ADD COLUMN ttl INT") should have size 1 + Parser("SELECT schema FROM users").isRight shouldBe true + } + + it should "reject an unquoted duration rather than read it as something else" in { + val rejection = Parser("ALTER TABLE users SET SCHEMA CACHE TTL = 10m") + rejection.isLeft shouldBe true + rejection.swap.map(_.msg).getOrElse("") should not startWith Parser.InternalParseFailure + } +} diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/PerIndexSchemaCacheTtlSpec.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/PerIndexSchemaCacheTtlSpec.scala new file mode 100644 index 00000000..0cdeca21 --- /dev/null +++ b/testkit/src/main/scala/app/softnetwork/elastic/client/PerIndexSchemaCacheTtlSpec.scala @@ -0,0 +1,131 @@ +/* + * 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 akka.actor.ActorSystem +import app.softnetwork.elastic.client.result.{ElasticFailure, ElasticSuccess} +import app.softnetwork.elastic.client.spi.ElasticClientFactory +import app.softnetwork.elastic.scalatest.ElasticDockerTestKit +import app.softnetwork.elastic.sql.schema.{Schema, SchemaCacheTtl} +import app.softnetwork.persistence.generateUUID +import org.scalatest.concurrent.ScalaFutures +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.matchers.should.Matchers +import org.slf4j.{Logger, LoggerFactory} + +import scala.concurrent.duration._ + +/** An index carries its own schema-cache TTL, in its own mapping metadata (story 21.8 Part D). + * + * The unit tests pin the RESOLUTION (metadata > `elastic.schema-cache.ttl` > built-in). What only + * a real cluster can settle is the ROUND TRIP: that `ALTER TABLE … SET SCHEMA CACHE TTL` reaches + * Elasticsearch's `_meta`, comes back through `loadSchema`, and — the part that has bitten this + * project before — SURVIVES a later ALTER, because `Table.update()` rebuilds the `_meta` keys it + * owns on every schema (a flag not recoverable from the column list is erased on the first round + * trip; see MEDIUM-7). + */ +trait PerIndexSchemaCacheTtlSpec + extends AnyFlatSpecLike + with ElasticDockerTestKit + with Matchers + with ScalaFutures { + + lazy val log: Logger = LoggerFactory.getLogger(getClass.getName) + + implicit val system: ActorSystem = ActorSystem(generateUUID()) + + implicit val patience: PatienceConfig = + PatienceConfig(timeout = 30.seconds, interval = 100.millis) + + lazy val client: ElasticClientApi = ElasticClientFactory.create(elasticConfig) + + private val index = "schema_cache_ttl" + + override def beforeAll(): Unit = { + super.beforeAll() + ddl(s"""CREATE TABLE IF NOT EXISTS $index ( + | id INT NOT NULL, + | name VARCHAR + |)""".stripMargin) + } + + override def afterAll(): Unit = { + client.deleteIndex(index) + system.terminate() + super.afterAll() + } + + private def ddl(sql: String): Unit = + client.run(sql).futureValue match { + case ElasticSuccess(_) => () + case ElasticFailure(error) => fail(s"[$sql] failed: ${error.message}") + } + + /** The schema as Elasticsearch answers it — never the in-memory table the DDL produced. The cache + * is invalidated first so this is a genuine round trip. + */ + private def freshSchema: Schema = { + client.invalidateSchema(index) + client.loadSchema(index) match { + case ElasticSuccess(schema) => schema + case ElasticFailure(error) => fail(s"loadSchema($index) failed: ${error.message}") + } + } + + "SET SCHEMA CACHE TTL" should "write the TTL into the index's own metadata" in { + ddl(s"ALTER TABLE $index SET SCHEMA CACHE TTL = '10m'") + SchemaCacheTtl.of(freshSchema) shouldBe Some(Right(600000L)) + } + + it should "survive an unrelated ALTER on the same index" in { + // `Table.update()` rebuilds `_meta`'s five owned keys on every schema; a key it does not own + // must come through untouched, or the TTL would silently revert to the default. + ddl(s"ALTER TABLE $index ADD COLUMN IF NOT EXISTS age INT") + SchemaCacheTtl.of(freshSchema) shouldBe Some(Right(600000L)) + } + + it should "be replaceable" in { + ddl(s"ALTER TABLE $index SET SCHEMA CACHE TTL = '1h'") + SchemaCacheTtl.of(freshSchema) shouldBe Some(Right(3600000L)) + } + + it should "accept the equivalent SET MAPPING spelling it desugars to" in { + ddl(s"ALTER TABLE $index SET MAPPING ${SchemaCacheTtl.MetadataPath} = '30s'") + SchemaCacheTtl.of(freshSchema) shouldBe Some(Right(30000L)) + } + + "DROP SCHEMA CACHE TTL" should "return the index to the client default" in { + // Sets its own precondition rather than inheriting the previous test's: an assertion that a + // value is ABSENT proves nothing unless this test is what removed it. + ddl(s"ALTER TABLE $index SET SCHEMA CACHE TTL = '10m'") + SchemaCacheTtl.of(freshSchema) shouldBe Some(Right(600000L)) + ddl(s"ALTER TABLE $index DROP SCHEMA CACHE TTL") + SchemaCacheTtl.of(freshSchema) shouldBe None + } + + "a TTL that is not a duration" should "be refused before it reaches the cluster" in { + ddl(s"ALTER TABLE $index SET SCHEMA CACHE TTL = '10m'") + client.run(s"ALTER TABLE $index SET SCHEMA CACHE TTL = 'soon'").futureValue match { + case ElasticFailure(error) => + error.message should include("not a duration") + case ElasticSuccess(other) => + fail(s"expected a rejection, got $other") + } + // The rejected statement changed nothing: the previous value still stands. + SchemaCacheTtl.of(freshSchema) shouldBe Some(Right(600000L)) + } +}