You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sz...@apache.org on 2026/08/11 21:42:04 UTC
(spark) branch branch-4.x updated: [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites
This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 8a5c2b777586 [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites
8a5c2b777586 is described below
commit 8a5c2b777586f52c517e57921ac5532a5daf8e2a
Author: Andreas Neumann <an...@apache.org>
AuthorDate: Tue Aug 11 14:41:41 2026 -0700
[SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites
### What changes were proposed in this pull request?
This adds end-to-end (graph-execution) test coverage for AutoCDC **SCD Type 2** flows, bringing it up to parity with the existing SCD Type 1 suites. SCD2 was wired end-to-end in SPARK-58321 but had only two E2E suites (single-pipeline and auxiliary-table-spec), whereas SCD1 has a broad set. This PR adds six SCD2 suites mirroring their SCD1 analogs:
| Suite | Tests | Covers |
|---|---|---|
| `AutoCdcScd2FullRefreshSuite` | 3 | Full refresh wipes the target and the (richer) SCD2 auxiliary table; selective refresh leaves non-refreshed targets intact. |
| `AutoCdcScd2TargetTableDurabilitySuite` | 5 | Interop with a hand-populated target: pre-loaded open records closed/opened by later upserts, lower-sequence weaving, closed-record bisection, lazy aux-table creation, framework-column auto-add. |
| `AutoCdcScd2MultiPipelineSuite` | 5 | Independent target/aux tables across pipelines, a downstream MV reading an SCD2 target, a shared target written by two pipelines, cross-pipeline schema evolution, and `KEY_SCHEMA_DRIFT`. |
| `AutoCdcScd2AuxiliaryTableDurabilitySuite` | 8 | Recorded per-key history persists across runs; dry run does not provision the aux table; a dropped aux table is transparently recreated. |
| `AutoCdcScd2KeyDriftSuite` | 14 | Key add/drop/swap, dataType drift, order/backtick/nullability invariance, case-sensitive vs case-insensitive resolution, and tampered-aux failure modes. |
| `AutoCdcScd2SchemaEvolutionSuite` | 10 | Nullable-column merge, new top-level column, additive target-column evolution, broadening the column selection, a new field inside an `array<struct>`, incompatible type changes, track-history drift rejection under default tracking, and a case-only source-column rename folding onto the target. |
These are test-only additions; there is no production code change.
Where SCD2 behaves differently from SCD1, the tests assert (and document) the SCD2 semantics rather than assuming parity:
- an SCD2 upsert to an existing key does not overwrite the row — it closes the prior record and opens a new one — so the assertions carry the full interval history;
- the SCD2 auxiliary table mirrors the full target row schema (plus the aux-only deleted-by-batch-id marker), so non-key additive evolution also extends the aux table, unlike SCD1's keys-only aux.
- the SCD2 track-history column set is recorded on the aux table and drift-checked across runs (SPARK-58391): because it is selection-derived, additive evolution under default tracking widens it and is rejected as `AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT`, so the additive-evolution tests pin `trackHistorySelection` and a dedicated test asserts the rejection.
The narrowing / dropped-column cases (a microbatch narrower than the evolved target, including dropped nested struct/array fields) are intentionally **not** duplicated here — they are owned by `AutoCdcScd2ColumnEvolutionSuite` under SPARK-58418, which makes them reconcile correctly.
### Why are the changes needed?
SCD2 AutoCDC shipped end-to-end (SPARK-58321) with much thinner test coverage than SCD1: only single-pipeline and aux-spec suites, versus SCD1's full-refresh, durability, multi-pipeline, key-drift, and schema-evolution suites. That left full-refresh semantics, cross-pipeline behavior, key-drift validation, auxiliary-table durability, and schema evolution unverified for SCD2. These suites close that gap and lock in the SCD2 semantics against regression.
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
This *is* tests. All six suites pass locally (45 tests total), alongside the existing SCD1 and SCD2 suites, plus scalastyle on the module.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57689 from anew/spark-58409-scd2-e2e-test-suites.
Authored-by: Andreas Neumann <an...@apache.org>
Signed-off-by: Szehon Ho <sz...@gmail.com>
(cherry picked from commit 297a730f57edeea2a652604441310b5b65e85613)
Signed-off-by: Szehon Ho <sz...@gmail.com>
---
.../AutoCdcScd2AuxiliaryTableDurabilitySuite.scala | 376 +++++++++++++++
.../graph/AutoCdcScd2FullRefreshSuite.scala | 263 +++++++++++
.../pipelines/graph/AutoCdcScd2KeyDriftSuite.scala | 487 +++++++++++++++++++
.../graph/AutoCdcScd2MultiPipelineSuite.scala | 302 ++++++++++++
.../graph/AutoCdcScd2SchemaEvolutionSuite.scala | 526 +++++++++++++++++++++
.../AutoCdcScd2TargetTableDurabilitySuite.scala | 285 +++++++++++
6 files changed, 2239 insertions(+)
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala
new file mode 100644
index 000000000000..e6b1119383ce
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala
@@ -0,0 +1,376 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ ColumnSelection,
+ Scd2BatchProcessor,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests covering the durability of the SCD Type 2 AutoCDC auxiliary table across pipeline runs:
+ * the per-key history recorded in the auxiliary table must persist between incremental runs, and
+ * the auxiliary table must be transparently recreated if it is deleted out-of-band. The SCD2
+ * analog of [[AutoCdcScd1AuxiliaryTableDurabilitySuite]]. Unlike SCD1, the SCD2 auxiliary table's
+ * schema is the full target row schema plus the aux-only deleted-by-batch-id marker, so the
+ * schema-layout assertions differ accordingly.
+ */
+class AutoCdcScd2AuxiliaryTableDurabilitySuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ import testImplicits._
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ test("a higher-sequence event in a later pipeline run correctly closes and opens records") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Single MemoryStream reused across both pipeline runs so the streaming checkpoint can
+ // resume cleanly.
+ val changeDataFeedStream = MemoryStream[(Int, String, Long)]
+ def buildGraphRegistrationContext(): TestGraphRegistrationContext =
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = changeDataFeedStream.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ // Run #1: insert id=1 at seq=1.
+ changeDataFeedStream.addData((1, "alice", 1L))
+ runPipeline(buildGraphRegistrationContext())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, 1L, null, scd2Meta(1L)))
+ )
+
+ // Run #2: upsert id=1 at seq=2 (closes the seq=1 record, opens a new one) and insert id=2 at
+ // seq=1 (new key). The auxiliary table from run #1 persists and supplies the prior history.
+ changeDataFeedStream.addData((1, "alice2", 2L), (2, "bob", 1L))
+ runPipeline(buildGraphRegistrationContext())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 1L, 1L, 2L, scd2Meta(1L)),
+ Row(1, "alice2", 2L, 2L, null, scd2Meta(2L)),
+ Row(2, "bob", 1L, 1L, null, scd2Meta(1L))
+ )
+ )
+ }
+
+ test("an event with a sequence lower than what was applied in a prior pipeline run " +
+ "is woven in as a closed prior record") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Single MemoryStream reused across both runs so the streaming checkpoint can resume.
+ val stream = MemoryStream[(Int, String, Long)]
+ def buildCtx(): TestGraphRegistrationContext =
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ // Run #1: upsert id=1 at seq=10. Auxiliary table records the open record at seq=10.
+ stream.addData((1, "alice", 10L))
+ runPipeline(buildCtx())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 10L, 10L, null, scd2Meta(10L)))
+ )
+
+ // Run #2: late upsert at seq=5 (< the persisted seq=10). Unlike SCD1, SCD2 does not suppress
+ // it: the aux history lets reconciliation weave it in as a closed prior record ending at 10,
+ // while the seq=10 record stays open.
+ stream.addData((1, "early", 5L))
+ runPipeline(buildCtx())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "early", 5L, 5L, 10L, scd2Meta(5L)),
+ Row(1, "alice", 10L, 10L, null, scd2Meta(10L))
+ )
+ )
+ }
+
+ test("the SCD2 auxiliary table schema is the full target row schema plus the " +
+ "deleted-by-batch-id marker, and records the key columns property") {
+ // Source DF column order is (name, id, version): the AutoCDC key column `id` does NOT appear
+ // first in the source DF. The SCD2 auxiliary table mirrors the full target row schema (all
+ // user + framework columns) with the aux-only deleted-by-batch-id marker appended, and records
+ // the key columns in the key-column-names property.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(name STRING, id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(String, Int, Long)]
+ stream.addData(("alice", 1, 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("name", "id", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2))
+
+ val targetSchema = spark.table(s"$catalog.$namespace.target").schema.fieldNames.toSeq
+ val auxSchema = spark.table(auxTableNameFor("target")).schema.fieldNames.toSeq
+ // The aux schema is the full target row schema with the marker appended.
+ assert(auxSchema == targetSchema :+ Scd2BatchProcessor.deletedByBatchIdColName)
+ assert(getAuxTableKeyColumnNames(target = "target") == Seq("id"))
+ }
+
+ test("the auxiliary table preserves the user's declared key order in the key-columns " +
+ "property, independent of the source DataFrame and target table column orders") {
+ // The user declares `keys = Seq("region", "id")` -- the OPPOSITE order from how those columns
+ // appear in both the source DF and the target. The recorded key-column-names property should
+ // honor the user's declared key order so subsequent runs compare keys against the same layout.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(value STRING, id INT NOT NULL, region STRING NOT NULL, " +
+ s"version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(String, Int, String, Long)]
+ stream.addData(("v", 1, "us", 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("value", "id", "region", "version"),
+ keys = Seq("region", "id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2))
+
+ assert(getAuxTableKeyColumnNames(target = "target") == Seq("region", "id"))
+ }
+
+ test("a dry run resolves and validates the graph without provisioning the auxiliary " +
+ "table") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 1L))
+ val ctx = singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("id", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ val updateCtx = TestPipelineUpdateContext(spark, ctx.toDataflowGraph, storageRoot)
+ updateCtx.pipelineExecution.dryRunPipeline()
+
+ assert(!spark.catalog.tableExists(auxTableNameFor("target")))
+ }
+
+ test("if the SCD2 AutoCDC auxiliary table is dropped between runs, it is transparently " +
+ "recreated") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Single MemoryStream reused across both runs so the streaming checkpoint can resume.
+ val stream = MemoryStream[(Int, Long)]
+ def buildCtx(): TestGraphRegistrationContext =
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("id", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ stream.addData((1, 1L))
+ runPipeline(buildCtx())
+ assert(spark.catalog.tableExists(auxTableNameFor("target")))
+
+ // Manually drop the auxiliary table.
+ spark.sql(s"DROP TABLE ${auxTableNameFor("target")}")
+ assert(!spark.catalog.tableExists(auxTableNameFor("target")))
+
+ stream.addData((1, 2L))
+ runPipeline(buildCtx())
+
+ // The dropped auxiliary table must be transparently recreated. Here the seq=1 record also
+ // lives in the target as a visible row, and SCD2 reconciliation reads affected rows from the
+ // target as well as the aux table, so this particular history survives the drop: the seq=2
+ // event still closes the seq=1 record and opens a new one. (This is NOT a general guarantee
+ // that the aux table is disposable -- state the aux holds that is NOT mirrored in the target,
+ // e.g. a tombstone from a delete-only run, is lost on a drop; see the aux-sole-holder test
+ // below.)
+ assert(spark.catalog.tableExists(auxTableNameFor("target")))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, 1L, 1L, 2L, scd2Meta(1L)),
+ Row(1, 2L, 2L, null, scd2Meta(2L))
+ )
+ )
+ }
+
+ test("the auxiliary table durably holds state absent from the target: a tombstone from a " +
+ "delete-only run closes a later lower-sequence upsert") {
+ // Unlike the transparently-recreated test above (where the surviving state also lived in the
+ // target as a visible row), here the auxiliary table is the SOLE holder of the state. A
+ // delete-only first run leaves the target empty but records a tombstone at seq=10 in the aux;
+ // the durability of THAT aux-only row is what lets a later, lower-sequence upsert land as a
+ // closed prior record. Drop the aux and this history is gone (the upsert would instead open a
+ // current record) -- which is exactly why the aux is not disposable.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Single MemoryStream reused across both runs so the streaming checkpoint can resume.
+ val stream = MemoryStream[(Int, String, Long, Boolean)]
+ def buildCtx(): TestGraphRegistrationContext =
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("id", "name", "version", "is_delete"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ deleteCondition = Some(functions.col("is_delete") === true),
+ columnSelection = Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("is_delete"))
+ )),
+ scdType = ScdType.Type2)
+
+ // Run #1: a delete at seq=10. The target stays empty; the aux records a tombstone at seq=10.
+ stream.addData((1, "alice", 10L, true))
+ runPipeline(buildCtx())
+ checkAnswer(spark.table(s"$catalog.$namespace.target"), Seq.empty)
+ // The tombstone lives only in the aux, as a live row (its deleted-by-batch-id marker is null;
+ // a non-null marker is what flags a row logically deleted), with no matching visible target
+ // row.
+ assert(spark.table(auxTableNameFor("target")).count() == 1,
+ "the delete-only run should record exactly one aux tombstone row")
+
+ // Run #2 (aux retained): a later upsert at seq=5, BELOW the recorded seq=10. Because the aux
+ // still holds the seq=10 tombstone, reconciliation weaves seq=5 in as a closed prior record
+ // ending at 10 rather than an open current record -- state the target alone could not supply.
+ stream.addData((1, "early", 5L, false))
+ runPipeline(buildCtx())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "early", 5L, 5L, 10L, scd2Meta(5L)))
+ )
+ }
+
+ test("auxiliary key-column-names property survives identifiers containing special " +
+ "characters that exercise both JSON and SQL string-literal escaping") {
+ // This test exercises the full identifier-text persistence path with composite keys whose
+ // names collectively cover every escape class:
+ // - `it's` -- single quote: not escaped by JSON; the writer must double it
+ // to `''` to keep the SQL TBLPROPERTIES literal well-formed.
+ // - `name with spaces` -- whitespace identifier: backtick-quoted in DDL, no escaping
+ // needed in the JSON or the property value.
+ // - `a"b` -- literal double quote: JSON escapes as `\"`.
+ // - `c\d` -- literal backslash: JSON escapes as `\\`.
+ // If any layer drops, splits, or misescapes a name, the post-run lookup of the
+ // [[AutoCdcAuxiliaryTable.keyColumnNamesProperty]] property either fails to read or
+ // returns a value that is no longer a parseable JSON array of strings.
+ val keyNames = Seq("it's", "name with spaces", "a\"b", "c\\d")
+
+ // SQL DDL identifier rendering: backticks delimit each identifier; an embedded backtick
+ // would have to be escaped by doubling, but none of these names contain one.
+ val targetTableDdl = keyNames
+ .map(name => s"`$name` STRING NOT NULL")
+ .mkString(", ") + s", version BIGINT NOT NULL, $scd2MetadataDdl"
+ spark.sql(s"CREATE TABLE $catalog.$namespace.target ($targetTableDdl)")
+
+ // The AutoCDC API runs every key through `UnqualifiedColumnName.apply`, which calls
+ // `CatalystSqlParser.parseMultipartIdentifier`. To get a single-part identifier whose
+ // text includes special characters, the API caller has to backtick-quote at the boundary;
+ // we mirror that here by wrapping each name in backticks (and doubling any embedded
+ // backtick -- not needed for these names but kept for parity with how a user would call
+ // the API).
+ val backtickQuotedKeys = keyNames.map(name => s"`${name.replace("`", "``")}`")
+
+ // Single MemoryStream reused across both runs so the streaming checkpoint can resume.
+ val stream = MemoryStream[(String, String, String, String, Long)]
+ def buildCtx(): TestGraphRegistrationContext =
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF((keyNames :+ "version"): _*),
+ keys = backtickQuotedKeys,
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ // Run #1: a single insert with arbitrary non-empty key values.
+ stream.addData(("v1", "v2", "v3", "v4", 1L))
+ runPipeline(buildCtx())
+
+ // The persisted property must round-trip every name byte-for-byte.
+ assert(getAuxTableKeyColumnNames(target = "target") == keyNames)
+
+ // Run #2: same keys, a higher sequence -- drift validation reads the property back, parses
+ // the JSON, and looks up each recorded name in the aux schema. If any layer mangled the
+ // identifier text (lost an escape, dropped a `'`, split on a `.`, ...), validation would
+ // either throw KEY_SCHEMA_DRIFT (name lookup miss) or INTERNAL_ERROR (recorded name absent
+ // from aux schema). Reaching the second run successfully proves the round-trip works.
+ stream.addData(("v1", "v2", "v3", "v4", 2L))
+ runPipeline(buildCtx())
+
+ // The persisted property is immutable across non-full-refresh runs, so it must still be
+ // intact after run #2.
+ assert(getAuxTableKeyColumnNames(target = "target") == keyNames)
+ }
+
+ private def getAuxTableKeyColumnNames(target: String): Seq[String] = {
+ val auxName = auxTableNameFor(target)
+ val rows = spark.sql(s"SHOW TBLPROPERTIES $auxName").collect()
+ val prop = rows
+ .find(_.getString(0) == AutoCdcAuxiliaryTable.keyColumnNamesProperty)
+ .getOrElse(fail(
+ s"auxiliary table $auxName is missing the " +
+ s"${AutoCdcAuxiliaryTable.keyColumnNamesProperty} property; got: ${rows.toSeq}"
+ ))
+ AutoCdcAuxiliaryTable.parseColumnNames(prop.getString(1))
+ .getOrElse(fail(
+ s"auxiliary table $auxName has a malformed " +
+ s"${AutoCdcAuxiliaryTable.keyColumnNamesProperty} property: '${prop.getString(1)}'"
+ ))
+ }
+}
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2FullRefreshSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2FullRefreshSuite.scala
new file mode 100644
index 000000000000..3b6151d6c929
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2FullRefreshSuite.scala
@@ -0,0 +1,263 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ ColumnSelection,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests covering AutoCDC's full-refresh semantics for SCD Type 2 targets: full refresh must wipe
+ * both the target rows and the (richer) SCD2 auxiliary table for the refreshed targets, and must
+ * leave non-refreshed targets untouched in selective-refresh mode. The SCD2 analog of
+ * [[AutoCdcScd1FullRefreshSuite]].
+ */
+class AutoCdcScd2FullRefreshSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ import testImplicits._
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ /** Create an SCD2 target with user columns `(id, name, version)` plus the framework columns. */
+ private def createScd2Target(table: String): Unit = {
+ spark.sql(
+ s"CREATE TABLE $table (" +
+ s"id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ }
+
+ test("full refresh wipes target rows and the auxiliary table for the refreshed flow") {
+ createScd2Target(s"$catalog.$namespace.target")
+
+ // Run #1: populate target + auxiliary table.
+ val stream1 = MemoryStream[(Int, String, Long)]
+ stream1.addData((1, "alice", 5L))
+ val ctx1 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream1.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+ runPipeline(ctx1)
+ assert(
+ spark.catalog.tableExists(auxTableNameFor("target")),
+ "Auxiliary table should exist after first run"
+ )
+
+ // Run #2 (full refresh): auxiliary table should be dropped by DatasetManager, target
+ // truncated. The new run brings only id=2 at seq=1.
+ val stream2 = MemoryStream[(Int, String, Long)]
+ stream2.addData((2, "bob", 1L))
+ val ctx2 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream2.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+ val updateCtx = TestPipelineUpdateContext(
+ spark,
+ ctx2.toDataflowGraph,
+ storageRoot,
+ fullRefreshTables = AllTables
+ )
+ updateCtx.pipelineExecution.runPipeline()
+ updateCtx.pipelineExecution.awaitCompletion()
+
+ // Only id=2 remains, as a single open current record; id=1 from run #1 is wiped.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(2, "bob", 1L, 1L, null, scd2Meta(1L)))
+ )
+ }
+
+ test("after a full refresh, an event with a sequence below the previous run's " +
+ "watermark now lands") {
+ createScd2Target(s"$catalog.$namespace.target")
+
+ // Run #1: delete at seq=10 sets a high watermark in the auxiliary table.
+ val stream1 = MemoryStream[(Int, String, Long, Boolean)]
+ stream1.addData((1, "alice", 10L, true))
+ val ctx1 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream1.toDF().toDF("id", "name", "version", "is_delete")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ deleteCondition = Some(functions.col("is_delete") === true),
+ columnSelection = Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("is_delete"))
+ )),
+ scdType = ScdType.Type2
+ ))
+ }
+ runPipeline(ctx1)
+
+ // Run #2 (full refresh): auxiliary table is dropped, watermark reset. seq=5 should
+ // now land as an open current record.
+ val stream2 = MemoryStream[(Int, String, Long, Boolean)]
+ stream2.addData((1, "fresh", 5L, false))
+ val ctx2 = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream2.toDF().toDF("id", "name", "version", "is_delete")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ deleteCondition = Some(functions.col("is_delete") === true),
+ columnSelection = Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("is_delete"))
+ )),
+ scdType = ScdType.Type2
+ ))
+ }
+ val updateCtx = TestPipelineUpdateContext(
+ spark,
+ ctx2.toDataflowGraph,
+ storageRoot,
+ fullRefreshTables = AllTables
+ )
+ updateCtx.pipelineExecution.runPipeline()
+ updateCtx.pipelineExecution.awaitCompletion()
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "fresh", 5L, 5L, null, scd2Meta(5L)))
+ )
+ }
+
+ test("selective full refresh wipes only the requested target's auxiliary state") {
+ createScd2Target(s"$catalog.$namespace.t_a")
+ createScd2Target(s"$catalog.$namespace.t_b")
+
+ // t_b's run #1 is delete-only so that the state proving the aux was spared lives ONLY in the
+ // aux: the target stays empty and the aux holds a seq=10 tombstone. In run #2, t_b's seq=5
+ // upsert landing closed at 10 is possible only because the selective refresh left t_b's aux
+ // intact -- target state alone could not supply the seq=10 closure. (An open upsert in run #1
+ // would instead route to the target and leave the aux empty, so the assertion would hold even
+ // if the aux were wiped, attributing the outcome to the wrong state.)
+ //
+ // streamA is replaced across runs because t_a is full-refreshed in run #2 (its streaming
+ // checkpoint is reset by full-refresh, so a fresh source is fine and matches the user-visible
+ // semantics). streamB is reused across runs because t_b is NOT full-refreshed -- its
+ // streaming checkpoint must resume against the same MemoryStream instance, otherwise the
+ // seq=5 assertion below could pass for the wrong reason (the source never produced seq=5
+ // in run #2 instead of the aux tombstone shaping it).
+ val streamA1 = MemoryStream[(Int, String, Long)]
+ val streamB = MemoryStream[(Int, String, Long, Boolean)]
+ streamA1.addData((1, "a", 10L))
+ streamB.addData((1, "b", 10L, true)) // delete at seq=10: target empty, aux tombstone at 10
+ // dfFlowFunc is a TestGraphRegistrationContext method, so it can only be called inside the
+ // context blocks below; flowB takes the already-built query and adds t_b's delete knobs.
+ def flowB(query: FlowFunction): AutoCdcFlow = autoCdcFlow(
+ name = "flow_b",
+ target = "t_b",
+ query = query,
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ deleteCondition = Some(functions.col("is_delete") === true),
+ columnSelection = Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("is_delete"))
+ )),
+ scdType = ScdType.Type2
+ )
+ val ctx1 = new TestGraphRegistrationContext(spark) {
+ registerTable("t_a", catalog = Some(catalog), database = Some(namespace))
+ registerTable("t_b", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "flow_a",
+ target = "t_a",
+ query = dfFlowFunc(streamA1.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ registerFlow(flowB(dfFlowFunc(streamB.toDF().toDF("id", "name", "version", "is_delete"))))
+ }
+ runPipeline(ctx1)
+ // Precondition: t_b's run #1 left the target empty with the seq=10 tombstone only in the aux.
+ checkAnswer(spark.table(s"$catalog.$namespace.t_b"), Seq.empty)
+
+ // Run #2: full refresh ONLY on t_a; t_b's auxiliary state must persist.
+ val streamA2 = MemoryStream[(Int, String, Long)]
+ // t_a's aux is wiped, so seq=5 is the only record it has ever seen: a fresh open record.
+ streamA2.addData((1, "a2", 5L))
+ // t_b keeps its aux (seq=10 tombstone). The late seq=5 upsert is woven into history as a
+ // closed prior record ending at seq=10 -- a closure only the retained aux can supply.
+ streamB.addData((1, "b2", 5L, false))
+ val ctx2 = new TestGraphRegistrationContext(spark) {
+ registerTable("t_a", catalog = Some(catalog), database = Some(namespace))
+ registerTable("t_b", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "flow_a",
+ target = "t_a",
+ query = dfFlowFunc(streamA2.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ registerFlow(flowB(dfFlowFunc(streamB.toDF().toDF("id", "name", "version", "is_delete"))))
+ }
+ val updateCtx = TestPipelineUpdateContext(
+ spark,
+ ctx2.toDataflowGraph,
+ storageRoot,
+ fullRefreshTables = SomeTables(Set(
+ fullyQualifiedIdentifier("t_a", Some(catalog), Some(namespace))
+ ))
+ )
+ updateCtx.pipelineExecution.runPipeline()
+ updateCtx.pipelineExecution.awaitCompletion()
+
+ // t_a: refreshed, so the seq=5 event lands as a fresh open current record.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.t_a"),
+ Seq(Row(1, "a2", 5L, 5L, null, scd2Meta(5L)))
+ )
+ // t_b: aux retained, so the late seq=5 event is woven in as a closed prior record ending at
+ // the tombstoned seq=10. With no open successor (the seq=10 event was a delete), the closed
+ // [5, 10) record is the only visible row.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.t_b"),
+ Seq(Row(1, "b2", 5L, 5L, 10L, scd2Meta(5L)))
+ )
+ }
+}
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala
new file mode 100644
index 000000000000..43fc60c23bdd
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala
@@ -0,0 +1,487 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.sql.classic.DataFrame
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.pipelines.autocdc.{Scd2BatchProcessor, ScdType}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.MetadataBuilder
+
+/**
+ * End-to-end tests covering AutoCDC SCD2 key-drift validation: the AutoCDC flow's declared keys
+ * are validated against the auxiliary table's recorded keys at flow execution-init time. A change
+ * in keys across runs without a full refresh corrupts the merge semantics; validation detects this
+ * and fails fast with a structured [[AUTOCDC_INVALID_STATE]] error. The SCD2 analog of
+ * [[AutoCdcScd1KeyDriftSuite]].
+ *
+ * Key-drift validation itself is SCD-type-agnostic (it compares recorded vs declared key
+ * (name, dataType) sets), so these mirror the SCD1 cases with SCD2 flows. The tests that
+ * pre-create a tampered auxiliary table build an SCD2-shaped aux table (full target row schema
+ * plus the deleted-by-batch-id marker) carrying the SCD2 scd-type property.
+ *
+ * Two deliberate departures from a strict one-to-one mirror keep both suites at fourteen tests:
+ * [[AutoCdcScd1KeyDriftSuite]]'s "AutoCDC key drift validation uses pipeline case sensitivity, not
+ * session default" has no analog here -- that the validator reads the pipeline's conf rather than
+ * the ambient session conf is SCD-type-agnostic and already pinned by the SCD1 suite -- and in its
+ * place this suite adds the SCD_TYPE_DRIFT case, which the SCD1 suite structurally cannot cover.
+ */
+class AutoCdcScd2KeyDriftSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ import testImplicits._
+
+ /**
+ * Properties clause seeding an SCD2 auxiliary table with the scd-type property and the given
+ * JSON key-column-names array, so drift validation classifies it as SCD2 and reads its keys.
+ */
+ private def scd2AuxProps(keyColumnNamesJson: String): String =
+ s"TBLPROPERTIES (" +
+ s"'${AutoCdcAuxiliaryTable.scdTypePropertyKey}' = '${ScdType.Type2.label}', " +
+ s"'${AutoCdcAuxiliaryTable.keyColumnNamesProperty}' = '$keyColumnNamesJson')"
+
+ test("a pipeline execution that adds a key column to an existing AutoCDC flow triggers " +
+ "KEY_SCHEMA_DRIFT") {
+ // Target table carries both candidate key columns up-front so only the AutoCDC `keys`
+ // declaration differs between the two pipelines.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, region STRING NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1 declares one key (`id`).
+ val stream1 = MemoryStream[(Int, String, Long)]
+ stream1.addData((1, "us", 1L))
+ runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "region", "version"), Seq("id")))
+
+ // Pipeline #2 declares two keys (`region` + `id`) - arity drift.
+ val stream2 = MemoryStream[(Int, String, Long)]
+ stream2.addData((1, "us", 2L))
+ val ctx2 = buildPipeline(
+ "flow_v2", stream2.toDF().toDF("id", "region", "version"), Seq("region", "id"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ // `region` is nullable here because Scala `String` is a reference type and the
+ // [[MemoryStream]] tuple encoder treats reference types as nullable. Only Scala
+ // primitives (`Int`, `Long`, ...) yield `NOT NULL` columns.
+ "expectedKeySchema" -> "region STRING,id INT NOT NULL",
+ "recordedKeySchema" -> "id INT NOT NULL"
+ )
+ )
+ }
+
+ test("a pipeline execution that drops a key column from an existing AutoCDC flow triggers " +
+ "KEY_SCHEMA_DRIFT") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(region STRING NOT NULL, id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1 declares two keys [region, id].
+ val stream1 = MemoryStream[(String, Int, Long)]
+ stream1.addData(("us", 1, 1L))
+ runPipeline(buildPipeline(
+ "flow_v1", stream1.toDF().toDF("region", "id", "version"), Seq("region", "id")))
+
+ // Pipeline #2 declares only [id] - arity drift.
+ val stream2 = MemoryStream[(String, Int, Long)]
+ stream2.addData(("us", 1, 2L))
+ val ctx2 = buildPipeline("flow_v2", stream2.toDF().toDF("region", "id", "version"), Seq("id"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ "expectedKeySchema" -> "id INT NOT NULL",
+ // `region` is nullable here because Scala `String` is a reference type; see the
+ // analogous comment in the "adds a key column" test above.
+ "recordedKeySchema" -> "region STRING,id INT NOT NULL"
+ )
+ )
+ }
+
+ test("a pipeline execution that swaps a key in an existing AutoCDC flow for a different name " +
+ "(same arity) triggers KEY_SCHEMA_DRIFT") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, region STRING NOT NULL, country STRING NOT NULL, " +
+ s"version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1 declares [id, region].
+ val stream1 = MemoryStream[(Int, String, String, Long)]
+ stream1.addData((1, "us", "USA", 1L))
+ runPipeline(buildPipeline(
+ "flow_v1", stream1.toDF().toDF("id", "region", "country", "version"), Seq("id", "region")))
+
+ // Pipeline #2 declares [id, country] - same arity, different key set.
+ val stream2 = MemoryStream[(Int, String, String, Long)]
+ stream2.addData((1, "us", "USA", 2L))
+ val ctx2 = buildPipeline(
+ "flow_v2", stream2.toDF().toDF("id", "region", "country", "version"), Seq("id", "country"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ // `country` and `region` are nullable here because Scala `String` is a reference type;
+ // see the analogous comment in the "adds a key column" test above.
+ "expectedKeySchema" -> "id INT NOT NULL,country STRING",
+ "recordedKeySchema" -> "id INT NOT NULL,region STRING"
+ )
+ )
+ }
+
+ test("a pipeline whose recorded aux key dataType differs from the flow's source dataType " +
+ "triggers KEY_SCHEMA_DRIFT") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ // Pre-seed an SCD2-shaped aux table whose recorded key `id` is BIGINT, differing from the
+ // flow's INT source key.
+ spark.sql(
+ s"""CREATE TABLE ${auxTableNameFor("target")} """ +
+ s"""(id BIGINT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ +
+ s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) ${scd2AuxProps("[\"id\"]")}"""
+ )
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 1L))
+ val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ "expectedKeySchema" -> "id INT NOT NULL",
+ "recordedKeySchema" -> "id BIGINT NOT NULL"
+ )
+ )
+ }
+
+ test("a composite key reorder ([a,b] -> [b,a]) does NOT trigger drift validation") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(a INT NOT NULL, b STRING NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1 declares keys [a, b]; pipeline #2 declares the same set reversed [b, a]. Drift
+ // validation is order-independent, so pipeline #2 must NOT throw.
+ val stream1 = MemoryStream[(Int, String, Long)]
+ stream1.addData((1, "x", 1L))
+ runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("a", "b", "version"), Seq("a", "b")))
+
+ val stream2 = MemoryStream[(Int, String, Long)]
+ stream2.addData((2, "y", 1L))
+ runPipeline(buildPipeline("flow_v2", stream2.toDF().toDF("a", "b", "version"), Seq("b", "a")))
+ }
+
+ test("a pipeline execution that changes a key column's nullability or metadata in an " +
+ "existing AutoCDC flow does NOT trigger drift") {
+ // Drift validation compares (name, dataType) pairs as a set; nullability and column metadata
+ // are not part of [[DataType]], so they do not gate semantic equivalence.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1: source carries `id INT NOT NULL` (Scala primitive `Int`), no metadata.
+ val stream1 = MemoryStream[(Int, Long)]
+ stream1.addData((1, 1L))
+ runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("id")))
+
+ // Pipeline #2: source carries `id INT` (nullable, via `Option[Int]`) AND attaches non-empty
+ // column metadata. Same name and `dataType` as the recorded key, but every [[StructField]]
+ // aspect outside `dataType` differs.
+ val stream2 = MemoryStream[(Option[Int], Long)]
+ stream2.addData((Some(2), 2L))
+ val baseDf = stream2.toDF().toDF("id", "version")
+ val md = new MetadataBuilder()
+ .putString("description", "primary key")
+ .build()
+ val sourceDfWithMetadata = baseDf.select(baseDf("id").as("id", md), baseDf("version"))
+ runPipeline(buildPipeline("flow_v2", sourceDfWithMetadata, Seq("id")))
+ }
+
+ test("a pipeline execution that wraps an existing AutoCDC flow's key in backticks does NOT " +
+ "trigger drift") {
+ // Backticks are a SQL-parse syntactic device, not part of the identifier itself.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Long)]
+ stream1.addData((1, 1L))
+ runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("id")))
+
+ val stream2 = MemoryStream[(Int, Long)]
+ stream2.addData((2, 1L))
+ runPipeline(buildPipeline("flow_v2", stream2.toDF().toDF("id", "version"), Seq("`id`")))
+ }
+
+ test("a pipeline execution that drops backticks around an existing AutoCDC flow's " +
+ "previously-backtick-quoted key does NOT trigger drift") {
+ // The reverse direction: drift validation must be backtick-invariant on both the write side
+ // (recorded property strips backticks) and the read side (resolver-aware lookup).
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Long)]
+ stream1.addData((1, 1L))
+ runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("`id`")))
+
+ val stream2 = MemoryStream[(Int, Long)]
+ stream2.addData((2, 1L))
+ runPipeline(buildPipeline("flow_v2", stream2.toDF().toDF("id", "version"), Seq("id")))
+ }
+
+ test("under spark.sql.caseSensitive = true, an AutoCDC flow whose key differs only in case " +
+ "from the recorded key triggers KEY_SCHEMA_DRIFT") {
+ // validateNoKeyColumnDrift uses spark.sessionState.conf.resolver, so its behavior on
+ // `Id` vs `id` flips with the session conf. Pipeline #1 seeds the aux under the default
+ // resolver with recorded key `["id"]`; pipeline #2 runs under the case-sensitive resolver
+ // with key `["Id"]`, which is a distinct identifier there, so drift must fire.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Long)]
+ stream1.addData((1, 1L))
+ runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("id")))
+
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ val stream2 = MemoryStream[(Int, Long)]
+ stream2.addData((1, 2L))
+ val ctx2 = buildPipeline("flow_v2", stream2.toDF().toDF("Id", "version"), Seq("Id"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ "expectedKeySchema" -> "Id INT NOT NULL",
+ "recordedKeySchema" -> "id INT NOT NULL"
+ )
+ )
+ }
+ }
+
+ test("under the default (case-insensitive) resolver, an AutoCDC flow whose key differs only " +
+ "in case from the recorded key does NOT trigger drift") {
+ // Pairs with the case-sensitive test above: under the default resolver the two identifiers
+ // are equivalent, so drift validation must accept pipeline #2.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Long)]
+ stream1.addData((1, 1L))
+ runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("id")))
+
+ val stream2 = MemoryStream[(Int, Long)]
+ stream2.addData((1, 2L))
+ runPipeline(buildPipeline("flow_v2", stream2.toDF().toDF("id", "version"), Seq("Id")))
+ }
+
+ test("a pipeline whose aux table is missing the keyColumnNames property fails with " +
+ "AUXILIARY_TABLE_PROPERTY_MISSING") {
+ // Pre-create the aux table directly without the [[keyColumnNamesProperty]] to simulate
+ // corrupt metadata. Validation must surface a structured AUTOCDC_INVALID_STATE error.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ spark.sql(
+ s"""CREATE TABLE ${auxTableNameFor("target")} """ +
+ s"""(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ +
+ s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) """ +
+ s"""TBLPROPERTIES ('${AutoCdcAuxiliaryTable.scdTypePropertyKey}' = """ +
+ s"""'${ScdType.Type2.label}')"""
+ )
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 1L))
+ val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty
+ )
+ )
+ }
+
+ test("a pipeline whose aux table has a malformed keyColumnNames property fails with " +
+ "AUXILIARY_TABLE_PROPERTY_MALFORMED") {
+ // Pre-create the aux table directly with a non-JSON-array property value to simulate corrupt
+ // metadata. Validation must surface a structured AUTOCDC_INVALID_STATE error.
+ val malformedKeysArray = "not-a-json-array"
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ spark.sql(
+ s"""CREATE TABLE ${auxTableNameFor("target")} """ +
+ s"""(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ +
+ s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) """ +
+ scd2AuxProps(malformedKeysArray)
+ )
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 1L))
+ val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MALFORMED",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty,
+ "rawValue" -> malformedKeysArray
+ )
+ )
+ }
+
+ test("a pipeline whose aux table records a key absent from its schema fails with " +
+ "AUXILIARY_TABLE_KEY_COLUMN_MISSING") {
+ // Pre-create the aux table with the [[keyColumnNamesProperty]] pointing at a column that does
+ // not exist in the aux schema. Validation must surface a structured AUTOCDC_INVALID_STATE
+ // error rather than KEY_SCHEMA_DRIFT.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ spark.sql(
+ s"""CREATE TABLE ${auxTableNameFor("target")} """ +
+ s"""(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ +
+ s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) ${scd2AuxProps("[\"region\"]")}"""
+ )
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 1L))
+ val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_KEY_COLUMN_MISSING",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ "keyColumnName" -> "region",
+ "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty
+ )
+ )
+ }
+
+ test("a pipeline execution whose recorded SCD type differs from the flow's SCD type triggers " +
+ "SCD_TYPE_DRIFT") {
+ // SCD_TYPE_DRIFT is reachable end-to-end from DatasetManager (validateNoScdTypeDrift runs
+ // right after validateNoKeyColumnDrift when evolving the aux table), but the SCD1 suite
+ // structurally cannot cover it, so this SCD2 suite is its home. Pre-create an SCD2-shaped aux
+ // table whose recorded scd-type property is Type1, with matching keys so key-drift validation
+ // passes and the scd-type check is the one that fires, then run a Type2 flow.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ spark.sql(
+ s"""CREATE TABLE ${auxTableNameFor("target")} """ +
+ s"""(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ +
+ s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) """ +
+ s"""TBLPROPERTIES (""" +
+ s"""'${AutoCdcAuxiliaryTable.scdTypePropertyKey}' = '${ScdType.Type1.label}', """ +
+ s"""'${AutoCdcAuxiliaryTable.keyColumnNamesProperty}' = '[\"id\"]')"""
+ )
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 1L))
+ val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id"))
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.SCD_TYPE_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString,
+ "expectedScdType" -> ScdType.Type2.label,
+ "recordedScdType" -> ScdType.Type1.label
+ )
+ )
+ }
+
+ /**
+ * Build a single-flow SCD2 pipeline targeting `cat.ns1.target` with the given source DF and key
+ * column list. Thin wrapper over [[singleAutoCdcFlowPipeline]] since every drift test targets
+ * the same `target` table.
+ */
+ private def buildPipeline(
+ flowName: String,
+ sourceDf: DataFrame,
+ keys: Seq[String]): TestGraphRegistrationContext =
+ singleAutoCdcFlowPipeline(
+ flowName = flowName,
+ target = "target",
+ sourceDf = sourceDf,
+ keys = keys,
+ sequencing = $"version",
+ scdType = ScdType.Type2)
+}
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2MultiPipelineSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2MultiPipelineSuite.scala
new file mode 100644
index 000000000000..cd161282e729
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2MultiPipelineSuite.scala
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType, UnqualifiedColumnName}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end tests that exercise interactions between separate SCD Type 2 AutoCDC pipelines (i.e.
+ * distinct [[DataflowGraph]] / [[TestPipelineUpdateContext]] invocations) sharing the same v2
+ * catalog. The SCD2 analog of [[AutoCdcScd1MultiPipelineSuite]]: independent target/auxiliary
+ * tables per target, downstream reads that ignore the framework columns, a shared target written
+ * by two pipelines, schema evolution across pipelines, and key-drift rejection.
+ */
+class AutoCdcScd2MultiPipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ import testImplicits._
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ test("two AutoCDC pipelines targeting separate tables maintain independent target and " +
+ "auxiliary tables") {
+ // Two distinct target tables created up-front.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_a " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_b " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1 only knows about `t_a`. Its auxiliary table must not affect pipeline #2's `t_b`.
+ val streamA = MemoryStream[(Int, String, Long)]
+ streamA.addData((1, "alice", 100L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "flow_a",
+ target = "t_a",
+ sourceDf = streamA.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2))
+
+ // Pipeline #2 only knows about `t_b`. Uses a deliberately *lower* sequence to verify the
+ // watermark from pipeline #1's auxiliary table (seq=100) does not leak into pipeline #2.
+ val streamB = MemoryStream[(Int, String, Long)]
+ streamB.addData((9, "bob", 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "flow_b",
+ target = "t_b",
+ sourceDf = streamB.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2))
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.t_a"),
+ Seq(Row(1, "alice", 100L, 100L, null, scd2Meta(100L)))
+ )
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.t_b"),
+ Seq(Row(9, "bob", 1L, 1L, null, scd2Meta(1L)))
+ )
+
+ // Each target has its own auxiliary table; no cross-contamination.
+ assert(spark.catalog.tableExists(auxTableNameFor("t_a")))
+ assert(spark.catalog.tableExists(auxTableNameFor("t_b")))
+ }
+
+ test("a downstream pipeline can read an AutoCDC target written by a different pipeline " +
+ "without observing the framework columns") {
+ // Pipeline #1 writes into target `src` via AutoCDC.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.src " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 1L), (2, "bob", 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "writer",
+ target = "src",
+ sourceDf = stream.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2))
+
+ // Pipeline #2 is a regular materialized view that selects the user-data columns from `src`
+ // (a different graph entirely). It must observe the merged AutoCDC rows and be able to ignore
+ // the framework columns without them polluting downstream consumers.
+ val ctxReader = new TestGraphRegistrationContext(spark) {
+ registerMaterializedView(
+ "downstream_mv",
+ query = dfFlowFunc(
+ spark.read.table(s"$catalog.$namespace.src").select("id", "name", "version")
+ )
+ )
+ }
+ runPipeline(ctxReader)
+
+ checkAnswer(
+ spark.table(fullyQualifiedIdentifier("downstream_mv").toString),
+ Seq(Row(1, "alice", 1L), Row(2, "bob", 1L))
+ )
+ }
+
+ test("two AutoCDC pipelines targeting the same table with identical key and data " +
+ "schemas merge into a shared target table") {
+ // Target table is created once up-front; both pipelines target it with the same AutoCDC
+ // `keys` and the same source-DF data schema. The two pipelines have distinct flow names so
+ // they own independent streaming checkpoints, but share the target and its auxiliary table.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.shared_target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1: inserts rows with id=1 and id=2 at version=1.
+ val stream1 = MemoryStream[(Int, String, Long)]
+ stream1.addData((1, "alice", 1L), (2, "bob", 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "flow_v1",
+ target = "shared_target",
+ sourceDf = stream1.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2))
+
+ // Sanity-check pipeline #1's effect before pipeline #2 runs.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.shared_target"),
+ Seq(
+ Row(1, "alice", 1L, 1L, null, scd2Meta(1L)),
+ Row(2, "bob", 1L, 1L, null, scd2Meta(1L))
+ )
+ )
+
+ // Pipeline #2: updates id=2 (existing key) to a higher sequence and inserts id=3 (new key).
+ // id=1 is untouched and must survive into the final target unchanged.
+ val stream2 = MemoryStream[(Int, String, Long)]
+ stream2.addData((2, "bob-v2", 2L), (3, "carol", 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "flow_v2",
+ target = "shared_target",
+ sourceDf = stream2.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2))
+
+ // Final target: id=1 untouched; id=2's original record closed at seq=2 with a new open record;
+ // id=3 freshly inserted by pipeline #2.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.shared_target"),
+ Seq(
+ Row(1, "alice", 1L, 1L, null, scd2Meta(1L)),
+ Row(2, "bob", 1L, 1L, 2L, scd2Meta(1L)),
+ Row(2, "bob-v2", 2L, 2L, null, scd2Meta(2L)),
+ Row(3, "carol", 1L, 1L, null, scd2Meta(1L))
+ )
+ )
+
+ // The auxiliary table for the shared target is itself shared across both pipelines.
+ assert(spark.catalog.tableExists(auxTableNameFor("shared_target")))
+ }
+
+ test("two AutoCDC pipelines targeting the same table with the same key but different " +
+ "data columns evolve the shared target schema") {
+ // Target is created up-front with pipeline #1's schema only; pipeline #2 brings a new
+ // top-level nullable `age` column that the dataset materialization layer is expected to
+ // schema-merge into the target.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.shared_target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1: source DF schema is (id, name, version); inserts id=1 and id=2.
+ val stream1 = MemoryStream[(Int, String, Long)]
+ stream1.addData((1, "alice", 1L), (2, "bob", 1L))
+ val ctx1 = singleAutoCdcFlowPipeline(
+ flowName = "flow_v1",
+ target = "shared_target",
+ sourceDf = stream1.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2,
+ // Both pipelines pin the tracked set to `name`, the one non-key column they share. Left to
+ // the default (selection-derived) tracking, pipeline #2's extra `age` would also widen the
+ // tracked set, which is rejected as TRACK_HISTORY_DRIFT (SPARK-58391); the axis under test
+ // here is the shared target's data-column evolution.
+ trackHistorySelection =
+ Option(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name")))))
+ runPipeline(ctx1)
+
+ // Sanity-check pipeline #1's state before schema evolution kicks in.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.shared_target"),
+ Seq(
+ Row(1, "alice", 1L, 1L, null, scd2Meta(1L)),
+ Row(2, "bob", 1L, 1L, null, scd2Meta(1L))
+ )
+ )
+
+ // Pipeline #2: source DF schema is (id, name, age, version). The new nullable `age` column
+ // should be added to the target by dataset materialization; pipeline #1's untouched id=1 row
+ // is backfilled to NULL. The `age` column lands after the framework columns in the target.
+ val stream2 = MemoryStream[(Int, String, Option[Int], Long)]
+ stream2.addData((2, "bob-v2", Some(25), 2L), (3, "carol", Some(30), 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "flow_v2",
+ target = "shared_target",
+ sourceDf = stream2.toDF().toDF("id", "name", "age", "version"),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2,
+ trackHistorySelection =
+ Option(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))))))
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.shared_target"),
+ Seq(
+ Row(1, "alice", 1L, 1L, null, scd2Meta(1L), null),
+ Row(2, "bob", 1L, 1L, 2L, scd2Meta(1L), null),
+ Row(2, "bob-v2", 2L, 2L, null, scd2Meta(2L), 25),
+ Row(3, "carol", 1L, 1L, null, scd2Meta(1L), 30)
+ )
+ )
+
+ // NOTE: the SCD1 analog of this test additionally re-runs the narrower pipeline #1 against the
+ // now-wider evolved target. For SCD2 that microbatch-narrower-than-target path is covered
+ // separately by AutoCdcScd2ColumnEvolutionSuite (SPARK-58418), so it is not duplicated here.
+ }
+
+ test("a second pipeline targeting an existing AutoCDC table with different keys " +
+ "fails with KEY_SCHEMA_DRIFT") {
+ // Target table with both candidate keys present so the second pipeline would otherwise be
+ // schema-compatible with the first; only the AutoCDC `keys` differ between flows.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.shared_target " +
+ s"(id INT NOT NULL, name STRING NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Pipeline #1: AutoCDC flow keyed on `id`.
+ val stream1 = MemoryStream[(Int, String, Long)]
+ stream1.addData((1, "alice", 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "flow_v1",
+ target = "shared_target",
+ sourceDf = stream1.toDF().toDF("id", "name", "version"),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2))
+
+ // Pipeline #2: completely separate graph, but targets the same physical `shared_target`
+ // table with `keys = Seq("name")`.
+ val stream2 = MemoryStream[(Int, String, Long)]
+ stream2.addData((2, "alice", 1L))
+ val ctx2 = singleAutoCdcFlowPipeline(
+ flowName = "flow_v2",
+ target = "shared_target",
+ sourceDf = stream2.toDF().toDF("id", "name", "version"),
+ keys = Seq("name"),
+ sequencing = $"version",
+ scdType = ScdType.Type2)
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("shared_target", Some(catalog), Some(namespace)).unquotedString,
+ // Pipeline #2's AutoCDC key resolves from the source DF, where `MemoryStream[(Int, String,
+ // Long)]` produces a nullable StringType for `name`.
+ "expectedKeySchema" -> "name STRING",
+ // Pipeline #1 persisted the aux table from a source DF whose `id` was a non-null Scala
+ // primitive (`Int`), so the recorded key carries `NOT NULL`.
+ "recordedKeySchema" -> "id INT NOT NULL"
+ )
+ )
+ }
+}
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala
new file mode 100644
index 000000000000..838a5f8ee2f5
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala
@@ -0,0 +1,526 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import java.sql.Timestamp
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType, UnqualifiedColumnName}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests covering SCD Type 2 AutoCDC's interaction with non-key schema evolution across pipeline
+ * runs. The SCD2 analog of [[AutoCdcScd1SchemaEvolutionSuite]]; documents the supported additive
+ * cases (new top-level columns, a new field inside an array<struct> element, broadening column
+ * selection) and the cases that fail loudly (incompatible type changes).
+ *
+ * Unlike SCD1, an SCD2 upsert to an existing key does not overwrite the row: it closes the prior
+ * record and opens a new one, so evolution assertions carry the full interval history.
+ *
+ * The additive cases pin `trackHistorySelection` to the columns present in every run. Under default
+ * tracking the tracked set is derived from the selected non-key columns, so adding a column widens
+ * it -- rejected as `AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT` (SPARK-58391) because it would
+ * reinterpret already-reconciled history. Pinning keeps each additive test on the axis it names;
+ * the drift rejection itself has its own test here. This is an intended SCD2-vs-SCD1 divergence:
+ * the same additive evolution needs no full refresh under SCD1.
+ *
+ * Scope notes -- cases intentionally covered elsewhere rather than duplicated here:
+ * - The *narrowing* / dropped-column cases (a microbatch narrower than the already-evolved
+ * target, incl. dropped nested struct/array fields) live in [[AutoCdcScd2ColumnEvolutionSuite]]
+ * under SPARK-58418, which makes them reconcile correctly.
+ * - Changing `trackHistorySelection` between runs -- the SCD2-only evolution axis, which decides
+ * whether an upsert opens a new record -- is also exercised end-to-end in
+ * [[AutoCdcScd2ColumnEvolutionSuite]], so it is not repeated here.
+ *
+ * One SCD1 evolution case has no SCD2 analog and so is absent here by design: "extra columns on
+ * the target that the AutoCDC flow does not emit are preserved" relies on SCD1's in-place overwrite
+ * (an SCD2 upsert instead reads unemitted target columns as NULL onto the newly-opened record), and
+ * there is no SCD2-specific preservation invariant to assert.
+ */
+class AutoCdcScd2SchemaEvolutionSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ import testImplicits._
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ test("a nullable non-key column merges correctly with mixed NULL and non-NULL values") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, email STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, Option[String], Long)]
+ def buildCtx(): TestGraphRegistrationContext =
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("id", "name", "email", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ // Run #1: insert with NULL email opens a current record.
+ stream.addData((1, "alice", None, 1L))
+ runPipeline(buildCtx())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", null, 1L, 1L, null, scd2Meta(1L)))
+ )
+
+ // Run #2: upsert with non-NULL email at higher seq closes the prior record and opens a new one.
+ stream.addData((1, "alice2", Some("a@x.com"), 2L))
+ runPipeline(buildCtx())
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", null, 1L, 1L, 2L, scd2Meta(1L)),
+ Row(1, "alice2", "a@x.com", 2L, 2L, null, scd2Meta(2L))
+ )
+ )
+ }
+
+ test("widening a non-key column's type between runs fails with " +
+ "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, age INT, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Int, Long)]
+ stream1.addData((1, 30, 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream1.toDF().toDF("id", "age", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2))
+
+ // Run #2: widen `age` from Int to Long.
+ val stream2 = MemoryStream[(Int, Long, Long)]
+ stream2.addData((1, 31L, 2L))
+ val ctx2 = singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream2.toDF().toDF("id", "age", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE",
+ sqlState = Some("42825"),
+ parameters = Map(
+ "left" -> "\"INT\"",
+ "right" -> "\"BIGINT\""
+ )
+ )
+ }
+
+ test("narrowing a non-key column's type between runs fails with " +
+ "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, payload BIGINT, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Long, Long)]
+ stream1.addData((1, 100L, 1L))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream1.toDF().toDF("id", "payload", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2))
+
+ // Run #2: narrow `payload` from Long (BIGINT) to Int (INT).
+ val stream2 = MemoryStream[(Int, Int, Long)]
+ stream2.addData((1, 5, 2L))
+ val ctx2 = singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream2.toDF().toDF("id", "payload", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE",
+ sqlState = Some("42825"),
+ parameters = Map(
+ "left" -> "\"BIGINT\"",
+ "right" -> "\"INT\""
+ )
+ )
+ }
+
+ test("a new top-level nullable column appearing in the source DF between runs is " +
+ "added to the target") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, Option[String], Long)]
+ def buildCtx(includeEmail: Boolean): TestGraphRegistrationContext = {
+ val sourceDf = stream.toDF().toDF("id", "name", "email", "version")
+ val projectedDf = if (includeEmail) sourceDf else sourceDf.drop("email")
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = projectedDf,
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2,
+ // Pin the tracked set to the columns present in both runs. Under default tracking the set
+ // is selection-derived, so adding `email` would also widen it -- a change that reinterprets
+ // already-reconciled history and is rejected as TRACK_HISTORY_DRIFT (SPARK-58391). Pinning
+ // isolates the axis under test here: additive evolution of the target's data columns.
+ trackHistorySelection =
+ Option(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name")))))
+ }
+
+ // Run #1: source projects (id, name, version). Target schema is unchanged.
+ stream.addData((1, "alice", None, 1L))
+ runPipeline(buildCtx(includeEmail = false))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, 1L, null, scd2Meta(1L)))
+ )
+
+ // Run #2: source projects (id, name, email, version) for a new key id=2. mergeSchemas appends
+ // `email` after the framework columns; the existing id=1 row gets NULL for the new column.
+ stream.addData((2, "bob", Some("b@x.com"), 2L))
+ runPipeline(buildCtx(includeEmail = true))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 1L, 1L, null, scd2Meta(1L), null),
+ Row(2, "bob", 2L, 2L, null, scd2Meta(2L), "b@x.com")
+ )
+ )
+ }
+
+ test("adding a source column under default tracking is rejected as track-history drift") {
+ // The counterpart to the additive tests above, which pin `trackHistorySelection` precisely to
+ // avoid this: with default tracking the tracked set is derived from the selected non-key
+ // columns (see `Scd2BatchProcessor.computeTrackedHistoryColumns`), so adding `email` silently
+ // widens it from [name, version] to [name, email, version]. That reinterprets which transitions
+ // open a new historical record and cannot be applied to already-reconciled history, so
+ // SPARK-58391 rejects it rather than letting the second run write history under different rules
+ // than the first. This is an intended SCD2-vs-SCD1 divergence: for SCD1 the same additive
+ // evolution needs no full refresh.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, Option[String], Long)]
+ def buildCtx(includeEmail: Boolean): TestGraphRegistrationContext = {
+ val sourceDf = stream.toDF().toDF("id", "name", "email", "version")
+ val projectedDf = if (includeEmail) sourceDf else sourceDf.drop("email")
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = projectedDf,
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+ }
+
+ // Run #1 records the tracked set as [name] on the auxiliary table.
+ stream.addData((1, "alice", None, 1L))
+ runPipeline(buildCtx(includeEmail = false))
+
+ // Run #2 would track [email, name]; the recorded set no longer matches.
+ stream.addData((2, "bob", Some("b@x.com"), 2L))
+ val ex = intercept[RuntimeException] { runPipeline(buildCtx(includeEmail = true)) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT",
+ parameters = Map(
+ "tableName" -> s"$catalog.$namespace.target",
+ // `version` is tracked too: only the keys and the reserved framework columns are excluded
+ // from the eligible set, and the sequencing column is neither.
+ "expectedTrackHistoryColumns" -> "name, email, version",
+ "recordedTrackHistoryColumns" -> "name, version"
+ )
+ )
+
+ // The rejection is a pre-write validation: the target still holds only run #1's row, and has
+ // not gained the `email` column.
+ val fieldNames = spark.table(s"$catalog.$namespace.target").schema.fieldNames.toSeq
+ assert(!fieldNames.contains("email"), s"target should not have gained `email`; got $fieldNames")
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, 1L, null, scd2Meta(1L)))
+ )
+ }
+
+ test("additive target-column evolution extends the SCD2 auxiliary table schema") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ // Shared (id, name, version) stream; run #1 projects away `name`, run #2 keeps it so the
+ // target (and, unlike SCD1, the aux table -- which mirrors the full target row) gain `name`.
+ val stream = MemoryStream[(Int, String, Long)]
+ def buildCtx(includeName: Boolean): TestGraphRegistrationContext = {
+ val sourceDf = stream.toDF().toDF("id", "name", "version")
+ val projectedDf = if (includeName) sourceDf else sourceDf.drop("name")
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = projectedDf,
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2,
+ // Pin the tracked set to `version` -- an eligible non-key column (the sequencing column is
+ // neither a key nor a reserved framework column) present in both runs, so the set stays
+ // stable while `name` is added. Tracking `name` itself would widen the set when it appears
+ // in run #2 and trip TRACK_HISTORY_DRIFT (SPARK-58391); see the note in the preceding test.
+ trackHistorySelection =
+ Option(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("version")))))
+ }
+
+ // Run #1: target is (id, version, framework); aux mirrors it plus the marker.
+ stream.addData((1, "ignored", 1L))
+ runPipeline(buildCtx(includeName = false))
+ val auxAfterRun1 = spark.table(auxTableNameFor("target")).schema.fieldNames.toSeq
+ assert(!auxAfterRun1.contains("name"),
+ s"aux schema after run #1 should not yet contain `name`; got $auxAfterRun1")
+
+ // Run #2: `name` is added to the target for a new key id=2. The SCD2 aux table mirrors the
+ // full target row schema, so it gains `name` too (unlike SCD1, whose aux holds only keys +
+ // metadata and is unaffected by non-key evolution).
+ stream.addData((2, "bob", 2L))
+ runPipeline(buildCtx(includeName = true))
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, 1L, 1L, null, scd2Meta(1L), null),
+ Row(2, 2L, 2L, null, scd2Meta(2L), "bob")
+ )
+ )
+ assert(spark.table(auxTableNameFor("target")).schema.fieldNames.contains("name"))
+ }
+
+ test("a new field added inside an array<struct> element between runs is added to the " +
+ "target") {
+ // SCD2 analog of AutoCdcScd1SchemaEvolutionSuite's array<struct> additive case: unlike the
+ // top-level scalar additions above, this exercises unionByName / mergeSchemas recursing into
+ // an array element struct. Unlike SCD1's overwrite-in-place, the SCD2 upsert closes the prior
+ // record (which never saw `vals.element.b.d`, so it reads NULL there) and opens a new one
+ // carrying the widened value.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(key INT NOT NULL, version BIGINT NOT NULL, " +
+ s"vals ARRAY<STRUCT<a:INT,b:STRUCT<c:INT>>>, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, Long, Int, Int, Int)]
+ def buildCtx(includeD: Boolean): TestGraphRegistrationContext = {
+ val src = stream.toDF().toDF("key", "version", "a", "b_c", "b_d")
+ val inner = if (includeD) {
+ functions.struct(functions.col("b_c").as("c"), functions.col("b_d").as("d"))
+ } else {
+ functions.struct(functions.col("b_c").as("c"))
+ }
+ val projected = src.select(
+ functions.col("key"),
+ functions.col("version"),
+ functions.array(
+ functions.struct(functions.col("a"), inner.as("b"))
+ ).as("vals")
+ )
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = projected,
+ keys = Seq("key"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+ }
+
+ // Run #1: element struct is (a, b.c); no b.d yet. Opens key=1's current record at version=1.
+ stream.addData((1, 1L, 1, 1, 99))
+ runPipeline(buildCtx(includeD = false))
+
+ // Run #2 widens the element struct with b.d. The version=2 upsert to key=1 closes its
+ // version=1 record (which predates b.d, so reads NULL) and opens a new one with b.d=2; the
+ // new key=3 lands as an open record with the full widened struct.
+ stream.addData((1, 2L, 1, 1, 2), (3, 1L, 3, 3, 3))
+ runPipeline(buildCtx(includeD = true))
+
+ // Inline-explode flattens the array<struct>; carry the interval bounds to prove the closed
+ // prior record reads NULL for the newly-added nested field.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target")
+ .selectExpr("key", "__START_AT", "__END_AT", "inline(vals) as (a, b)")
+ .select("key", "__START_AT", "__END_AT", "a", "b.c", "b.d"),
+ Seq(
+ Row(1, 1L, 2L, 1, 1, null),
+ Row(1, 2L, null, 1, 1, 2),
+ Row(3, 1L, null, 3, 3, 3)
+ )
+ )
+ }
+
+ test("broadening the column selection between runs adds the newly-included column to " +
+ "the target") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, String, Long)]
+ def buildCtx(selection: Option[ColumnSelection]): TestGraphRegistrationContext =
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("id", "name", "email", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ columnSelection = selection,
+ scdType = ScdType.Type2,
+ // Broadening `columnSelection` would also widen the default (selection-derived) tracked
+ // set, which is rejected as TRACK_HISTORY_DRIFT (SPARK-58391). Pin it to `name` -- selected
+ // in both runs -- so this test covers only the column-selection axis.
+ trackHistorySelection =
+ Option(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name")))))
+
+ // Run #1: only (id, name, version) selected; `email` is dropped before the MERGE.
+ stream.addData((1, "alice", "ignored", 1L))
+ runPipeline(buildCtx(selection = Some(ColumnSelection.IncludeColumns(
+ Seq("id", "name", "version").map(UnqualifiedColumnName(_))
+ ))))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, 1L, null, scd2Meta(1L)))
+ )
+
+ // Run #2: broaden to no selection for a new key id=2. mergeSchemas adds `email`; the existing
+ // id=1 row gets NULL, the new row gets the actual value.
+ stream.addData((2, "bob", "b@x.com", 2L))
+ runPipeline(buildCtx(selection = None))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 1L, 1L, null, scd2Meta(1L), null),
+ Row(2, "bob", 2L, 2L, null, scd2Meta(2L), "b@x.com")
+ )
+ )
+ }
+
+ test("a source DF column whose name differs from the target only by case folds onto the " +
+ "target column under case-insensitive resolution") {
+ // SPARK-58517: schema evolution honors `spark.sql.caseSensitive`, so a source `Value` maps onto
+ // the target's existing `value` instead of evolving the target to carry both spellings. The
+ // target keeps its own spelling (the merge is left-biased) and no new column appears.
+ //
+ // Before that fix the merge ran case-sensitively and the target gained a second, case-differing
+ // column -- a schema self-inconsistent under the case-insensitive resolver. The breakage
+ // surfaced later, during microbatch reconciliation rather than at table creation:
+ // `Scd2ForeachBatchHandler.reconcileMicrobatch` read the two-column target back into its
+ // affected-rows `unionByName`, where `ResolveUnion`'s case-insensitive duplicate check reported
+ // COLUMN_ALREADY_EXISTS.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(key INT NOT NULL, version BIGINT NOT NULL, value STRING, $scd2MetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, Long, String)]
+ stream.addData((1, 1L, "alice"))
+ val df = stream.toDF().toDF("key", "version", "Value")
+ val ctx = singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = df,
+ keys = Seq("key"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ runPipeline(ctx)
+
+ // A single `value` column, spelled as the target declared it -- not a second `Value`.
+ assert(
+ spark.table(s"$catalog.$namespace.target").schema.fieldNames.count(
+ _.equalsIgnoreCase("value")) === 1)
+ assert(spark.table(s"$catalog.$namespace.target").schema.fieldNames.contains("value"))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, 1L, "alice", 1L, null, scd2Meta(1L)))
+ )
+ }
+ }
+
+ test("changing a non-key column type from TIMESTAMP to STRING between runs fails with " +
+ "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(key INT NOT NULL, version BIGINT NOT NULL, value TIMESTAMP, $scd2MetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Long, Timestamp)]
+ stream1.addData((1, 1L, Timestamp.valueOf("2024-01-01 10:00:00")))
+ runPipeline(singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream1.toDF().toDF("key", "version", "value"),
+ keys = Seq("key"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2))
+
+ // Run #2 emits `value` as STRING. mergeSchemas rejects the type change.
+ val stream2 = MemoryStream[(Int, Long, String)]
+ stream2.addData((1, 2L, "2024-01-02 11:00:00"))
+ val ctx2 = singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream2.toDF().toDF("key", "version", "value"),
+ keys = Seq("key"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2)
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE",
+ sqlState = Some("42825"),
+ parameters = Map(
+ "left" -> "\"TIMESTAMP\"",
+ "right" -> "\"STRING\""
+ )
+ )
+ }
+}
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala
new file mode 100644
index 000000000000..ff4e90311740
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{AutoCdcReservedNames, Scd2BatchProcessor, ScdType}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests covering SCD Type 2 AutoCDC's behavior when the target table is pre-populated by something
+ * other than a prior AutoCDC run: hand-loaded open ("current") records and a target created
+ * without the framework columns. These verify AutoCDC interoperates gracefully with users who
+ * hand-populate the target. The SCD2 analog of [[AutoCdcScd1TargetTableDurabilitySuite]].
+ */
+class AutoCdcScd2TargetTableDurabilitySuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ import testImplicits._
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ /** Create an SCD2 target with user columns `(id, name, version)` plus the framework columns. */
+ private def createScd2Target(table: String): Unit = {
+ spark.sql(
+ s"CREATE TABLE $table (" +
+ s"id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ }
+
+ /**
+ * Insert a pre-existing open ("current") SCD2 record into a target table, as if a previous
+ * AutoCDC run had opened it at sequencing version [[sequence]]: `__START_AT` = `sequence`,
+ * `__END_AT` = NULL (still active), and `_cdc_metadata.__RECORD_START_AT` = `sequence`.
+ *
+ * @param table Fully-qualified table name (catalog.schema.table).
+ * @param colValues Comma-separated SQL literals for the user-defined columns, in declared
+ * order, excluding the trailing framework columns.
+ * @param sequence Value to seed the interval start and the record-start-at with.
+ */
+ private def insertPreloadedCurrentRecord(
+ table: String, colValues: String, sequence: Long): Unit = {
+ val recordStartAt = Scd2BatchProcessor.recordStartAtFieldName
+ spark.sql(
+ s"INSERT INTO $table SELECT $colValues, " +
+ s"CAST($sequence AS BIGINT), CAST(NULL AS BIGINT), " +
+ s"named_struct('$recordStartAt', CAST($sequence AS BIGINT))"
+ )
+ }
+
+ /**
+ * Insert a pre-existing closed (historical) SCD2 record into a target table, as if a previous
+ * AutoCDC run had opened it at [[startAt]] and later closed it at [[endAt]]: `__START_AT` =
+ * `startAt`, `__END_AT` = `endAt` (no longer active), and `_cdc_metadata.__RECORD_START_AT` =
+ * `startAt`.
+ *
+ * @param table Fully-qualified table name (catalog.schema.table).
+ * @param colValues Comma-separated SQL literals for the user-defined columns, in declared
+ * order, excluding the trailing framework columns.
+ * @param startAt Interval start (and record-start-at) of the pre-existing record.
+ * @param endAt Interval end (exclusive) at which the pre-existing record was closed.
+ */
+ private def insertPreloadedClosedRecord(
+ table: String, colValues: String, startAt: Long, endAt: Long): Unit = {
+ val recordStartAt = Scd2BatchProcessor.recordStartAtFieldName
+ spark.sql(
+ s"INSERT INTO $table SELECT $colValues, " +
+ s"CAST($startAt AS BIGINT), CAST($endAt AS BIGINT), " +
+ s"named_struct('$recordStartAt', CAST($startAt AS BIGINT))"
+ )
+ }
+
+ test("pre-loaded current record: a higher-sequence upsert closes it and opens a new record") {
+ createScd2Target(s"$catalog.$namespace.target")
+ insertPreloadedCurrentRecord(s"$catalog.$namespace.target", "1, 'alice', 5", 5L)
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alicia", 10L)) // > pre-existing seq=5 -> closes it, opens a new record
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+ runPipeline(ctx)
+
+ // The pre-existing record is closed at the incoming event's sequence; the new value is open.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 5L, 5L, 10L, scd2Meta(5L)),
+ Row(1, "alicia", 10L, 10L, null, scd2Meta(10L))
+ )
+ )
+ }
+
+ test("pre-loaded current record: a lower-sequence upsert is woven in as a closed prior record") {
+ createScd2Target(s"$catalog.$namespace.target")
+ insertPreloadedCurrentRecord(s"$catalog.$namespace.target", "1, 'alice', 10", 10L)
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "early", 5L)) // < pre-existing seq=10 -> closed prior record ending at 10
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+ runPipeline(ctx)
+
+ // Unlike SCD1, the late lower-sequence event is not suppressed: it becomes a closed prior
+ // record ending where the pre-existing record starts, which stays the open current record.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "early", 5L, 5L, 10L, scd2Meta(5L)),
+ Row(1, "alice", 10L, 10L, null, scd2Meta(10L))
+ )
+ )
+ }
+
+ test("pre-loaded closed record: an event landing inside its interval bisects it") {
+ // The interop shape unique to SCD2: a hand-loaded *closed* record -- a target row with no aux
+ // counterpart -- split by an event landing inside its interval. Pre-load [5, 20) and feed an
+ // event at seq=10. The pre-existing record is closed early, at 10, and the incoming event
+ // takes over the remainder of the span, [10, 20), so the two records partition the original
+ // interval with no row left open.
+ createScd2Target(s"$catalog.$namespace.target")
+ insertPreloadedClosedRecord(s"$catalog.$namespace.target", "1, 'alice', 5", startAt = 5L,
+ endAt = 20L)
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "mid", 10L)) // 5 < 10 < 20 -> bisects the pre-existing [5, 20) record
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+ runPipeline(ctx)
+
+ // The pre-existing record is split at the incoming event's sequence: its value carries into
+ // [5, 10), the incoming "mid" opens [10, 20), and no row remains open (endAt=20 was the
+ // pre-existing closure).
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 5L, 5L, 10L, scd2Meta(5L)),
+ Row(1, "mid", 10L, 10L, 20L, scd2Meta(10L))
+ )
+ )
+ }
+
+ test("pre-loaded target rows merge correctly on the first AutoCDC run, and the " +
+ "auxiliary table is created lazily") {
+ // Target was populated by some external process; this is the first AutoCDC run.
+ createScd2Target(s"$catalog.$namespace.target")
+ insertPreloadedCurrentRecord(s"$catalog.$namespace.target", "1, 'alice', 1", 1L)
+
+ assert(
+ !spark.catalog.tableExists(auxTableNameFor("target")),
+ "Auxiliary table should not exist before the first AutoCDC run"
+ )
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "bob", 2L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+ runPipeline(ctx)
+
+ // seq=2 > pre-existing seq=1, so the pre-existing record closes at 2 and "bob" opens.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 1L, 1L, 2L, scd2Meta(1L)),
+ Row(1, "bob", 2L, 2L, null, scd2Meta(2L))
+ )
+ )
+ assert(
+ spark.catalog.tableExists(auxTableNameFor("target")),
+ "Auxiliary table should be created lazily on the first AutoCDC run"
+ )
+ }
+
+ test("a target table created without the framework columns gets them " +
+ "auto-added on the first AutoCDC run") {
+ // User creates the target without the AutoCDC framework columns. DatasetManager evolves the
+ // existing table schema by merging it with the AutoCdcMergeFlow's output schema, which
+ // includes __START_AT / __END_AT and the metadata column. The first run therefore proceeds
+ // normally, and subsequent reads see the framework columns alongside the user's data columns.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL)"
+ )
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 1L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+ runPipeline(ctx)
+
+ val schema = spark.table(s"$catalog.$namespace.target").schema
+ Seq(
+ Scd2BatchProcessor.startAtColName,
+ Scd2BatchProcessor.endAtColName,
+ AutoCdcReservedNames.cdcMetadataColName
+ ).foreach { col =>
+ assert(
+ schema.fieldNames.contains(col),
+ s"Target must have $col after first AutoCDC run; got ${schema.fieldNames.toSeq}"
+ )
+ }
+ // Schema evolution appends the framework columns after the user columns in the flow's output
+ // order (__START_AT, __END_AT, then the metadata column -- the same order scd2MetadataDdl
+ // declares for a pre-created target). Assert by name so the row matches regardless of the
+ // physical column order.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target").select(
+ "id", "name", "version",
+ Scd2BatchProcessor.startAtColName,
+ Scd2BatchProcessor.endAtColName,
+ AutoCdcReservedNames.cdcMetadataColName
+ ),
+ Seq(Row(1, "alice", 1L, 1L, null, scd2Meta(1L)))
+ )
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org