You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "szehon-ho (via GitHub)" <gi...@apache.org> on 2026/08/01 00:46:25 UTC
Re: [PR] [SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites [spark]
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