From 5dd0a1a9ec4d0616d938643199596118bdb533ed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Manciot?= Date: Wed, 9 Sep 2026 10:50:56 +0200 Subject: [PATCH] fix(21.8 Part F.1): an ALTER must not rewrite processors it was not asked to touch The issue record named one cause. Building the gate it asked for -- "no test asserts the pipeline diff is empty for an unchanged processor" -- found three, each reachable from a different processor kind, so no single statement could exhibit more than one. 1. An anonymous script processor has no column. Processor `description` is an ES 7.9+ field that 6.8 drops (and PipelineApi strips before sending on 6.x), and it is what re-types a script processor on read-back, so keyed by column it could never match the processor that DECLARED it: every ALTER reported it removed and re-added. The identity was in the script all along -- ScriptProcessor writes `ctx. = ...` as the last statement of every source it generates -- so ScriptTarget.of reads back what ScriptTarget.assign wrote, in IngestPipeline.diff's KEY, on both sides. 2. A stored non-textual value was read back as a STRING, so `reputation DOUBLE DEFAULT 0.0` reported ProcessorPropertyChanged(value, 0.0, 0.0) forever -- a diff that prints as no diff. It never depended on `description`, so it was never confined to 6.8. readValue is the typed read IndexField already uses for null_value; Value.unwrap is its inverse on the write side, because Values.value holds WRAPPED elements that serialise as Jackson beans. 3. loadTablePipelineDiff compared what it had read back under the FINAL name against table.defaultPipeline, producing default-pipeline changes drawn from the wrong pipeline. Table.declaredPipeline is exhaustive. Container-valued SET processors are fixed with it: a list or object was lost to the empty string on read-back and is now round-tripped. Proof, not inference: GatewayApiIntegrationSpec requires DdlResult(false) on a repeated no-op ALTER, with DdlResult(true) on the first as a positive control. Reverting the sql fixes reds it on real ES 6.8.23 with the record's own churn line. assertDdl could never have caught this -- a churning ALTER succeeds. Nine mutations, nine predicted REDs. Green on all five clients; extensions verified against a local publish, materialized views green on ES 6/7/8/9. Story: _bmad-output/implementation-artifacts/21-8-temporal-and-boolean-conversion-defects.md Record: docs/issues/local-21.8-alter-churns-unchanged-pipeline-processor.md Co-Authored-By: Claude Opus 5 (1M context) --- .../elastic/client/GatewayApi.scala | 9 +- .../app/softnetwork/elastic/sql/package.scala | 22 + .../elastic/sql/schema/package.scala | 170 ++++++- .../sql/schema/AlterPipelineRenderSpec.scala | 41 +- .../PipelineRoundTripIdentitySpec.scala | 437 ++++++++++++++++++ .../client/GatewayApiIntegrationSpec.scala | 48 +- 6 files changed, 708 insertions(+), 19 deletions(-) create mode 100644 sql/src/test/scala/app/softnetwork/elastic/sql/schema/PipelineRoundTripIdentitySpec.scala diff --git a/core/src/main/scala/app/softnetwork/elastic/client/GatewayApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/GatewayApi.scala index 86c50847d..ad83f879d 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/GatewayApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/GatewayApi.scala @@ -1209,7 +1209,14 @@ class TableExecutor( pipelineType = Some(pipelineType) ) // compute diff for pipeline update - val pipelineDiff: List[PipelineDiff] = pipeline.diff(table.defaultPipeline) + // + // 🔴 21.8 Part F.1 — this read `table.defaultPipeline` for BOTH pipeline types, so what + // had just been read back under the FINAL name was compared against the DEFAULT + // pipeline. Measured: the Final-typed entries the caller applies are the same either + // way, but the old comparison also produced Default-typed `ProcessorAdded` entries that + // `alterExistingIndex` splices in and applies as default-pipeline changes. + val pipelineDiff: List[PipelineDiff] = + pipeline.diff(table.declaredPipeline(pipelineType)) ElasticSuccess(Some(pipelineDiff)) case ElasticSuccess(_) => val error = diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala index f787685b2..b5c06a59e 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala @@ -622,6 +622,28 @@ package object sql { } } + /** The raw Scala value behind a [[Value]], all the way down. + * + * 🔴 21.8 Part F.1. `Value.value` is NOT this for a container: `Values.value` is the + * `Seq[Value[_]]` of WRAPPED elements and `ObjectValue.value` is a `ListMap[String, + * Value[_]]`. Handing either to `mapToJsonNode` serialises each element as a Jackson BEAN — + * `{"value":"a","regex":{…},"expr":{…}}` — so a list-valued ingest processor was written back + * to Elasticsearch in a shape Elasticsearch would store verbatim. MEASURED both ways: from + * `.value` the JSON is that bean; from here it is `["a","b"]`. + * + * Recursive, because one level is not enough: `{"tags":["a","b"],"n":1}` still emitted beans + * for `tags` after the outer map was unwrapped. + * + * `Values.innerValues` is the one-level cousin of this and stays as it is — it is typed + * `Seq[R]`, which is what its callers want, and it cannot express the nested case. + */ + def unwrap(value: Value[_]): Any = value match { + case Null => null + case values: Values[_, _] => values.values.map(unwrap) + case obj: ObjectValue => obj.value.map { case (k, v) => k -> unwrap(v) } + case other => other.value + } + def apply(node: JsonNode): Option[Any] = { node match { case n if n.isNull => Some(null) diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/schema/package.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/schema/package.scala index d3b3f8d78..09d91feb4 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/schema/package.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/schema/package.scala @@ -174,6 +174,38 @@ package object schema { private val ScriptDescRegex = """^\s*([a-zA-Z0-9_\\.]+)\s([a-zA-Z]+)\s+SCRIPT\s+AS\s*\((.*)\)\s*$""".r + /** Read a stored processor value back WITHOUT changing its type. + * + * 🔴 21.8 Part F.1, second half. This was `Value(v.asText())`, which renders every JSON scalar + * as text, so `reputation DOUBLE DEFAULT 0.0` was declared as the number `0.0` and read back + * as the string `"0.0"`. `IngestProcessor.diff` compares `properties`, so that processor + * reported `ProcessorPropertyChanged("value", "0.0", 0.0)` on EVERY ALTER — the same churn as + * the anonymous script processor, and unlike it not confined to Elasticsearch 6.8: it did not + * depend on `description`, so it fired on every version. It also reads identically in the log, + * both sides rendering as `0.0`. + * + * `Value(node)` is the typed read `IndexField` already uses for `null_value`; using it here is + * the same derivation rather than a second one. + * + * Containers are included, and the write side is what makes that safe. `Values.value` and + * `ObjectValue.value` hold WRAPPED elements, so handing either to `mapToJsonNode` serialises + * Jackson BEANS; `SetProcessor.defaultValue` therefore goes through `Value.unwrap`, and an + * array read here is written back as `["a","b"]`. Before this story a container value was lost + * to the EMPTY STRING, because Jackson's `asText()` on a container node is `""`. + * + * 🔴 The `Try` is not defensive padding. `Value.apply(Any)` THROWS + * `IllegalArgumentException("Unsupported Values type")` for a list whose head has no `Values` + * companion — MEASURED on `"value": [null, "a"]`, where the head is `Null` and no arm of the + * sealed hierarchy matches. That read runs inside `PipelineApi`'s pipeline load and inside + * `IndicesApi`'s schema cache, neither of which catches it, so without the guard a value that + * was merely wrong would become an ALTER that dies. Such a value keeps the old text read, i.e. + * exactly today's behaviour. + */ + private def readValue(node: JsonNode): Value[_] = + Value(node) + .flatMap(v => scala.util.Try(Value(v)).toOption) + .getOrElse(Value(node.asText())) + def apply(processorType: IngestProcessorType, properties: ObjectValue): IngestProcessor = { val node = mapper.createObjectNode() node.set(processorType.name, properties.toJson) @@ -229,7 +261,7 @@ package object schema { pipelineType = pipelineType, description = desc, column = field, - value = valueNode.map(v => Value(v.asText())).getOrElse(Null), + value = valueNode.map(readValue).getOrElse(Null), copyFrom = copyFrom, doOverride = doOverride, ignoreEmptyValue = ignoreEmptyValue, @@ -430,12 +462,14 @@ package object schema { * * Collisions: 32 bits over the handful of anonymous processors a pipeline can hold. If two DID * collide the diff would treat them as one processor and compare their properties, so the - * failure mode is a spurious `ProcessorChanged` — noisy, not silent, and no worse than the - * churn this class already has (21.8 Part F.1). + * failure mode is a spurious `ProcessorChanged` — noisy, not silent. * - * This does NOT by itself stop the churn on ES 6.8: a read-back anonymous processor still - * cannot match a declared one keyed by its real column. It makes the churn DETERMINISTIC and - * the emitted DDL VALID. The remaining half is 21.8 Part F.1. + * ⚠️ That reassurance covers THIS fallback only. Once 21.8 Part F.1 keys a script processor by + * the column its source assigns, two scripts writing the SAME column collide, and the diff + * builds its side maps with `toMap`, so one is dropped SILENTLY. That is not new behaviour on + * ES 7+ — `ScriptDescRegex` has always keyed them by their declared column there — so F.1 + * aligns 6.8 with every other version rather than introducing it. It is stated because the + * paragraph above would otherwise read as a guarantee it does not give. */ override def column: String = properties.get("field") match { case Some(s: String) => s @@ -572,6 +606,50 @@ package object schema { case other => other } + /** The ONE place that knows how an ingest script names the column it computes. + * + * A `script` processor has no `field` property, so Elasticsearch stores nothing that says which + * column it feeds. The only carrier was the processor `description` — an Elasticsearch **7.9+** + * field, which 6.8 silently drops — and `IngestProcessor.apply` re-types a script processor from + * it (`ScriptDescRegex`). Without it the read-back is an anonymous `GenericProcessor`, and + * `IngestPipeline.diff` keys processors by `"--"`, so it could never + * match the processor that declared it: every ALTER reported it REMOVED and re-ADDED, on a + * processor nobody had touched (21.8 Part F.1, the churn). + * + * 🔴 The identity was already in the script all along. `assign` is what `ScriptProcessor` emits + * as the LAST statement of every generated source, so `of` reads back exactly what `assign` + * wrote: ONE derivation of "which column does this script feed", used by both sides, rather than + * a second one that can drift from it. `Column.update` relies on the same contract when it + * rewrites the target of a nested column's script. + * + * `of` returns `None` for any source this object did not write — a hand-authored `ALTER PIPELINE + * … ADD PROCESSOR SCRIPT(…)`, say — and the caller keeps its content-addressed fallback, i.e. + * exactly today's behaviour for anything unrecognised. + */ + private[schema] object ScriptTarget { + + /** How an ingest script NAMES a column. `assign` writes an assignment to it; `Column.update` + * rewrites it when a nested column's path changes. Both go through here so the format has one + * spelling — the claim this object's name makes. + */ + def reference(column: String): String = s"ctx.$column" + + def assign(column: String, expression: String): String = s"${reference(column)} = $expression" + + /** The LAST `ctx. = …` in `source`, which is the one `assign` appended. + * + * Greedy on purpose: every statement before it is a `def paramN = …` preamble, and those read + * `ctx.` on the RIGHT of the `=`, never on the left, so anchoring on the assignment and + * taking the last match cannot pick one of them up. + */ + def of(source: String): Option[String] = source match { + case Assignment(column) => Some(column) + case _ => None + } + + private val Assignment = """(?s).*(?:^|;)\s*ctx\.([A-Za-z0-9_.]+)\s*=[^=].*""".r + } + object ScriptProcessor { def fromScript( column: String, @@ -586,10 +664,10 @@ package object schema { val source = painless.split(";") match { case Array(single) if single.trim.startsWith("return ") => val stripped = single.trim.stripPrefix("return ").trim - s"ctx.$column = $stripped" + ScriptTarget.assign(column, stripped) case parts => val last = parts.last.trim - val updated = parts.dropRight(1) :+ s" ctx.$column = $last" + val updated = parts.dropRight(1) :+ s" ${ScriptTarget.assign(column, last)}" updated.mkString(";") } ScriptProcessor( @@ -712,13 +790,20 @@ package object schema { withIf } + /** 🔴 `Value.unwrap`, not `value.value`. For a container the latter is the WRAPPED elements — + * `Seq[Value[_]]` / `ListMap[String, Value[_]]` — and `properties` feeds this straight into + * `mapToJsonNode`, which then serialises each element as a Jackson BEAN. `ALTER PIPELINE` + * rewrites every processor of the merged pipeline, so an untouched list-valued processor would + * be written back to Elasticsearch in that shape. For a scalar `unwrap` IS `value.value`, so + * nothing else moves. + */ lazy val defaultValue: Option[Any] = { if (copyFrom.isDefined) None else value match { case IdValue | IngestTimestampValue => Some(s"{{${value.value}}}") case Null => None - case _ => Some(value.value) + case _ => Some(Value.unwrap(value)) } } @@ -897,7 +982,38 @@ package object schema { val desired = pipeline.processors // 1. Index processors by logical key - def key(p: IngestProcessor) = s"${p.pipelineType.name}-${p.processorType.name}-${p.column}" + // + // 🔴 21.8 Part F.1 — the churn lived in this one expression. A `script` processor is the only + // kind Elasticsearch stores with no `field`, so an anonymous one has no column; processor + // `description`, which `IngestProcessor.apply` re-types it from, is an ES 7.9+ field that 6.8 + // drops and that `PipelineApi` strips before sending on 6.x besides. Keyed by `column`, such + // a processor could never match the one that DECLARED it: every ALTER reported it REMOVED and + // re-ADDED, on a processor nobody had touched. + // + // The identity was in the script all along — `ScriptProcessor` writes `ctx. = …` as + // the last statement of every source it generates — so `ScriptTarget.of` reads back exactly + // what `ScriptTarget.assign` wrote. + // + // 🔴 It is applied HERE and to BOTH sides, rather than inside `GenericProcessor.column`, + // and that placement is the design. `column` is read by `Table.defaultPipeline`'s + // `filterNot`, by `IngestPipeline.merge`, by `ProcessorRemoved.stmt` and by `describe`; + // giving an anonymous processor a real column name there made a hand-added + // `ADD PROCESSOR SCRIPT(ctx.age = 1)` collide with a declared `age SCRIPT AS (…)` column and + // get DROPPED — data loss, to fix a cosmetic diff. Confined to the key, the recovery reaches + // the only consumer that needs it, and because both sides derive it the same way from the + // same string, a source `of` misreads still yields the SAME key on both sides and so cannot + // manufacture a difference. + def identity(p: IngestProcessor): String = + if (p.processorType == IngestProcessorType.Script) + p.properties + .get("source") + .collect { case source: String => source } + .flatMap(ScriptTarget.of) + .getOrElse(p.column) + else p.column + + def key(p: IngestProcessor) = + s"${p.pipelineType.name}-${p.processorType.name}-${identity(p)}" val desiredMap = desired.map(p => key(p) -> p).toMap val actualMap = actual.map(p => key(p) -> p).toMap @@ -1133,7 +1249,8 @@ package object schema { script.map { sc => sc.copy( column = updated.path, - source = sc.source.replace(s"ctx.$name", s"ctx.${updated.path}") + source = + sc.source.replace(ScriptTarget.reference(name), ScriptTarget.reference(updated.path)) ) } updated.copy( @@ -2094,6 +2211,37 @@ package object schema { ) } + /** The pipeline this table DECLARES for `pipelineType`. + * + * 🔴 21.8 Part F.1. `GatewayApi.loadTablePipelineDiff` takes the pipeline type, uses it to + * READ the right pipeline out of Elasticsearch, and then compared what it read against + * `defaultPipeline` for BOTH types. + * + * MEASURED, because the obvious reading of that is wrong in both directions. Filtered to the + * Final-typed entries the caller actually applies, comparing a stored final pipeline against + * `defaultPipeline` and against `finalPipeline` order the SAME `ProcessorRemoved` — so this is + * not, as first recorded, a change that turns churn into a deletion; the deletion is already + * there. What the old comparison ADDS is a set of Default-typed `ProcessorAdded` entries drawn + * from the default pipeline, which `alterExistingIndex` splices into `diff.pipeline` and then + * applies as DEFAULT-pipeline changes. Comparing like with like removes those. + * + * ⚠️ It does not make final-pipeline handling correct. A table never DECLARES a final script + * processor — the parser has no syntax for one, and the `Index -> Table` load attaches a final + * pipeline's `ScriptProcessor` to its COLUMN, which puts it in `tableProcessors` and hence in + * the DEFAULT pipeline — so a stored final script still diffs as removed. Only `rename`, + * `remove` and a non-default `set` survive the load as Final-typed. Fixing that is a change to + * the load path, not to this choice. + * + * Exhaustive on purpose: a catch-all answered `Custom` with the default pipeline, which is the + * same defect one enum value over. + */ + def declaredPipeline(pipelineType: IngestPipelineType): IngestPipeline = + pipelineType match { + case IngestPipelineType.Final => finalPipeline + case IngestPipelineType.Default => defaultPipeline + case IngestPipelineType.Custom => diffPipeline + } + def setDefaultPipelineName(pipelineName: String): Table = { this.copy( settings = this.settings + ("default_pipeline" -> StringValue(pipelineName)) diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/schema/AlterPipelineRenderSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/schema/AlterPipelineRenderSpec.scala index a4f3355c1..a4edfb26f 100644 --- a/sql/src/test/scala/app/softnetwork/elastic/sql/schema/AlterPipelineRenderSpec.scala +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/schema/AlterPipelineRenderSpec.scala @@ -26,6 +26,16 @@ import scala.collection.immutable.ListMap */ class AlterPipelineRenderSpec extends AnyFlatSpec with Matchers { + /** 🔴 The source must be one `ScriptTarget.of` DECLINES, or this whole file stops testing what it + * was written to test. + * + * Story 21.8 Part F.1 keys an anonymous script processor by the column its source assigns, and + * `ctx.a = 1` — the fixture this file originally used — now resolves to `a`. The + * content-addressed fallback below was then never reached, so every assertion here passed + * without exercising it, and 21.5's F8 falsification (restore the UUID, watch three of these + * red) would have scored GREEN. Bracket notation is a form the recovery does not read, so the + * fallback is reachable again. See `PipelineRoundTripIdentitySpec` for the recovery itself. + */ private def anonymous(source: String): GenericProcessor = GenericProcessor( processorType = IngestProcessorType.Script, @@ -37,7 +47,7 @@ class AlterPipelineRenderSpec extends AnyFlatSpec with Matchers { ) "an anonymous processor's column" should "be stable across reads" in { - val p = anonymous("ctx.a = 1") + val p = anonymous("ctx['a'] = 1") // the defect was that these three differ; `ProcessorRemoved` reads it a second time, so a // random value also meant the key and the rendered statement disagreed with each other. p.column shouldBe p.column @@ -45,8 +55,8 @@ class AlterPipelineRenderSpec extends AnyFlatSpec with Matchers { } it should "be a function of the processor's CONTENT, not of the call" in { - anonymous("ctx.a = 1").column shouldBe anonymous("ctx.a = 1").column - anonymous("ctx.a = 1").column should not be anonymous("ctx.b = 2").column + anonymous("ctx['a'] = 1").column shouldBe anonymous("ctx['a'] = 1").column + anonymous("ctx['a'] = 1").column should not be anonymous("ctx['b'] = 2").column } it should "not depend on the order the properties were parsed in" in { @@ -58,7 +68,7 @@ class AlterPipelineRenderSpec extends AnyFlatSpec with Matchers { processorType = IngestProcessorType.Script, properties = ListMap[String, Any]( "lang" -> "painless", - "source" -> "ctx.a = 1", + "source" -> "ctx['a'] = 1", "ignore_failure" -> true ) ) @@ -66,7 +76,7 @@ class AlterPipelineRenderSpec extends AnyFlatSpec with Matchers { processorType = IngestProcessorType.Script, properties = ListMap[String, Any]( "ignore_failure" -> true, - "source" -> "ctx.a = 1", + "source" -> "ctx['a'] = 1", "lang" -> "painless" ) ) @@ -79,7 +89,7 @@ class AlterPipelineRenderSpec extends AnyFlatSpec with Matchers { // was rejected by our own parser with "Mismatched closing parentheses in ALTER PIPELINE // statement" -- a message that names the wrong cause, which is why this asserts the ROUND TRIP // rather than the message. - val stmt = ProcessorRemoved(anonymous("ctx.a = 1")).stmt + val stmt = ProcessorRemoved(anonymous("ctx['a'] = 1")).stmt val sql = AlterPipeline("my_pipeline", ifExists = false, List(stmt)).sql withClue(s"generated [$sql]: ") { sql should include("DROP PROCESSOR") @@ -90,6 +100,25 @@ class AlterPipelineRenderSpec extends AnyFlatSpec with Matchers { } } + "the fixtures in this file" should "actually reach the content-addressed fallback" in { + // The guard for the paragraph above. Without it, a change to the recovery silently empties + // every other assertion in this file and nothing goes red. + anonymous("ctx['a'] = 1").column should startWith("anonymous_") + ScriptTarget.of("ctx['a'] = 1") shouldBe None + } + + "a source the recovery DOES read" should "STILL be content-addressed here" in { + // 🔴 The seam between the two stories, pinned in one place so they cannot drift. + // + // 21.8 Part F.1 recovers the column a script assigns, but it does so in `IngestPipeline.diff`'s + // KEY, not in `column`. `column` is read by `Table.defaultPipeline`'s `filterNot`, by + // `IngestPipeline.merge`, by `ProcessorRemoved.stmt` and by `describe`; giving an anonymous + // processor a real column name there made a hand-added `ADD PROCESSOR SCRIPT(ctx.age = 1)` + // collide with a declared `age SCRIPT AS (…)` column and get DROPPED. This assertion is what + // keeps the recovery out of those four consumers. + anonymous("ctx.a = 1").column should startWith("anonymous_") + } + "a processor that HAS a field" should "still use it, unchanged" in { // The compatibility case. `column` is computed at diff time and NEVER persisted, so no stored // pipeline carries the old value and nothing has to migrate. What an existing installation can diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/schema/PipelineRoundTripIdentitySpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/schema/PipelineRoundTripIdentitySpec.scala new file mode 100644 index 000000000..4affa6faf --- /dev/null +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/schema/PipelineRoundTripIdentitySpec.scala @@ -0,0 +1,437 @@ +package app.softnetwork.elastic.sql.schema + +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.IngestTimestampValue +import app.softnetwork.elastic.sql.query.{CreatePipeline, CreateTable} +import com.fasterxml.jackson.databind.node.ObjectNode +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import scala.jdk.CollectionConverters._ + +/** Story 21.8 Part F.1 — the ALTER pipeline churn. + * + * The production comparison is `GatewayApi.loadTablePipelineDiff`: the pipeline READ BACK out of + * Elasticsearch on one side, the one the DDL DECLARES on the other. For a processor nobody touched + * that diff must be empty, and until this file nothing asserted it — which is why the churn + * survived story 21.5's fix of the other half of the same root cause. + * + * Three independent causes made it non-empty, and only the first was in the issue record. Each is + * reachable from a DIFFERENT processor kind, which is why the assertions below run over a corpus: + * one statement can only ever exhibit one of them. + * + * ==the anonymous script processor — Elasticsearch 6.8 only== + * Processor `description` is an ES 7.9+ field; 6.8 drops it, and `PipelineApi` strips it before + * sending on 6.x besides. `IngestProcessor.apply` re-types a script processor from that + * description (`ScriptDescRegex`), so without it the read-back is an anonymous `GenericProcessor`. + * `IngestPipeline.diff` keys processors by `"--"`, and an anonymous + * processor has no column, so it could never match its declared counterpart: every ALTER reported + * it REMOVED and re-ADDED. + * + * ==the retyped stored value — EVERY version== + * The read was `Value(v.asText())`, so `reputation DOUBLE DEFAULT 0.0` was declared as the number + * `0.0` and read back as the string `"0.0"`: a permanent `ProcessorChanged` that does not depend + * on `description` at all. Both sides render as `0.0` in the warning the operator sees, so it read + * as a diff that is not a diff, and that — not rarity — is why it survived every version run. + * + * ==the final pipeline compared against the default one — RECORDED, NOT FIXED== + * `loadTablePipelineDiff` takes the pipeline type, uses it to read the right pipeline out of + * Elasticsearch, and then compares it against `table.defaultPipeline` regardless. 🔴 The obvious + * fix is MEASURABLY WORSE and was reverted: a table never DECLARES a final processor, so diffing + * against `table.finalPipeline` turns churn into an order to DELETE processors nobody touched. The + * test below is that measurement. + */ +class PipelineRoundTripIdentitySpec extends AnyFlatSpec with Matchers { + + /** Elasticsearch 6.8 stores a pipeline WITHOUT the processor descriptions it was given. */ + private def asStoredOnEs68(pipeline: IngestPipeline): String = { + val node = pipeline.node + node.get("processors").elements().asScala.foreach { p => + val processorType = p.fieldNames().next() + p.get(processorType).asInstanceOf[ObjectNode].remove("description") + () + } + mapper.writeValueAsString(node) + } + + /** Elasticsearch 7.9+ stores it whole. */ + private def asStoredOnEs79(pipeline: IngestPipeline): String = pipeline.json + + private def declared(sql: String): IngestPipeline = + Parser(sql) match { + case Right(ct: CreateTable) => ct.schema.defaultPipeline + case Right(cp: CreatePipeline) => cp.ddlPipeline + case other => fail(s"[$sql] expected a pipeline-bearing DDL, got $other") + } + + private def readBack( + declaredPipeline: IngestPipeline, + stored: String, + as: Option[IngestPipelineType] = None + ): IngestPipeline = + IngestPipeline( + name = declaredPipeline.name, + json = stored, + pipelineType = Some(as.getOrElse(declaredPipeline.pipelineType)) + ) + + // -- the corpus ------------------------------------------------------------------------------ + // Between them these cover every processor kind a CREATE TABLE deploys: `script` (top level and + // nested under a STRUCT, whose target is a dotted PATH), `set` as a textual / numeric / boolean / + // `_ingest.timestamp` default, `date_index_name` from PARTITION BY, and the `_id` `set` from + // PRIMARY KEY. `rename`, `remove` and `enrich` reach a pipeline only through CREATE PIPELINE and + // are covered by `handWritten` below, which is asserted separately because its declared + // processors carry a pipeline type its own pipeline does not (see the divergence recorded after + // it). + + private val everyProcessorKind = + """CREATE TABLE users ( + | id INT NOT NULL, + | name VARCHAR DEFAULT 'anonymous', + | birthdate DATE, + | age INT SCRIPT AS (DATE_DIFF(birthdate, CURRENT_DATE, YEAR)), + | ingested_at TIMESTAMP DEFAULT _ingest.timestamp, + | profile STRUCT FIELDS( + | join_date DATE, + | seniority INT SCRIPT AS (DATE_DIFF(profile.join_date, CURRENT_DATE, DAY)) + | ), + | PRIMARY KEY (id) + |) PARTITION BY birthdate (MONTH)""".stripMargin + + private val numericDefault = + "CREATE TABLE users (id KEYWORD, reputation DOUBLE DEFAULT 0.0)" + + private val integerDefault = + "CREATE TABLE users (id KEYWORD, visits INT DEFAULT 0)" + + private val booleanDefault = + "CREATE TABLE users (id KEYWORD, active BOOLEAN DEFAULT true)" + + private val scriptOnly = + """CREATE TABLE users ( + | join_date DATE, + | seniority INT SCRIPT AS (DATE_DIFF(join_date, CURRENT_DATE, DAY)) + |)""".stripMargin + + private val handWritten = + """CREATE OR REPLACE PIPELINE user_pipeline WITH PROCESSORS ( + | SET ( + | field = "name", + | if = "ctx.name == null", + | ignore_failure = true, + | value = "anonymous" + | ), + | SCRIPT ( + | lang = "painless", + | source = "def param1 = ctx.birthdate; ctx.age = (param1 == null) ? null : 1", + | ignore_failure = true + | ), + | RENAME ( + | field = "old_name", + | target_field = "new_name", + | ignore_failure = true + | ), + | REMOVE ( + | field = "obsolete", + | ignore_failure = true + | ), + | ENRICH ( + | field = "zip", + | policy_name = "zip_policy", + | target_field = "location", + | max_matches = 1, + | ignore_failure = true + | ) + |)""".stripMargin + + private val corpus: Seq[(String, String)] = Seq( + "every processor kind" -> everyProcessorKind, + "a numeric default" -> numericDefault, + "an integer default" -> integerDefault, + "a boolean default" -> booleanDefault, + "a computed column" -> scriptOnly, + // The value read is now TYPE-sensitive, so the shapes where a JSON scalar and a SQL literal + // could disagree are guarded rather than assumed: an integer literal on a floating column, a + // magnitude past what a Double holds exactly, and a scale that renders differently. + "an integer literal on a DOUBLE column" -> + "CREATE TABLE users (id KEYWORD, x DOUBLE DEFAULT 0)", + "a value beyond 2^53" -> + "CREATE TABLE users (id KEYWORD, x BIGINT DEFAULT 9007199254740993)", + "a trailing-zero scale" -> + "CREATE TABLE users (id KEYWORD, x DOUBLE DEFAULT 0.10)" + ) + + // -- the property ---------------------------------------------------------------------------- + + for ((label, sql) <- corpus) { + s"a pipeline declaring $label" should "diff EMPTY after an Elasticsearch 6.8 round trip" in { + val desired = declared(sql) + val actual = readBack(desired, asStoredOnEs68(desired)) + withClue(s"read back as ${actual.processors.map(p => p.processorType.name -> p.column)}: ") { + actual.diff(desired) shouldBe Nil + } + } + + it should "diff EMPTY after an Elasticsearch 7.9+ round trip" in { + val desired = declared(sql) + val actual = readBack(desired, asStoredOnEs79(desired)) + withClue(s"read back as ${actual.processors.map(p => p.processorType.name -> p.column)}: ") { + actual.diff(desired) shouldBe Nil + } + } + } + + "a pipeline declared directly, with hand-written processors" should "diff EMPTY on both versions" in { + // The other DDL surface. Its `source` was NOT written by `ScriptProcessor`, so the script + // target is recovered from a script this codebase did not emit. + // + // Read back under the pipeline type its processors DECLARE -- see the divergence recorded + // immediately below, which this case deliberately holds still so it tests one thing. + val desired = declared(handWritten) + val asDeclared = Some(IngestPipelineType.Default) + readBack(desired, asStoredOnEs68(desired), asDeclared).diff(desired) shouldBe Nil + readBack(desired, asStoredOnEs79(desired), asDeclared).diff(desired) shouldBe Nil + + // 🔴 The empty diff ALONE proves little here: both sides are anonymous processors over + // IDENTICAL content, so any deterministic identity satisfies it. What matters is WHERE the + // recovery lives -- these processors keep an anonymous `column`, because the recovery is + // confined to the diff key and must not reach `Table.defaultPipeline`'s `filterNot`, + // `IngestPipeline.merge`, `ProcessorRemoved.stmt` or `describe`. + desired.processors + .filter(_.processorType == IngestProcessorType.Script) + .map(_.column) + .foreach(_ should startWith("anonymous_")) + } + + "a CUSTOM pipeline read back under its OWN name" should + "diverge on `pipelineType` -- RECORDED, not fixed here" in { + // 🔴 A THIRD divergence of the same class, found by this file and deliberately left standing. + // + // `IngestPipeline(name, json, …)` stamps the PIPELINE's type onto every processor it parses; + // the case-class constructor `IngestPipeline(name, type, processors)` does not, and the DDL + // parse hard-codes `Default` because the pipeline type is not known yet when a processor is + // read. So a custom pipeline's processors say `DEFAULT` on the declared side and `CUSTOM` on + // the read-back side, and `pipelineType` is part of `IngestPipeline.diff`'s key: EVERY + // processor comes back added AND removed. + // + // It is not reached by shipped code -- `PipelineApi` merges custom pipelines rather than + // diffing them, and `GatewayApi` only ever diffs the default and final ones, both of which are + // consistent on both sides once the FINAL-vs-DEFAULT comparison is fixed. Closing it means + // re-typing the declared processors through the read-back derivation, which changes the JSON + // `PipelineApi` sends to Elasticsearch; that is a lead call, not a drive-by. + // + // This assertion is the record. It FAILS the day someone fixes it, which is the point. + // 🔴 Pinned on the CAUSE — the two pipeline-type stamps — and NOT on the size of the resulting + // diff. A magnitude assertion inside a file whose thesis is that the round trip IS an identity + // would certify both, and the cheapest way past it for a future fixer is to edit the number. + // Naming the stamps means the day they agree, this test says so. + val desired = declared(handWritten) + val actual = IngestPipeline(name = desired.name, json = asStoredOnEs79(desired)) + desired.processors.map(_.pipelineType.name).distinct shouldBe List("DEFAULT") + actual.processors.map(_.pipelineType.name).distinct shouldBe List("CUSTOM") + actual.diff(desired) should not be Nil + } + + // -- the counter-property: an unchanged processor is not bought by matching everything --------- + + "a script processor whose expression GENUINELY changed" should "be reported as CHANGED" in { + // 🔴 The SHAPE is the assertion. `should not be Nil` is satisfied by the DEFECT too -- the + // churn made every diff non-empty -- so it proves nothing. Recovering the identity is what + // turns a remove-plus-add into a change, and only naming `ProcessorChanged` tells them apart. + val desired = declared( + """CREATE TABLE users ( + | join_date DATE, + | seniority INT SCRIPT AS (DATE_DIFF(join_date, CURRENT_DATE, MONTH)) + |)""".stripMargin + ) + val actual = readBack(declared(scriptOnly), asStoredOnEs68(declared(scriptOnly))) + actual.diff(desired).map(_.getClass.getSimpleName) shouldBe List("ProcessorChanged") + } + + "a default value that GENUINELY changed" should "be reported as CHANGED" in { + val desired = declared("CREATE TABLE users (id KEYWORD, reputation DOUBLE DEFAULT 1.0)") + val actual = readBack(declared(numericDefault), asStoredOnEs79(declared(numericDefault))) + actual.diff(desired).map(_.getClass.getSimpleName) shouldBe List("ProcessorChanged") + } + + "a computed column that was DROPPED" should "still be reported as removed" in { + val desired = declared("CREATE TABLE users (join_date DATE)") + val actual = readBack(declared(scriptOnly), asStoredOnEs68(declared(scriptOnly))) + actual.diff(desired).map(_.getClass.getSimpleName) shouldBe List("ProcessorRemoved") + } + + // -- which pipeline a diff is against --------------------------------------------------------- + + "the pipeline a table declares for a type" should "be the FINAL one for FINAL" in { + // 🔴 This assertion replaced one that recorded the opposite, and the correction is the point. + // + // `GatewayApi.loadTablePipelineDiff` compared what it had read back under the FINAL name + // against `table.defaultPipeline`. The first reading of that was "the obvious fix is worse, + // because a table declares no final processor, so diffing against the empty `finalPipeline` + // orders a DELETE". Tracing it through `TableDiff.finalPipeline` -- which is what the caller + // actually applies -- refutes it: BOTH comparisons order the same `ProcessorRemoved`. The + // deletion is already there. What the old comparison adds is Default-typed `ProcessorAdded` + // entries that get spliced in and applied to the DEFAULT pipeline. + // + // A diff has to be read through the filter its caller applies. Read raw, it says the opposite. + val table = Parser(scriptOnly) match { + case Right(ct: CreateTable) => ct.schema + case other => fail(s"expected a CreateTable, got $other") + } + val storedFinal = IngestPipeline( + name = "final-users", + pipelineType = IngestPipelineType.Final, + processors = Seq( + table.columns + .find(_.name == "seniority") + .flatMap(_.script) + .getOrElse(fail("no script processor for seniority")) + .copy(pipelineType = IngestPipelineType.Final) + ) + ) + + def finalEntries(d: IngestPipeline): List[String] = + storedFinal + .diff(d) + .filter(_.pipelineType == IngestPipelineType.Final) + .map(_.getClass.getSimpleName) + + // the same order either way -- the claim that the fix introduces a deletion is false + finalEntries(table.defaultPipeline) shouldBe List("ProcessorRemoved") + finalEntries(table.declaredPipeline(IngestPipelineType.Final)) shouldBe List("ProcessorRemoved") + + // what actually differs: comparing like with like stops producing DEFAULT-typed additions + storedFinal + .diff(table.defaultPipeline) + .count( + _.pipelineType == IngestPipelineType.Default + ) should be > 0 + storedFinal + .diff(table.declaredPipeline(IngestPipelineType.Final)) + .count(_.pipelineType == IngestPipelineType.Default) shouldBe 0 + + // and the selection is exhaustive, so no type silently answers with the default pipeline + table.declaredPipeline(IngestPipelineType.Default) shouldBe table.defaultPipeline + table.declaredPipeline(IngestPipelineType.Custom) should not be table.defaultPipeline + } + + "a stored final SCRIPT processor" should + "still diff as removed -- RECORDED, it needs the load path, not this choice" in { + // The residual, stated so it is not mistaken for something this story closed. A table never + // DECLARES a final script processor: the parser has no syntax for one, and the + // `Index -> Table` load attaches a final pipeline's `ScriptProcessor` to its COLUMN, which puts + // it in `tableProcessors` and so in the DEFAULT pipeline. Only `rename`, `remove` and a + // non-default `set` survive the load as Final-typed. + val table = Parser(scriptOnly) match { + case Right(ct: CreateTable) => ct.schema + case other => fail(s"expected a CreateTable, got $other") + } + table.processors shouldBe empty + table.declaredPipeline(IngestPipelineType.Final).processors shouldBe empty + } + + // -- a stored value must survive its own round trip, container or not ------------------------ + + private def storedSet(value: String): IngestPipeline = + IngestPipeline( + name = "p", + json = s"""{"processors":[{"set":{"field":"tags","value":$value}}]}""", + pipelineType = Some(IngestPipelineType.Default) + ) + + /** What the next `ALTER PIPELINE` would write back to Elasticsearch for that processor. */ + private def writtenBack(pipeline: IngestPipeline): String = + mapper.writeValueAsString(pipeline.processors.head.node.get("set").get("value")) + + "a stored LIST value" should "be written back as the same list" in { + // 🔴 Two defects met here, and the second is why the first could not simply be waved through. + // + // The read was `Value(node.asText())`, and Jackson's `asText()` on a container node is the + // EMPTY STRING, so a list-valued `set` processor came back as `""` and was rewritten as `""`. + // + // Reading it properly is not enough: `Values.value` is the WRAPPED `Seq[Value[_]]`, and + // `properties` feeds that straight to `mapToJsonNode`, which serialises each element as a + // Jackson BEAN — `{"value":"a","regex":{…},"expr":{…}}`. `ALTER PIPELINE` rewrites every + // processor of the merged pipeline, so an UNTOUCHED processor would have been written back to + // the cluster in that shape. `Value.unwrap` is the other half of the fix. + writtenBack(storedSet("""["a","b"]""")) shouldBe """["a","b"]""" + } + + it should "be written back as the same list for every scalar element type" in { + writtenBack(storedSet("[1,2]")) shouldBe "[1,2]" + writtenBack(storedSet("[1.5,2.5]")) shouldBe "[1.5,2.5]" + writtenBack(storedSet("[true,false]")) shouldBe "[true,false]" + } + + "a stored OBJECT value" should "be written back as the same object" in { + writtenBack(storedSet("""{"lat":48.8,"lon":2.3}""")) shouldBe """{"lat":48.8,"lon":2.3}""" + } + + it should "survive NESTING, which one level of unwrapping does not" in { + // 🔴 Measured: unwrapping only the outer map left `tags` as beans. The unwrap has to recurse. + writtenBack(storedSet("""{"tags":["a","b"],"n":1}""")) shouldBe """{"tags":["a","b"],"n":1}""" + } + + "a stored value the typed read cannot type" should "fall back, never throw" in { + // 🔴 A list whose head is `Null` has no arm in the sealed `Values` hierarchy, so + // `Value.apply(Any)` THROWS — and this read runs inside `PipelineApi`'s pipeline load and + // `IndicesApi`'s schema cache, neither of which catches it. Reading a value more precisely must + // not turn a wrong ALTER into one that dies; such a value keeps the old text read. + val pipeline = storedSet("""[null,"a"]""") + pipeline.processors.map(_.column) shouldBe List("tags") + writtenBack(pipeline) shouldBe "\"\"" // the JSON empty string, i.e. today's text read + } + + "a container-valued processor" should "diff EMPTY against itself across the round trip" in { + // The property this file exists for, applied to the shapes the corpus above cannot reach: a + // container value can only enter a pipeline through CREATE PIPELINE or an externally authored + // one, never through a CREATE TABLE default. + Seq("""["a","b"]""", "[1,2]", """{"lat":48.8,"lon":2.3}""", """{"tags":["a","b"],"n":1}""") + .foreach { value => + val declared = storedSet(value) + val actual = readBack(declared, asStoredOnEs79(declared)) + withClue(s"value $value: ") { actual.diff(declared) shouldBe Nil } + } + } + + // -- the identity itself ---------------------------------------------------------------------- + + "the script target" should "round-trip through the one derivation that writes it" in { + ScriptTarget.of(ScriptTarget.assign("age", "1")) shouldBe Some("age") + ScriptTarget.of(ScriptTarget.assign("profile.seniority", "param2")) shouldBe + Some("profile.seniority") + } + + it should "read the LAST assignment, which is the one `assign` appends" in { + // 🔴 Added because a falsification pass found this untested: switching the match from greedy to + // lazy left every other case green. A generated preamble holds `def paramN = …` only, so the + // sources this codebase writes have exactly one assignment and cannot tell the two apart. A + // hand-written SCRIPT processor can have several, and the target is the last one — that is + // where `assign` puts it. + ScriptTarget.of("ctx.tmp = 1; ctx.age = 2") shouldBe Some("age") + ScriptTarget.of("def p = ctx.a; ctx.tmp = p; ctx.profile.seniority = p") shouldBe + Some("profile.seniority") + } + + it should "read the ASSIGNMENT, not a `ctx.` reference in the preamble" in { + // Every generated preamble statement mentions `ctx.` on the RIGHT of its `=`. Picking one + // of those up would key the processor by an INPUT column instead of the output one -- which + // still matches nothing, and does it silently. + ScriptTarget.of( + "def param1 = ctx.join_date; def param2 = ctx.hired_date; ctx.seniority = param1" + ) shouldBe Some("seniority") + } + + it should "read a source whose statements are newline-separated" in { + // Painless requires the `;`, so the separator is `;\n` rather than a bare newline. `(?s)` plus + // `\s*` already covers it; asserted because a multi-line operator-authored pipeline is the + // realistic hand-written shape and nothing else in this file has one. + ScriptTarget.of("def p = ctx.a;\nctx.age = p") shouldBe Some("age") + } + + it should "decline a source it did not write, so the caller keeps its fallback" in { + ScriptTarget.of("ctx['age'] = 1") shouldBe None + ScriptTarget.of("def x = 1") shouldBe None + ScriptTarget.of("if (ctx.age == null) { return }") shouldBe None + } +} diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/GatewayApiIntegrationSpec.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/GatewayApiIntegrationSpec.scala index 1ef128a68..96477cd2a 100644 --- a/testkit/src/main/scala/app/softnetwork/elastic/client/GatewayApiIntegrationSpec.scala +++ b/testkit/src/main/scala/app/softnetwork/elastic/client/GatewayApiIntegrationSpec.scala @@ -16,7 +16,7 @@ package app.softnetwork.elastic.client -import app.softnetwork.elastic.client.result.{DmlResult, ElasticSuccess} +import app.softnetwork.elastic.client.result.{DdlResult, DmlResult, ElasticSuccess} import app.softnetwork.elastic.scalatest.ElasticTestKit import app.softnetwork.elastic.sql.{DoubleValue, IdValue} import app.softnetwork.elastic.sql.`type`.SQLTypes @@ -414,6 +414,52 @@ trait GatewayApiIntegrationSpec extends GatewayIntegrationTestKit { } } + it should "report NO changes when an ALTER leaves every processor untouched" in { + // 🔴 Story 21.8 Part F.1 — the ALTER pipeline churn, end to end. + // + // Nothing gated this: `assertDdl` only checks that the statement SUCCEEDED, and a churning + // ALTER succeeds. `DdlResult(false)` is the engine saying "no changes detected", and it can + // only say it when the pipeline diff is EMPTY — so re-issuing an ALTER that changes nothing is + // the one assertion that observes the churn from outside. + // + // The table carries both causes on purpose: + // - `seniority … SCRIPT AS (…)` — a script processor, which Elasticsearch stores with no + // `field`. On 6.8 the `description` that identified it is dropped, so it came back + // anonymous and could never match the column that declared it. 6.8 ONLY. + // - `reputation DOUBLE DEFAULT 0.0` — a non-textual stored value, which was read back as the + // STRING "0.0". EVERY version, and it renders as `0.0` on both sides of the warning. + val create = + """CREATE TABLE IF NOT EXISTS users_alter_churn ( + | id INT NOT NULL, + | join_date DATE, + | reputation DOUBLE DEFAULT 0.0, + | seniority INT SCRIPT AS (DATEDIFF(join_date, CURRENT_DATE, DAY)) + |);""".stripMargin + + assertDdl(System.nanoTime(), client.run(create).futureValue) + + val alter = + "ALTER TABLE users_alter_churn ADD COLUMN IF NOT EXISTS nickname VARCHAR;" + + // 🔴 The positive control. `assertDdl` alone would let this test pass VACUOUSLY: an engine that + // answered "no changes" to EVERY alter would satisfy the assertion below without ever having + // created the processors under test, and so would a stale index left by an earlier run, since + // both statements are IF [NOT] EXISTS. Requiring `true` here means the second call's `false` is + // a real transition. + val firstAlter = client.run(alter).futureValue + renderResults(System.nanoTime(), firstAlter) + firstAlter.isSuccess shouldBe true + firstAlter.toOption.get shouldBe DdlResult(true) + + // second time: nothing to do. Before the fix the untouched script processor (6.8) and the + // untouched numeric default (every version) each reported themselves as changed, so the engine + // rewrote the pipeline and answered `DdlResult(true)`. + val res = client.run(alter).futureValue + renderResults(System.nanoTime(), res) + res.isSuccess shouldBe true + res.toOption.get shouldBe DdlResult(false) + } + // --------------------------------------------------------------------------- // ALTER TABLE — NOT NULL // ---------------------------------------------------------------------------