[PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew opened a new pull request, #57689:
URL: https://github.com/apache/spark/pull/57689

   ### 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` | 4 | Interop with a hand-populated target: pre-loaded open records closed/opened by later upserts, 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` | 7 | Recorded per-key history persists across runs; dry run does not provision the aux table; a dropped aux table is transparently recreated. |
     | `AutoCdcScd2KeyDriftSuite` | 13 | Key add/drop/swap, dataType drift, order/backtick/nullability invariance, case-sensitive vs case-insensitive resolution, and tampered-aux failure modes. |
     | `AutoCdcScd2SchemaEvolutionSuite` | 8 | Nullable-column merge, new top-level column, additive target-column evolution, broadening the column selection, incompatible type changes, and a case-only source-column rename. |
     
   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 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 (40 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)           


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3705109822


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -314,10 +386,13 @@ class AutoCdcScd2SchemaEvolutionSuite
     import session.implicits._
 
     // DatasetManager's schema-merge compares schemas case-sensitively, so a target `value` and a
-    // source `Value` are treated as distinct and the merge tries to add `Value` alongside the
-    // existing `value`. Under case-insensitive resolution that collides, and (unlike SCD1, which
-    // surfaces AMBIGUOUS_REFERENCE deeper in the MERGE plan) the SCD2 write path reports
-    // COLUMN_ALREADY_EXISTS when adding the duplicate column.
+    // source `Value` are treated as distinct and the target is evolved to carry both columns.

Review Comment:
   And the corresponding SCD1 test also merely asserts the bug and not the correct bahavior!



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["szehon-ho (via GitHub)" <gi...@apache.org>]

szehon-ho commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3755553312


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    // The interop shape unique to SCD2: a hand-loaded *closed* record (a target row with no aux
+    // counterpart) reaches decomposeOutOfOrderRows with a target row that must be split. Pre-load
+    // [__START_AT=5, __END_AT=20); feed an event at seq=10, strictly inside that interval. The
+    // event bisects the pre-existing record into a head [5, 10) and a tail [10, 20): the head
+    // closes where the incoming event opens, the incoming event closes where the tail begins, and
+    // the redundant closed upsert the event itself would form drops out against the tail.

Review Comment:
   The outcome this describes is right -- you do end up with `[5, 10)` and `[10, 20)` -- but the mechanism isn't what runs.
   
   `decomposeOutOfOrderRows` splits the pre-loaded row into a head `[5, null)` and a synthetic tail `[null, 20)` carrying a null `recordStartAt`, not into `[5, 10)` and `[10, 20)`; those are the post-reconciliation intervals, and `[10, 20)` is the incoming event's own row rather than the tail. The tail has no `startAt` at all, so "the incoming event closes where the tail begins" isn't describing anything -- the event closes at 20, which is the boundary the tail carries, reached via `next.effectiveRecordStartAt` falling back to `endAt`. And the drop goes the other way: `dropLeftoverDeletesPostReconciliation` drops the *tail*, because its immediately preceding upsert (the event, now `[10, 20)`) closes exactly on the tail's boundary. The event's row survives and is written to the target, as the expected rows below show.
   
   Rather than correct it, I'd just drop the internals. The observable behavior needs no head/tail vocabulary, and `Scd2BatchProcessor`'s scaladoc already documents decomposition in full -- restating it here is a second copy that can drift, which is how this one ended up inverted. Something like:
   
   ```scala
       // 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.
   ```



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,554 @@
+/*
+ * 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._

Review Comment:
   Could the new suites use a single class-level `import testImplicits._` instead of repeating `val session = spark` / `import session.implicits._` in every test? `QueryTest` provides `testImplicits` bound to `self.spark`, and `AutoCdcScd2KeyDriftSuite` in this PR already does exactly this -- same `MemoryStream[(Int, String, Long)]` construction, same `$"version"` -- so it's proven in this context.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2MultiPipelineSuite.scala:
##########
@@ -0,0 +1,315 @@
+/*
+ * 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._

Review Comment:
   Same as the note on `AutoCdcScd2SchemaEvolutionSuite`: a single class-level `import testImplicits._` would replace the per-test `val session = spark` / `import session.implicits._` here too.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._

Review Comment:
   Same as the note on `AutoCdcScd2SchemaEvolutionSuite`: a single class-level `import testImplicits._` would replace the per-test `val session = spark` / `import session.implicits._` here too.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2FullRefreshSuite.scala:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._

Review Comment:
   Same as the note on `AutoCdcScd2SchemaEvolutionSuite`: a single class-level `import testImplicits._` would replace the per-test `val session = spark` / `import session.implicits._` here too.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala:
##########
@@ -0,0 +1,398 @@
+/*
+ * 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._

Review Comment:
   Same as the note on `AutoCdcScd2SchemaEvolutionSuite`: a single class-level `import testImplicits._` would replace the per-test `val session = spark` / `import session.implicits._` here too.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["szehon-ho (via GitHub)" <gi...@apache.org>]

szehon-ho commented on PR #57689:
URL: https://github.com/apache/spark/pull/57689#issuecomment-5259185919

   **Merge Summary:**
   - merged into master https://github.com/apache/spark/commit/297a730f57edeea2a652604441310b5b65e85613
   - merged into branch-4.x https://github.com/apache/spark/commit/8a5c2b777586f52c517e57921ac5532a5daf8e2a
   - merged into branch-4.3 https://github.com/apache/spark/commit/bdbbfabe012ff3f9e7ddf34436bb14cceef08246
   
   *Posted by `merge_spark_pr.py`*


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3754184755


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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 {
+
+  /** 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))"
+    )
+  }
+
+  test("pre-loaded current record: a higher-sequence upsert closes it and opens a new record") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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 target rows merge correctly on the first AutoCDC run, and the " +
+    "auxiliary table is created lazily") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    // 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 (metadata, then the interval bounds), which differs from the declared order of a
+    // pre-created SCD2 target. Assert by name so the row matches regardless of physical order.

Review Comment:
   updated



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["szehon-ho (via GitHub)" <gi...@apache.org>]

szehon-ho commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3694150181


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,389 @@
+/*
+ * 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, broadening column selection) and the cases that fail loudly
+ * (incompatible type changes, case-only renames).
+ *
+ * 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.
+ *
+ * Note: the *narrowing* / dropped-column cases (a microbatch narrower than the already-evolved
+ * target, incl. dropped nested struct/array fields) are covered by
+ * [[AutoCdcScd2ColumnEvolutionSuite]] under SPARK-58418, which makes them reconcile correctly, so
+ * they are intentionally not duplicated here.
+ */
+class AutoCdcScd2SchemaEvolutionSuite
+    extends ExecutionTest
+    with SharedSparkSession
+    with AutoCdcGraphExecutionTestMixin {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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: 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("additive target-column evolution extends the SCD2 auxiliary table schema") {
+    val session = spark
+    import session.implicits._
+
+    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)
+    }
+
+    // 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("broadening the column selection between runs adds the newly-included column to " +
+    "the target") {

Review Comment:
   This covers broadening `columnSelection`. Is the SCD2-only analog — changing `trackHistorySelection` between runs — deliberately out of scope? It directly determines whether an upsert opens a new record, and its only end-to-end exercise today is in `AutoCdcScd2ColumnEvolutionSuite`. A line in the suite header noting the deferral would keep the parity claim precise.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,389 @@
+/*
+ * 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, broadening column selection) and the cases that fail loudly
+ * (incompatible type changes, case-only renames).
+ *
+ * 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.
+ *
+ * Note: the *narrowing* / dropped-column cases (a microbatch narrower than the already-evolved
+ * target, incl. dropped nested struct/array fields) are covered by
+ * [[AutoCdcScd2ColumnEvolutionSuite]] under SPARK-58418, which makes them reconcile correctly, so
+ * they are intentionally not duplicated here.

Review Comment:
   This exclusion is accurate for the drop cases, but it leaves an additive one uncovered: `AutoCdcScd1SchemaEvolutionSuite` has "a new field added inside an `array<struct>` element between runs is added to the target", and `AutoCdcScd2ColumnEvolutionSuite` covers only *dropped* nested struct/array fields. So no SCD2 suite exercises a nested type additively — everything here is top-level scalars. Same for SCD1's "extra columns on the target that the AutoCDC flow does not emit are preserved". Either add the additive nested case or narrow this note so it doesn't read as full parity.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala:
##########
@@ -0,0 +1,338 @@
+/*
+ * 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.{Scd2BatchProcessor, ScdType}
+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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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. The seq=1 record still lives
+    // in the target, and SCD2 reconciliation reads affected rows from the target as well as the
+    // aux table, so the seq=2 event closes the seq=1 record and opens a new one -- the recorded
+    // history survives the aux-table drop.

Review Comment:
   The last clause overstates what this test shows. Both records here are visible rows in the target, so dropping the aux table changes nothing observable — the aux contributes no state to this outcome. But the SCD2 aux exists precisely to hold rows that are *not* in the target (tombstones and coalesced no-op upserts, per `buildScd2AuxiliaryTableSpecFor`), and those don't survive a drop.
   
   Could we cover the case where the aux is the only holder of the state? Roughly the shape of `AutoCdcScd2FullRefreshSuite`'s second test: run #1 delete-only at seq=10 (target stays empty, aux holds the tombstone), drop the aux, then run #2 upsert at seq=5. With the aux the event lands closed at 10; without it, open. At minimum I'd soften the comment so a future reader doesn't take the aux as disposable.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala:
##########
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ */
+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')"

Review Comment:
   `validateNoScdTypeDrift` is reachable end-to-end from `DatasetManager`, right after `validateNoKeyColumnDrift`, but `SCD_TYPE_DRIFT` only has unit coverage in `AutoCdcAuxiliaryTableSuite`. The SCD1 suite structurally can't cover it, so this suite seems like the right home — and `scd2AuxProps` makes it cheap: pre-create the aux with `scdTypePropertyKey` set to `ScdType.Type1.label`, run an SCD2 flow, expect `AUTOCDC_INVALID_STATE.SCD_TYPE_DRIFT`.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,389 @@
+/*
+ * 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, broadening column selection) and the cases that fail loudly
+ * (incompatible type changes, case-only renames).
+ *
+ * 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.
+ *
+ * Note: the *narrowing* / dropped-column cases (a microbatch narrower than the already-evolved
+ * target, incl. dropped nested struct/array fields) are covered by
+ * [[AutoCdcScd2ColumnEvolutionSuite]] under SPARK-58418, which makes them reconcile correctly, so
+ * they are intentionally not duplicated here.
+ */
+class AutoCdcScd2SchemaEvolutionSuite
+    extends ExecutionTest
+    with SharedSparkSession
+    with AutoCdcGraphExecutionTestMixin {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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: 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("additive target-column evolution extends the SCD2 auxiliary table schema") {
+    val session = spark
+    import session.implicits._
+
+    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)
+    }
+
+    // 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("broadening the column selection between runs adds the newly-included column to " +
+    "the target") {
+    val session = spark
+    import session.implicits._
+
+    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)
+
+    // 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 fails with " +
+    "COLUMN_ALREADY_EXISTS under case-insensitive resolution") {
+    val session = spark
+    import session.implicits._
+
+    // DatasetManager's schema-merge compares schemas case-sensitively, so a target `value` and a
+    // source `Value` are treated as distinct and the merge tries to add `Value` alongside the
+    // existing `value`. Under case-insensitive resolution that collides, and (unlike SCD1, which
+    // surfaces AMBIGUOUS_REFERENCE deeper in the MERGE plan) the SCD2 write path reports
+    // COLUMN_ALREADY_EXISTS when adding the duplicate column.

Review Comment:
   Where exactly is `COLUMN_ALREADY_EXISTS` raised here? My reading is that it comes from creating the SCD2 aux table, whose schema mirrors the target and therefore contains both `value` and `Value`, tripping duplicate-column validation before the MERGE plan is analyzed — which would also explain why SCD1, with its keys-only aux, instead reaches `AMBIGUOUS_REFERENCE` deeper in. If that's right, "the SCD2 write path" is vague and naming the aux-table creation would make the comment durable.
   
   Separately: is it intended that the same user mistake surfaces two different error conditions depending on SCD type? Between this test and the SCD1 one we're now locking both in.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3705100305


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -314,10 +386,13 @@ class AutoCdcScd2SchemaEvolutionSuite
     import session.implicits._
 
     // DatasetManager's schema-merge compares schemas case-sensitively, so a target `value` and a
-    // source `Value` are treated as distinct and the merge tries to add `Value` alongside the
-    // existing `value`. Under case-insensitive resolution that collides, and (unlike SCD1, which
-    // surfaces AMBIGUOUS_REFERENCE deeper in the MERGE plan) the SCD2 write path reports
-    // COLUMN_ALREADY_EXISTS when adding the duplicate column.
+    // source `Value` are treated as distinct and the target is evolved to carry both columns.

Review Comment:
   Claude explains this carefully here. But this is a real bug, pre-existing in  SDP and unrelated to AutoCDC, it happens for all tables. Filed SPARK-58517 for this. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3754160442


##########
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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    createScd2Target(s"$catalog.$namespace.t_a")
+    createScd2Target(s"$catalog.$namespace.t_b")
+
+    // 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 watermark suppressing it).
+    val streamA1 = MemoryStream[(Int, String, Long)]
+    val streamB = MemoryStream[(Int, String, Long)]
+    streamA1.addData((1, "a", 10L))
+    streamB.addData((1, "b", 10L))
+    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(autoCdcFlow(
+        name = "flow_b",
+        target = "t_b",
+        query = dfFlowFunc(streamB.toDF().toDF("id", "name", "version")),
+        keys = Seq("id"),
+        sequencing = functions.col("version"),
+        scdType = ScdType.Type2
+      ))
+    }
+    runPipeline(ctx1)
+
+    // 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 current record). Unlike SCD1, SCD2 does not suppress a late
+    // lower-sequence event; it weaves it into history as a closed prior record ending at seq=10.
+    streamB.addData((1, "b2", 5L))
+    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(autoCdcFlow(
+        name = "flow_b",
+        target = "t_b",
+        query = dfFlowFunc(streamB.toDF().toDF("id", "name", "version")),
+        keys = Seq("id"),
+        sequencing = functions.col("version"),
+        scdType = ScdType.Type2
+      ))
+    }
+    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
+    // (endAt=10), and the pre-existing seq=10 record remains the open current record.

