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 // ---------------------------------------------------------------------------