Review Comment:
   ok, changed it to start with a delete-only



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew commented on PR #57689:
URL: https://github.com/apache/spark/pull/57689#issuecomment-5247832333

   @szehon-ho I addressed your comments and updated the PR description.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3754169099


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,552 @@
+/*
+ * 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.
+ *
+ * Two SCD1 evolution cases have no SCD2 analog and so are 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    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,
+        // `name` is the only non-key data column and it is absent in run #1, so an empty tracked
+        // set is what keeps the set stable across the two runs (see the note in the preceding
+        // test). Every upsert then opens a new record on sequencing alone.
+        trackHistorySelection = Option(ColumnSelection.IncludeColumns(Seq.empty)))

Review Comment:
   The empty set of racking columns is indeed degenerate. I wonder whether that should even be legal. Changed this test to use the fixed singleton. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["szehon-ho (via GitHub)" <gi...@apache.org>]

szehon-ho closed pull request #57689:   [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites
URL: https://github.com/apache/spark/pull/57689


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["szehon-ho (via GitHub)" <gi...@apache.org>]

szehon-ho commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3752418219


##########
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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    createScd2Target(s"$catalog.$namespace.t_a")
+    createScd2Target(s"$catalog.$namespace.t_b")
+
+    // 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 watermark suppressing it).
+    val streamA1 = MemoryStream[(Int, String, Long)]
+    val streamB = MemoryStream[(Int, String, Long)]
+    streamA1.addData((1, "a", 10L))
+    streamB.addData((1, "b", 10L))
+    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(autoCdcFlow(
+        name = "flow_b",
+        target = "t_b",
+        query = dfFlowFunc(streamB.toDF().toDF("id", "name", "version")),
+        keys = Seq("id"),
+        sequencing = functions.col("version"),
+        scdType = ScdType.Type2
+      ))
+    }
+    runPipeline(ctx1)
+
+    // 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 current record). Unlike SCD1, SCD2 does not suppress a late
+    // lower-sequence event; it weaves it into history as a closed prior record ending at seq=10.
+    streamB.addData((1, "b2", 5L))
+    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(autoCdcFlow(
+        name = "flow_b",
+        target = "t_b",
+        query = dfFlowFunc(streamB.toDF().toDF("id", "name", "version")),
+        keys = Seq("id"),
+        sequencing = functions.col("version"),
+        scdType = ScdType.Type2
+      ))
+    }
+    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
+    // (endAt=10), and the pre-existing seq=10 record remains the open current record.

Review Comment:
   Could we make `t_b`'s run #1 delete-only, so the retained aux is the sole holder of the state? Then the target stays empty, the aux holds the seq=10 tombstone, and the seq=5 event landing closed at 10 is only possible because the selective refresh spared `t_b`'s aux -- which is what the test name claims.
   
   As written, run #1's open upsert is neither a tombstone nor a no-op continuation (it has no successor), so `identifyAndTagAuxRows` routes it to the target and leaves `t_b`'s aux empty. In run #2, `findAffectedRowsFromTargetTable` picks that row up because its `__END_AT` is null, so seq=5 is woven in from target state alone -- the assertion would hold identically had the aux been wiped. This comment attributes the outcome to the wrong table.
   
   Same shape as the aux-drop concern that the new aux-sole-holder test fixed.
   
   Lesser and related: the first test in this suite asserts nothing about the aux after the refresh either, but there the aux is empty either way, so there is nothing meaningful to assert -- the second test is what proves the wipe.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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]].

Review Comment:
   Consider adding a pre-loaded *closed* record here, or confirming it was deliberately left out. That's the interop shape unique to SCD2, and the one that reaches `decomposeOutOfOrderRows` with a target row that has no aux counterpart; the suite covers only hand-loaded open records today.
   
   The outcome is deterministic: pre-load `[__START_AT=5, __END_AT=20)` and feed an event at 10, and you get a head/tail split at 5, the event closing at 20, the tail dropping as redundant against it, leaving `[5,10)` and `[10,20)`.
   
   `AutoCdcOutOfOrderConvergenceSuite` reaches bisection randomly, but only through AutoCDC-written state -- which is precisely what this suite exists to sidestep.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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 {
+
+  /** 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))"
+    )
+  }
+
+  test("pre-loaded current record: a higher-sequence upsert closes it and opens a new record") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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 target rows merge correctly on the first AutoCDC run, and the " +
+    "auxiliary table is created lazily") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    // 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 (metadata, then the interval bounds), which differs from the declared order of a
+    // pre-created SCD2 target. Assert by name so the row matches regardless of physical order.

Review Comment:
   Could we drop this ordering claim, or correct it to `__START_AT`, `__END_AT`, then metadata -- which is the same as the declared order, not different from it?
   
   `AutoCdcMergeFlow.schema`'s Type2 branch emits `__START_AT`, `__END_AT`, then the metadata column, and `DatasetManager` merges via `SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)` -> `StructType.merge`, which is left-biased and appends new fields in right-hand order. So the evolved order matches what `scd2MetadataDdl` declares, and both halves of the comment are off. The select-by-name keeps the assertion order-independent, so only the comment is affected.
   
   (The analogous comments about `email` and `age` landing after the framework columns, in the schema-evolution and multi-pipeline suites, are correct.)



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,552 @@
+/*
+ * 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.
+ *
+ * Two SCD1 evolution cases have no SCD2 analog and so are 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    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,
+        // `name` is the only non-key data column and it is absent in run #1, so an empty tracked
+        // set is what keeps the set stable across the two runs (see the note in the preceding
+        // test). Every upsert then opens a new record on sequencing alone.
+        trackHistorySelection = Option(ColumnSelection.IncludeColumns(Seq.empty)))

Review Comment:
   Pinning to `IncludeColumns(Seq("version"))` instead of the empty set would satisfy the same constraint this comment is reaching for -- `version` is an eligible non-key column present in both runs, so the tracked set stays stable -- while keeping the semantics the comment actually describes.
   
   "Every upsert then opens a new record on sequencing alone" is the opposite of what an empty tracked set does. In `reconcileStartAndEndAt`, an empty set collapses `areTrackedColumnsEqual` to `lit(true)`, so every consecutive upsert pair becomes a no-op continuation and coalesces into one run; the scaladoc there calls this "the correct degenerate behavior when the user tracks nothing." The test passes because its two runs use different keys, so no continuation ever forms -- but anyone adding a same-key event would be steered wrong.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,552 @@
+/*
+ * 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.
+ *
+ * Two SCD1 evolution cases have no SCD2 analog and so are 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.

Review Comment:
   Either name the second case or change the count to one. Walking `AutoCdcScd1SchemaEvolutionSuite`'s fourteen tests against this suite's ten plus the four deferred to `AutoCdcScd2ColumnEvolutionSuite` leaves exactly one unaccounted for. Only "extra columns on the target that the AutoCDC flow does not emit are preserved" is named here; the clause after it is that case's justification, not a second case.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala:
##########
@@ -0,0 +1,481 @@
+/*
+ * 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

Review Comment:
   A line noting that the pipeline-case-sensitivity case is deliberately skipped would keep this parity claim precise. `AutoCdcScd1KeyDriftSuite`'s "AutoCDC key drift validation uses pipeline case sensitivity, not session default" -- which pins that the validator reads the pipeline's conf rather than the ambient session conf -- has no mirror here. Both suites land at fourteen tests because that one is dropped and `SCD_TYPE_DRIFT` is added. Skipping it is defensible for exactly the reason stated here (validation is SCD-type-agnostic); it's only the unqualified "these mirror the SCD1 cases" that overshoots.



##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala:
##########
@@ -0,0 +1,396 @@
+/*
+ * 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    // 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 (marker column set), with no matching visible target row.

Review Comment:
   "(marker column set)" reads as though the tombstone carries a set marker, which would mean the opposite of what the test needs -- worth rewording. `mergeRowsIntoAuxiliaryTable` writes live aux rows with `deletedByBatchId = null`; a non-null value is what marks a row logically deleted.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3754190987


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala:
##########
@@ -0,0 +1,552 @@
+/*
+ * 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.
+ *
+ * Two SCD1 evolution cases have no SCD2 analog and so are 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.

Review Comment:
   fixed to one



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["anew (via GitHub)" <gi...@apache.org>]

anew commented on code in PR #57689:
URL: https://github.com/apache/spark/pull/57689#discussion_r3754196187


##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala:
##########
@@ -0,0 +1,396 @@
+/*
+ * 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 {
+
+  /** 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    // 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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    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") {
+    val session = spark
+    import session.implicits._
+
+    // 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 (marker column set), with no matching visible target row.

Review Comment:
   updated



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]

["uros-b (via GitHub)" <gi...@apache.org>]

uros-b commented on PR #57689:
URL: https://github.com/apache/spark/pull/57689#issuecomment-5147625251

   Thank you @anew!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org