From 751d8217d0941ff533196135160906542d75fbfe Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Wed, 29 Jul 2026 07:49:01 +0000 Subject: [PATCH 1/8] [SPARK-58409][SDP] Add AutoCdcScd2FullRefreshSuite SCD2 analog of AutoCdcScd1FullRefreshSuite: full refresh wipes both the target rows and the richer SCD2 auxiliary table for the refreshed flow, resets the watermark so a previously-suppressed lower-sequence event lands, and under selective refresh leaves a non-refreshed target's aux state intact. The selective test also documents the SCD2-specific behavior that a non-refreshed target weaves a late lower-sequence event into history as a closed prior record rather than suppressing it as SCD1 does. Co-authored-by: Opus 4.8 --- .../graph/AutoCdcScd2FullRefreshSuite.scala | 263 ++++++++++++++++++ 1 file changed, 263 insertions(+) create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2FullRefreshSuite.scala diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2FullRefreshSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2FullRefreshSuite.scala new file mode 100644 index 000000000000..f7113258e8a1 --- /dev/null +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2FullRefreshSuite.scala @@ -0,0 +1,263 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.pipelines.graph + +import org.apache.spark.sql.Row +import org.apache.spark.sql.execution.streaming.runtime.MemoryStream +import org.apache.spark.sql.functions +import org.apache.spark.sql.pipelines.autocdc.{ + ColumnSelection, + ScdType, + UnqualifiedColumnName +} +import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext} +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Tests covering AutoCDC's full-refresh semantics for SCD Type 2 targets: full refresh must wipe + * both the target rows and the (richer) SCD2 auxiliary table for the refreshed targets, and must + * leave non-refreshed targets untouched in selective-refresh mode. The SCD2 analog of + * [[AutoCdcScd1FullRefreshSuite]]. + */ +class AutoCdcScd2FullRefreshSuite + extends ExecutionTest + with SharedSparkSession + with AutoCdcGraphExecutionTestMixin { + + /** 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. + checkAnswer( + spark.table(s"$catalog.$namespace.t_b"), + Seq( + Row(1, "b2", 5L, 5L, 10L, scd2Meta(5L)), + Row(1, "b", 10L, 10L, null, scd2Meta(10L)) + ) + ) + } +} From aadd9b35cfa203670772ec89a9f74862bd99689a Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Wed, 29 Jul 2026 07:54:27 +0000 Subject: [PATCH 2/8] [SPARK-58409][SDP] Add AutoCdcScd2TargetTableDurabilitySuite SCD2 analog of AutoCdcScd1TargetTableDurabilitySuite, covering interop with a hand-populated target: a pre-loaded open current record closed/opened by a higher-sequence upsert, a lower-sequence upsert woven in as a closed prior record (SCD2-specific, no SCD1 equivalent), lazy auxiliary-table creation on the first run over a pre-loaded target, and framework-column auto-add when the target is created without __START_AT / __END_AT / _cdc_metadata. Adds a suite-local helper to seed a pre-existing open SCD2 record. Co-authored-by: Opus 4.8 --- ...utoCdcScd2TargetTableDurabilitySuite.scala | 234 ++++++++++++++++++ 1 file changed, 234 insertions(+) create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2TargetTableDurabilitySuite.scala new file mode 100644 index 000000000000..680f242f6647 --- /dev/null +++ b/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. + checkAnswer( + spark.table(s"$catalog.$namespace.target").select( + "id", "name", "version", + Scd2BatchProcessor.startAtColName, + Scd2BatchProcessor.endAtColName, + AutoCdcReservedNames.cdcMetadataColName + ), + Seq(Row(1, "alice", 1L, 1L, null, scd2Meta(1L))) + ) + } +} From 37a79296e512849b223846ab6fe938d441b11c80 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Wed, 29 Jul 2026 08:05:18 +0000 Subject: [PATCH 3/8] [SPARK-58409][SDP] Add AutoCdcScd2MultiPipelineSuite SCD2 analog of AutoCdcScd1MultiPipelineSuite: independent target/auxiliary tables per target across pipelines, a downstream materialized view reading an SCD2 target without the framework columns, two pipelines merging into a shared SCD2 target, cross-pipeline schema evolution, and KEY_SCHEMA_DRIFT rejection when a second pipeline uses different keys. The schema-evolution test stops short of re-running the narrower pipeline against the widened target (which the SCD1 analog does): that path currently fails with NUM_COLUMNS_MISMATCH because Scd2ForeachBatchHandler unions the microbatch with affected rows without allowMissingColumns. Tracked as a separate bug (SPARK-58418). Co-authored-by: Opus 4.8 --- .../graph/AutoCdcScd2MultiPipelineSuite.scala | 307 ++++++++++++++++++ 1 file changed, 307 insertions(+) create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2MultiPipelineSuite.scala diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2MultiPipelineSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2MultiPipelineSuite.scala new file mode 100644 index 000000000000..329843f846dc --- /dev/null +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2MultiPipelineSuite.scala @@ -0,0 +1,307 @@ +/* + * 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.ScdType +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._ + + // Two distinct target tables created up-front. + spark.sql( + s"CREATE TABLE $catalog.$namespace.t_a " + + s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + spark.sql( + s"CREATE TABLE $catalog.$namespace.t_b " + + s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1 only knows about `t_a`. Its auxiliary table must not affect pipeline #2's `t_b`. + val streamA = MemoryStream[(Int, String, Long)] + streamA.addData((1, "alice", 100L)) + runPipeline(singleAutoCdcFlowPipeline( + flowName = "flow_a", + target = "t_a", + sourceDf = streamA.toDF().toDF("id", "name", "version"), + keys = Seq("id"), + sequencing = $"version", + scdType = ScdType.Type2)) + + // Pipeline #2 only knows about `t_b`. Uses a deliberately *lower* sequence to verify the + // watermark from pipeline #1's auxiliary table (seq=100) does not leak into pipeline #2. + val streamB = MemoryStream[(Int, String, Long)] + streamB.addData((9, "bob", 1L)) + runPipeline(singleAutoCdcFlowPipeline( + flowName = "flow_b", + target = "t_b", + sourceDf = streamB.toDF().toDF("id", "name", "version"), + keys = Seq("id"), + sequencing = $"version", + scdType = ScdType.Type2)) + + checkAnswer( + spark.table(s"$catalog.$namespace.t_a"), + Seq(Row(1, "alice", 100L, 100L, null, scd2Meta(100L))) + ) + checkAnswer( + spark.table(s"$catalog.$namespace.t_b"), + Seq(Row(9, "bob", 1L, 1L, null, scd2Meta(1L))) + ) + + // Each target has its own auxiliary table; no cross-contamination. + assert(spark.catalog.tableExists(auxTableNameFor("t_a"))) + assert(spark.catalog.tableExists(auxTableNameFor("t_b"))) + } + + test("a downstream pipeline can read an AutoCDC target written by a different pipeline " + + "without observing the framework columns") { + val session = spark + import session.implicits._ + + // Pipeline #1 writes into target `src` via AutoCDC. + spark.sql( + s"CREATE TABLE $catalog.$namespace.src " + + s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + val stream = MemoryStream[(Int, String, Long)] + stream.addData((1, "alice", 1L), (2, "bob", 1L)) + runPipeline(singleAutoCdcFlowPipeline( + flowName = "writer", + target = "src", + sourceDf = stream.toDF().toDF("id", "name", "version"), + keys = Seq("id"), + sequencing = $"version", + scdType = ScdType.Type2)) + + // Pipeline #2 is a regular materialized view that selects the user-data columns from `src` + // (a different graph entirely). It must observe the merged AutoCDC rows and be able to ignore + // the framework columns without them polluting downstream consumers. + val ctxReader = new TestGraphRegistrationContext(spark) { + registerMaterializedView( + "downstream_mv", + query = dfFlowFunc( + spark.read.table(s"$catalog.$namespace.src").select("id", "name", "version") + ) + ) + } + runPipeline(ctxReader) + + checkAnswer( + spark.table(fullyQualifiedIdentifier("downstream_mv").toString), + Seq(Row(1, "alice", 1L), Row(2, "bob", 1L)) + ) + } + + test("two AutoCDC pipelines targeting the same table with identical key and data " + + "schemas merge into a shared target table") { + val session = spark + import session.implicits._ + + // Target table is created once up-front; both pipelines target it with the same AutoCDC + // `keys` and the same source-DF data schema. The two pipelines have distinct flow names so + // they own independent streaming checkpoints, but share the target and its auxiliary table. + spark.sql( + s"CREATE TABLE $catalog.$namespace.shared_target " + + s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1: inserts rows with id=1 and id=2 at version=1. + val stream1 = MemoryStream[(Int, String, Long)] + stream1.addData((1, "alice", 1L), (2, "bob", 1L)) + runPipeline(singleAutoCdcFlowPipeline( + flowName = "flow_v1", + target = "shared_target", + sourceDf = stream1.toDF().toDF("id", "name", "version"), + keys = Seq("id"), + sequencing = $"version", + scdType = ScdType.Type2)) + + // Sanity-check pipeline #1's effect before pipeline #2 runs. + checkAnswer( + spark.table(s"$catalog.$namespace.shared_target"), + Seq( + Row(1, "alice", 1L, 1L, null, scd2Meta(1L)), + Row(2, "bob", 1L, 1L, null, scd2Meta(1L)) + ) + ) + + // Pipeline #2: updates id=2 (existing key) to a higher sequence and inserts id=3 (new key). + // id=1 is untouched and must survive into the final target unchanged. + val stream2 = MemoryStream[(Int, String, Long)] + stream2.addData((2, "bob-v2", 2L), (3, "carol", 1L)) + runPipeline(singleAutoCdcFlowPipeline( + flowName = "flow_v2", + target = "shared_target", + sourceDf = stream2.toDF().toDF("id", "name", "version"), + keys = Seq("id"), + sequencing = $"version", + scdType = ScdType.Type2)) + + // Final target: id=1 untouched; id=2's original record closed at seq=2 with a new open record; + // id=3 freshly inserted by pipeline #2. + checkAnswer( + spark.table(s"$catalog.$namespace.shared_target"), + Seq( + Row(1, "alice", 1L, 1L, null, scd2Meta(1L)), + Row(2, "bob", 1L, 1L, 2L, scd2Meta(1L)), + Row(2, "bob-v2", 2L, 2L, null, scd2Meta(2L)), + Row(3, "carol", 1L, 1L, null, scd2Meta(1L)) + ) + ) + + // The auxiliary table for the shared target is itself shared across both pipelines. + assert(spark.catalog.tableExists(auxTableNameFor("shared_target"))) + } + + test("two AutoCDC pipelines targeting the same table with the same key but different " + + "data columns evolve the shared target schema") { + val session = spark + import session.implicits._ + + // Target is created up-front with pipeline #1's schema only; pipeline #2 brings a new + // top-level nullable `age` column that the dataset materialization layer is expected to + // schema-merge into the target. + spark.sql( + s"CREATE TABLE $catalog.$namespace.shared_target " + + s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1: source DF schema is (id, name, version); inserts id=1 and id=2. + val stream1 = MemoryStream[(Int, String, Long)] + stream1.addData((1, "alice", 1L), (2, "bob", 1L)) + val ctx1 = singleAutoCdcFlowPipeline( + flowName = "flow_v1", + target = "shared_target", + sourceDf = stream1.toDF().toDF("id", "name", "version"), + keys = Seq("id"), + sequencing = $"version", + scdType = ScdType.Type2) + runPipeline(ctx1) + + // Sanity-check pipeline #1's state before schema evolution kicks in. + checkAnswer( + spark.table(s"$catalog.$namespace.shared_target"), + Seq( + Row(1, "alice", 1L, 1L, null, scd2Meta(1L)), + Row(2, "bob", 1L, 1L, null, scd2Meta(1L)) + ) + ) + + // Pipeline #2: source DF schema is (id, name, age, version). The new nullable `age` column + // should be added to the target by dataset materialization; pipeline #1's untouched id=1 row + // is backfilled to NULL. The `age` column lands after the framework columns in the target. + val stream2 = MemoryStream[(Int, String, Option[Int], Long)] + stream2.addData((2, "bob-v2", Some(25), 2L), (3, "carol", Some(30), 1L)) + runPipeline(singleAutoCdcFlowPipeline( + flowName = "flow_v2", + target = "shared_target", + sourceDf = stream2.toDF().toDF("id", "name", "age", "version"), + keys = Seq("id"), + sequencing = $"version", + scdType = ScdType.Type2)) + + checkAnswer( + spark.table(s"$catalog.$namespace.shared_target"), + Seq( + Row(1, "alice", 1L, 1L, null, scd2Meta(1L), null), + Row(2, "bob", 1L, 1L, 2L, scd2Meta(1L), null), + Row(2, "bob-v2", 2L, 2L, null, scd2Meta(2L), 25), + Row(3, "carol", 1L, 1L, null, scd2Meta(1L), 30) + ) + ) + + // NOTE: the SCD1 analog of this test additionally re-runs the narrower pipeline #1 against the + // now-wider evolved target. For SCD2 that microbatch-narrower-than-target path is covered + // separately by AutoCdcScd2ColumnEvolutionSuite (SPARK-58418), so it is not duplicated here. + } + + test("a second pipeline targeting an existing AutoCDC table with different keys " + + "fails with KEY_SCHEMA_DRIFT") { + val session = spark + import session.implicits._ + + // Target table with both candidate keys present so the second pipeline would otherwise be + // schema-compatible with the first; only the AutoCDC `keys` differ between flows. + spark.sql( + s"CREATE TABLE $catalog.$namespace.shared_target " + + s"(id INT NOT NULL, name STRING NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1: AutoCDC flow keyed on `id`. + val stream1 = MemoryStream[(Int, String, Long)] + stream1.addData((1, "alice", 1L)) + runPipeline(singleAutoCdcFlowPipeline( + flowName = "flow_v1", + target = "shared_target", + sourceDf = stream1.toDF().toDF("id", "name", "version"), + keys = Seq("id"), + sequencing = $"version", + scdType = ScdType.Type2)) + + // Pipeline #2: completely separate graph, but targets the same physical `shared_target` + // table with `keys = Seq("name")`. + val stream2 = MemoryStream[(Int, String, Long)] + stream2.addData((2, "alice", 1L)) + val ctx2 = singleAutoCdcFlowPipeline( + flowName = "flow_v2", + target = "shared_target", + sourceDf = stream2.toDF().toDF("id", "name", "version"), + keys = Seq("name"), + sequencing = $"version", + scdType = ScdType.Type2) + + val ex = intercept[RuntimeException] { runPipeline(ctx2) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("shared_target", Some(catalog), Some(namespace)).unquotedString, + // Pipeline #2's AutoCDC key resolves from the source DF, where `MemoryStream[(Int, String, + // Long)]` produces a nullable StringType for `name`. + "expectedKeySchema" -> "name STRING", + // Pipeline #1 persisted the aux table from a source DF whose `id` was a non-null Scala + // primitive (`Int`), so the recorded key carries `NOT NULL`. + "recordedKeySchema" -> "id INT NOT NULL" + ) + ) + } +} From 8cff6d4304720cfc5dbb540fa39072fc28c25b16 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Wed, 29 Jul 2026 08:15:48 +0000 Subject: [PATCH 4/8] [SPARK-58409][SDP] Add AutoCdcScd2AuxiliaryTableDurabilitySuite SCD2 analog of AutoCdcScd1AuxiliaryTableDurabilitySuite: recorded per-key history persists across incremental runs, a dry run does not provision the aux table, and a dropped aux table is transparently recreated. Schema-layout assertions reflect the SCD2 aux table (full target row schema plus the aux-only deleted-by-batch-id marker) rather than SCD1's keys+metadata, and the lower-sequence and dropped-aux cases document SCD2's history-preserving behavior (closed prior record) where SCD1 suppresses/overwrites. Co-authored-by: Opus 4.8 --- ...CdcScd2AuxiliaryTableDurabilitySuite.scala | 338 ++++++++++++++++++ 1 file changed, 338 insertions(+) create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala new file mode 100644 index 000000000000..6a4c8126380e --- /dev/null +++ b/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. + 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("auxiliary key-column-names property survives identifiers containing special " + + "characters that exercise both JSON and SQL string-literal escaping") { + val session = spark + import session.implicits._ + + // This test exercises the full identifier-text persistence path with composite keys whose + // names collectively cover every escape class: + // - `it's` -- single quote: not escaped by JSON; the writer must double it + // to `''` to keep the SQL TBLPROPERTIES literal well-formed. + // - `name with spaces` -- whitespace identifier: backtick-quoted in DDL, no escaping + // needed in the JSON or the property value. + // - `a"b` -- literal double quote: JSON escapes as `\"`. + // - `c\d` -- literal backslash: JSON escapes as `\\`. + // If any layer drops, splits, or misescapes a name, the post-run lookup of the + // [[AutoCdcAuxiliaryTable.keyColumnNamesProperty]] property either fails to read or + // returns a value that is no longer a parseable JSON array of strings. + val keyNames = Seq("it's", "name with spaces", "a\"b", "c\\d") + + // SQL DDL identifier rendering: backticks delimit each identifier; an embedded backtick + // would have to be escaped by doubling, but none of these names contain one. + val targetTableDdl = keyNames + .map(name => s"`$name` STRING NOT NULL") + .mkString(", ") + s", version BIGINT NOT NULL, $scd2MetadataDdl" + spark.sql(s"CREATE TABLE $catalog.$namespace.target ($targetTableDdl)") + + // The AutoCDC API runs every key through `UnqualifiedColumnName.apply`, which calls + // `CatalystSqlParser.parseMultipartIdentifier`. To get a single-part identifier whose + // text includes special characters, the API caller has to backtick-quote at the boundary; + // we mirror that here by wrapping each name in backticks (and doubling any embedded + // backtick -- not needed for these names but kept for parity with how a user would call + // the API). + val backtickQuotedKeys = keyNames.map(name => s"`${name.replace("`", "``")}`") + + // Single MemoryStream reused across both runs so the streaming checkpoint can resume. + val stream = MemoryStream[(String, String, String, String, Long)] + def buildCtx(): TestGraphRegistrationContext = + singleAutoCdcFlowPipeline( + flowName = "auto_cdc_flow", + target = "target", + sourceDf = stream.toDF().toDF((keyNames :+ "version"): _*), + keys = backtickQuotedKeys, + sequencing = functions.col("version"), + scdType = ScdType.Type2) + + // Run #1: a single insert with arbitrary non-empty key values. + stream.addData(("v1", "v2", "v3", "v4", 1L)) + runPipeline(buildCtx()) + + // The persisted property must round-trip every name byte-for-byte. + assert(getAuxTableKeyColumnNames(target = "target") == keyNames) + + // Run #2: same keys, a higher sequence -- drift validation reads the property back, parses + // the JSON, and looks up each recorded name in the aux schema. If any layer mangled the + // identifier text (lost an escape, dropped a `'`, split on a `.`, ...), validation would + // either throw KEY_SCHEMA_DRIFT (name lookup miss) or INTERNAL_ERROR (recorded name absent + // from aux schema). Reaching the second run successfully proves the round-trip works. + stream.addData(("v1", "v2", "v3", "v4", 2L)) + runPipeline(buildCtx()) + + // The persisted property is immutable across non-full-refresh runs, so it must still be + // intact after run #2. + assert(getAuxTableKeyColumnNames(target = "target") == keyNames) + } + + private def getAuxTableKeyColumnNames(target: String): Seq[String] = { + val auxName = auxTableNameFor(target) + val rows = spark.sql(s"SHOW TBLPROPERTIES $auxName").collect() + val prop = rows + .find(_.getString(0) == AutoCdcAuxiliaryTable.keyColumnNamesProperty) + .getOrElse(fail( + s"auxiliary table $auxName is missing the " + + s"${AutoCdcAuxiliaryTable.keyColumnNamesProperty} property; got: ${rows.toSeq}" + )) + AutoCdcAuxiliaryTable.parseKeyColumnNames(prop.getString(1)) + .getOrElse(fail( + s"auxiliary table $auxName has a malformed " + + s"${AutoCdcAuxiliaryTable.keyColumnNamesProperty} property: '${prop.getString(1)}'" + )) + } +} From 3397d5fa6fd42e8908d00d30bb9c4dc84f5b83ea Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Wed, 29 Jul 2026 08:21:16 +0000 Subject: [PATCH 5/8] [SPARK-58409][SDP] Add AutoCdcScd2KeyDriftSuite SCD2 analog of AutoCdcScd1KeyDriftSuite (13 tests): key-arity add/drop, key swap, dataType drift, order-invariance, nullability/metadata invariance, backtick invariance (both directions), case-sensitive vs case-insensitive resolver behavior, and the tampered-aux failure modes (missing/malformed keyColumnNames property, recorded key absent from the aux schema). Key-drift validation is SCD-type-agnostic, so the tampered-aux tests hand-build an SCD2-shaped aux table (full target row schema plus the deleted-by-batch-id marker) carrying the SCD2 scd-type property. Co-authored-by: Opus 4.8 --- .../graph/AutoCdcScd2KeyDriftSuite.scala | 443 ++++++++++++++++++ 1 file changed, 443 insertions(+) create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala new file mode 100644 index 000000000000..d52ec8461e8f --- /dev/null +++ b/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')" + + test("a pipeline execution that adds a key column to an existing AutoCDC flow triggers " + + "KEY_SCHEMA_DRIFT") { + // Target table carries both candidate key columns up-front so only the AutoCDC `keys` + // declaration differs between the two pipelines. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, region STRING NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1 declares one key (`id`). + val stream1 = MemoryStream[(Int, String, Long)] + stream1.addData((1, "us", 1L)) + runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "region", "version"), Seq("id"))) + + // Pipeline #2 declares two keys (`region` + `id`) - arity drift. + val stream2 = MemoryStream[(Int, String, Long)] + stream2.addData((1, "us", 2L)) + val ctx2 = buildPipeline( + "flow_v2", stream2.toDF().toDF("id", "region", "version"), Seq("region", "id")) + + val ex = intercept[RuntimeException] { runPipeline(ctx2) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + // `region` is nullable here because Scala `String` is a reference type and the + // [[MemoryStream]] tuple encoder treats reference types as nullable. Only Scala + // primitives (`Int`, `Long`, ...) yield `NOT NULL` columns. + "expectedKeySchema" -> "region STRING,id INT NOT NULL", + "recordedKeySchema" -> "id INT NOT NULL" + ) + ) + } + + test("a pipeline execution that drops a key column from an existing AutoCDC flow triggers " + + "KEY_SCHEMA_DRIFT") { + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(region STRING NOT NULL, id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1 declares two keys [region, id]. + val stream1 = MemoryStream[(String, Int, Long)] + stream1.addData(("us", 1, 1L)) + runPipeline(buildPipeline( + "flow_v1", stream1.toDF().toDF("region", "id", "version"), Seq("region", "id"))) + + // Pipeline #2 declares only [id] - arity drift. + val stream2 = MemoryStream[(String, Int, Long)] + stream2.addData(("us", 1, 2L)) + val ctx2 = buildPipeline("flow_v2", stream2.toDF().toDF("region", "id", "version"), Seq("id")) + + val ex = intercept[RuntimeException] { runPipeline(ctx2) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + "expectedKeySchema" -> "id INT NOT NULL", + // `region` is nullable here because Scala `String` is a reference type; see the + // analogous comment in the "adds a key column" test above. + "recordedKeySchema" -> "region STRING,id INT NOT NULL" + ) + ) + } + + test("a pipeline execution that swaps a key in an existing AutoCDC flow for a different name " + + "(same arity) triggers KEY_SCHEMA_DRIFT") { + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, region STRING NOT NULL, country STRING NOT NULL, " + + s"version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1 declares [id, region]. + val stream1 = MemoryStream[(Int, String, String, Long)] + stream1.addData((1, "us", "USA", 1L)) + runPipeline(buildPipeline( + "flow_v1", stream1.toDF().toDF("id", "region", "country", "version"), Seq("id", "region"))) + + // Pipeline #2 declares [id, country] - same arity, different key set. + val stream2 = MemoryStream[(Int, String, String, Long)] + stream2.addData((1, "us", "USA", 2L)) + val ctx2 = buildPipeline( + "flow_v2", stream2.toDF().toDF("id", "region", "country", "version"), Seq("id", "country")) + + val ex = intercept[RuntimeException] { runPipeline(ctx2) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + // `country` and `region` are nullable here because Scala `String` is a reference type; + // see the analogous comment in the "adds a key column" test above. + "expectedKeySchema" -> "id INT NOT NULL,country STRING", + "recordedKeySchema" -> "id INT NOT NULL,region STRING" + ) + ) + } + + test("a pipeline whose recorded aux key dataType differs from the flow's source dataType " + + "triggers KEY_SCHEMA_DRIFT") { + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + // Pre-seed an SCD2-shaped aux table whose recorded key `id` is BIGINT, differing from the + // flow's INT source key. + spark.sql( + s"""CREATE TABLE ${auxTableNameFor("target")} """ + + s"""(id BIGINT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ + + s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) ${scd2AuxProps("[\"id\"]")}""" + ) + + val stream = MemoryStream[(Int, Long)] + stream.addData((1, 1L)) + val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id")) + + val ex = intercept[RuntimeException] { runPipeline(ctx) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + "expectedKeySchema" -> "id INT NOT NULL", + "recordedKeySchema" -> "id BIGINT NOT NULL" + ) + ) + } + + test("a composite key reorder ([a,b] -> [b,a]) does NOT trigger drift validation") { + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(a INT NOT NULL, b STRING NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1 declares keys [a, b]; pipeline #2 declares the same set reversed [b, a]. Drift + // validation is order-independent, so pipeline #2 must NOT throw. + val stream1 = MemoryStream[(Int, String, Long)] + stream1.addData((1, "x", 1L)) + runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("a", "b", "version"), Seq("a", "b"))) + + val stream2 = MemoryStream[(Int, String, Long)] + stream2.addData((2, "y", 1L)) + runPipeline(buildPipeline("flow_v2", stream2.toDF().toDF("a", "b", "version"), Seq("b", "a"))) + } + + test("a pipeline execution that changes a key column's nullability or metadata in an " + + "existing AutoCDC flow does NOT trigger drift") { + // Drift validation compares (name, dataType) pairs as a set; nullability and column metadata + // are not part of [[DataType]], so they do not gate semantic equivalence. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + // Pipeline #1: source carries `id INT NOT NULL` (Scala primitive `Int`), no metadata. + val stream1 = MemoryStream[(Int, Long)] + stream1.addData((1, 1L)) + runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("id"))) + + // Pipeline #2: source carries `id INT` (nullable, via `Option[Int]`) AND attaches non-empty + // column metadata. Same name and `dataType` as the recorded key, but every [[StructField]] + // aspect outside `dataType` differs. + val stream2 = MemoryStream[(Option[Int], Long)] + stream2.addData((Some(2), 2L)) + val baseDf = stream2.toDF().toDF("id", "version") + val md = new MetadataBuilder() + .putString("description", "primary key") + .build() + val sourceDfWithMetadata = baseDf.select(baseDf("id").as("id", md), baseDf("version")) + runPipeline(buildPipeline("flow_v2", sourceDfWithMetadata, Seq("id"))) + } + + test("a pipeline execution that wraps an existing AutoCDC flow's key in backticks does NOT " + + "trigger drift") { + // Backticks are a SQL-parse syntactic device, not part of the identifier itself. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + val stream1 = MemoryStream[(Int, Long)] + stream1.addData((1, 1L)) + runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("id"))) + + val stream2 = MemoryStream[(Int, Long)] + stream2.addData((2, 1L)) + runPipeline(buildPipeline("flow_v2", stream2.toDF().toDF("id", "version"), Seq("`id`"))) + } + + test("a pipeline execution that drops backticks around an existing AutoCDC flow's " + + "previously-backtick-quoted key does NOT trigger drift") { + // The reverse direction: drift validation must be backtick-invariant on both the write side + // (recorded property strips backticks) and the read side (resolver-aware lookup). + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + val stream1 = MemoryStream[(Int, Long)] + stream1.addData((1, 1L)) + runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("`id`"))) + + val stream2 = MemoryStream[(Int, Long)] + stream2.addData((2, 1L)) + runPipeline(buildPipeline("flow_v2", stream2.toDF().toDF("id", "version"), Seq("id"))) + } + + test("under spark.sql.caseSensitive = true, an AutoCDC flow whose key differs only in case " + + "from the recorded key triggers KEY_SCHEMA_DRIFT") { + // validateNoKeyColumnDrift uses spark.sessionState.conf.resolver, so its behavior on + // `Id` vs `id` flips with the session conf. Pipeline #1 seeds the aux under the default + // resolver with recorded key `["id"]`; pipeline #2 runs under the case-sensitive resolver + // with key `["Id"]`, which is a distinct identifier there, so drift must fire. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + val stream1 = MemoryStream[(Int, Long)] + stream1.addData((1, 1L)) + runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("id"))) + + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + val stream2 = MemoryStream[(Int, Long)] + stream2.addData((1, 2L)) + val ctx2 = buildPipeline("flow_v2", stream2.toDF().toDF("Id", "version"), Seq("Id")) + + val ex = intercept[RuntimeException] { runPipeline(ctx2) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + "expectedKeySchema" -> "Id INT NOT NULL", + "recordedKeySchema" -> "id INT NOT NULL" + ) + ) + } + } + + test("under the default (case-insensitive) resolver, an AutoCDC flow whose key differs only " + + "in case from the recorded key does NOT trigger drift") { + // Pairs with the case-sensitive test above: under the default resolver the two identifiers + // are equivalent, so drift validation must accept pipeline #2. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + + val stream1 = MemoryStream[(Int, Long)] + stream1.addData((1, 1L)) + runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"), Seq("id"))) + + val stream2 = MemoryStream[(Int, Long)] + stream2.addData((1, 2L)) + runPipeline(buildPipeline("flow_v2", stream2.toDF().toDF("id", "version"), Seq("Id"))) + } + + test("a pipeline whose aux table is missing the keyColumnNames property fails with " + + "AUXILIARY_TABLE_PROPERTY_MISSING") { + // Pre-create the aux table directly without the [[keyColumnNamesProperty]] to simulate + // corrupt metadata. Validation must surface a structured AUTOCDC_INVALID_STATE error. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + spark.sql( + s"""CREATE TABLE ${auxTableNameFor("target")} """ + + s"""(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ + + s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) """ + + s"""TBLPROPERTIES ('${AutoCdcAuxiliaryTable.scdTypePropertyKey}' = """ + + s"""'${ScdType.Type2.label}')""" + ) + + val stream = MemoryStream[(Int, Long)] + stream.addData((1, 1L)) + val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id")) + + val ex = intercept[RuntimeException] { runPipeline(ctx) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty + ) + ) + } + + test("a pipeline whose aux table has a malformed keyColumnNames property fails with " + + "AUXILIARY_TABLE_PROPERTY_MALFORMED") { + // Pre-create the aux table directly with a non-JSON-array property value to simulate corrupt + // metadata. Validation must surface a structured AUTOCDC_INVALID_STATE error. + val malformedKeysArray = "not-a-json-array" + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + spark.sql( + s"""CREATE TABLE ${auxTableNameFor("target")} """ + + s"""(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ + + s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) """ + + scd2AuxProps(malformedKeysArray) + ) + + val stream = MemoryStream[(Int, Long)] + stream.addData((1, 1L)) + val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id")) + + val ex = intercept[RuntimeException] { runPipeline(ctx) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MALFORMED", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty, + "rawValue" -> malformedKeysArray + ) + ) + } + + test("a pipeline whose aux table records a key absent from its schema fails with " + + "AUXILIARY_TABLE_KEY_COLUMN_MISSING") { + // Pre-create the aux table with the [[keyColumnNamesProperty]] pointing at a column that does + // not exist in the aux schema. Validation must surface a structured AUTOCDC_INVALID_STATE + // error rather than KEY_SCHEMA_DRIFT. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + spark.sql( + s"""CREATE TABLE ${auxTableNameFor("target")} """ + + s"""(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ + + s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) ${scd2AuxProps("[\"region\"]")}""" + ) + + val stream = MemoryStream[(Int, Long)] + stream.addData((1, 1L)) + val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id")) + + val ex = intercept[RuntimeException] { runPipeline(ctx) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_KEY_COLUMN_MISSING", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + "keyColumnName" -> "region", + "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty + ) + ) + } + + /** + * Build a single-flow SCD2 pipeline targeting `cat.ns1.target` with the given source DF and key + * column list. Thin wrapper over [[singleAutoCdcFlowPipeline]] since every drift test targets + * the same `target` table. + */ + private def buildPipeline( + flowName: String, + sourceDf: DataFrame, + keys: Seq[String]): TestGraphRegistrationContext = + singleAutoCdcFlowPipeline( + flowName = flowName, + target = "target", + sourceDf = sourceDf, + keys = keys, + sequencing = $"version", + scdType = ScdType.Type2) +} From d902d4cd114f51d100cb90a42bc09fc258d4bd87 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Wed, 29 Jul 2026 08:38:08 +0000 Subject: [PATCH 6/8] [SPARK-58409][SDP] Add AutoCdcScd2SchemaEvolutionSuite SCD2 analog of AutoCdcScd1SchemaEvolutionSuite: nullable-column merge, new top-level column add, additive target-column evolution (which for SCD2 also extends the aux table since it mirrors the full target row), broadening the column selection, incompatible type changes (widen/narrow/timestamp->string), and a case-only source-column rename. The narrowing / dropped-column cases (a microbatch narrower than the evolved target, including dropped nested struct/array fields) are owned by AutoCdcScd2ColumnEvolutionSuite under SPARK-58418, which makes them reconcile correctly, so they are not duplicated here. Co-authored-by: Opus 4.8 --- .../AutoCdcScd2SchemaEvolutionSuite.scala | 389 ++++++++++++++++++ 1 file changed, 389 insertions(+) create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala new file mode 100644 index 000000000000..ddbac740219b --- /dev/null +++ b/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. + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(key INT NOT NULL, version BIGINT NOT NULL, value STRING, $scd2MetadataDdl)" + ) + + val stream = MemoryStream[(Int, Long, String)] + stream.addData((1, 1L, "alice")) + val df = stream.toDF().toDF("key", "version", "Value") + val ctx = singleAutoCdcFlowPipeline( + flowName = "auto_cdc_flow", + target = "target", + sourceDf = df, + keys = Seq("key"), + sequencing = functions.col("version"), + scdType = ScdType.Type2) + + val ex = intercept[RuntimeException] { runPipeline(ctx) } + checkErrorInPipelineFailure( + failure = ex, + condition = "COLUMN_ALREADY_EXISTS", + parameters = Map("columnName" -> "`value`") + ) + } + } + + test("changing a non-key column type from TIMESTAMP to STRING between runs fails with " + + "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE") { + val session = spark + import session.implicits._ + + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(key INT NOT NULL, version BIGINT NOT NULL, value TIMESTAMP, $scd2MetadataDdl)" + ) + + val stream1 = MemoryStream[(Int, Long, Timestamp)] + stream1.addData((1, 1L, Timestamp.valueOf("2024-01-01 10:00:00"))) + runPipeline(singleAutoCdcFlowPipeline( + flowName = "auto_cdc_flow", + target = "target", + sourceDf = stream1.toDF().toDF("key", "version", "value"), + keys = Seq("key"), + sequencing = functions.col("version"), + scdType = ScdType.Type2)) + + // Run #2 emits `value` as STRING. mergeSchemas rejects the type change. + val stream2 = MemoryStream[(Int, Long, String)] + stream2.addData((1, 2L, "2024-01-02 11:00:00")) + val ctx2 = singleAutoCdcFlowPipeline( + flowName = "auto_cdc_flow", + target = "target", + sourceDf = stream2.toDF().toDF("key", "version", "value"), + keys = Seq("key"), + sequencing = functions.col("version"), + scdType = ScdType.Type2) + + val ex = intercept[RuntimeException] { runPipeline(ctx2) } + checkErrorInPipelineFailure( + failure = ex, + condition = "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE", + sqlState = Some("42825"), + parameters = Map( + "left" -> "\"TIMESTAMP\"", + "right" -> "\"STRING\"" + ) + ) + } +} From 0497c91f6849a94f1865f1627389193f66c0a2d4 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Mon, 3 Aug 2026 05:59:43 +0000 Subject: [PATCH 7/8] [SPARK-58409][SDP] Address review comments - Add an aux-sole-holder durability test (delete-only run leaves a tombstone only in the aux; a later lower-sequence upsert lands closed) and soften the transparently-recreated comment so the aux is not read as disposable. - Add an SCD_TYPE_DRIFT end-to-end test (recorded SCD1 aux, SCD2 flow), which the SCD1 suite structurally cannot cover. - Add an additive array nested-field evolution test for SCD2, and note in the header that trackHistorySelection changes and SCD1's extra-target-column preservation are covered elsewhere / have no SCD2 analog. - Correct the case-only-rename comment: COLUMN_ALREADY_EXISTS is raised by ResolveUnion over the reconcile unionByName, not at aux-table creation. Co-authored-by: Opus 4.8 --- ...CdcScd2AuxiliaryTableDurabilitySuite.scala | 68 ++++++++++++- .../graph/AutoCdcScd2KeyDriftSuite.scala | 38 ++++++++ .../AutoCdcScd2SchemaEvolutionSuite.scala | 95 +++++++++++++++++-- 3 files changed, 186 insertions(+), 15 deletions(-) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala index 6a4c8126380e..45ae6a8d082d 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableDurabilitySuite.scala @@ -20,7 +20,12 @@ 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.autocdc.{ + ColumnSelection, + Scd2BatchProcessor, + ScdType, + UnqualifiedColumnName +} import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext} import org.apache.spark.sql.test.SharedSparkSession @@ -242,10 +247,13 @@ class AutoCdcScd2AuxiliaryTableDurabilitySuite 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. + // 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"), @@ -256,6 +264,56 @@ class AutoCdcScd2AuxiliaryTableDurabilitySuite ) } + 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. + assert(spark.table(auxTableNameFor("target")).count() == 1, + "the delete-only run should record exactly one aux tombstone row") + + // Run #2 (aux retained): a later upsert at seq=5, BELOW the recorded seq=10. Because the aux + // still holds the seq=10 tombstone, reconciliation weaves seq=5 in as a closed prior record + // ending at 10 rather than an open current record -- state the target alone could not supply. + stream.addData((1, "early", 5L, false)) + runPipeline(buildCtx()) + checkAnswer( + spark.table(s"$catalog.$namespace.target"), + Seq(Row(1, "early", 5L, 5L, 10L, scd2Meta(5L))) + ) + } + test("auxiliary key-column-names property survives identifiers containing special " + "characters that exercise both JSON and SQL string-literal escaping") { val session = spark diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala index d52ec8461e8f..f23167646f6b 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2KeyDriftSuite.scala @@ -424,6 +424,44 @@ class AutoCdcScd2KeyDriftSuite ) } + test("a pipeline execution whose recorded SCD type differs from the flow's SCD type triggers " + + "SCD_TYPE_DRIFT") { + // SCD_TYPE_DRIFT is reachable end-to-end from DatasetManager (validateNoScdTypeDrift runs + // right after validateNoKeyColumnDrift when evolving the aux table), but the SCD1 suite + // structurally cannot cover it, so this SCD2 suite is its home. Pre-create an SCD2-shaped aux + // table whose recorded scd-type property is Type1, with matching keys so key-drift validation + // passes and the scd-type check is the one that fires, then run a Type2 flow. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl)" + ) + spark.sql( + s"""CREATE TABLE ${auxTableNameFor("target")} """ + + s"""(id INT NOT NULL, version BIGINT NOT NULL, $scd2MetadataDdl, """ + + s"""${Scd2BatchProcessor.deletedByBatchIdColName} BIGINT) """ + + s"""TBLPROPERTIES (""" + + s"""'${AutoCdcAuxiliaryTable.scdTypePropertyKey}' = '${ScdType.Type1.label}', """ + + s"""'${AutoCdcAuxiliaryTable.keyColumnNamesProperty}' = '[\"id\"]')""" + ) + + val stream = MemoryStream[(Int, Long)] + stream.addData((1, 1L)) + val ctx = buildPipeline("flow", stream.toDF().toDF("id", "version"), Seq("id")) + + val ex = intercept[RuntimeException] { runPipeline(ctx) } + checkErrorInPipelineFailure( + failure = ex, + condition = "AUTOCDC_INVALID_STATE.SCD_TYPE_DRIFT", + sqlState = Some("42000"), + parameters = Map( + "tableName" -> + fullyQualifiedIdentifier("target", Some(catalog), Some(namespace)).unquotedString, + "expectedScdType" -> ScdType.Type2.label, + "recordedScdType" -> ScdType.Type1.label + ) + ) + } + /** * Build a single-flow SCD2 pipeline targeting `cat.ns1.target` with the given source DF and key * column list. Thin wrapper over [[singleAutoCdcFlowPipeline]] since every drift test targets diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala index ddbac740219b..34cfe9e88790 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala @@ -30,16 +30,24 @@ 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). + * cases (new top-level columns, a new field inside an array element, 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. + * 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 @@ -264,6 +272,70 @@ class AutoCdcScd2SchemaEvolutionSuite assert(spark.table(auxTableNameFor("target")).schema.fieldNames.contains("name")) } + test("a new field added inside an array element between runs is added to the " + + "target") { + val session = spark + import session.implicits._ + + // SCD2 analog of AutoCdcScd1SchemaEvolutionSuite's array additive case: unlike the + // top-level scalar additions above, this exercises unionByName / mergeSchemas recursing into + // an array element struct. Unlike SCD1's overwrite-in-place, the SCD2 upsert closes the prior + // record (which never saw `vals.element.b.d`, so it reads NULL there) and opens a new one + // carrying the widened value. + spark.sql( + s"CREATE TABLE $catalog.$namespace.target " + + s"(key INT NOT NULL, version BIGINT NOT NULL, " + + s"vals ARRAY>>, $scd2MetadataDdl)" + ) + + val stream = MemoryStream[(Int, Long, Int, Int, Int)] + def buildCtx(includeD: Boolean): TestGraphRegistrationContext = { + val src = stream.toDF().toDF("key", "version", "a", "b_c", "b_d") + val inner = if (includeD) { + functions.struct(functions.col("b_c").as("c"), functions.col("b_d").as("d")) + } else { + functions.struct(functions.col("b_c").as("c")) + } + val projected = src.select( + functions.col("key"), + functions.col("version"), + functions.array( + functions.struct(functions.col("a"), inner.as("b")) + ).as("vals") + ) + singleAutoCdcFlowPipeline( + flowName = "auto_cdc_flow", + target = "target", + sourceDf = projected, + keys = Seq("key"), + sequencing = functions.col("version"), + scdType = ScdType.Type2) + } + + // Run #1: element struct is (a, b.c); no b.d yet. Opens key=1's current record at version=1. + stream.addData((1, 1L, 1, 1, 99)) + runPipeline(buildCtx(includeD = false)) + + // Run #2 widens the element struct with b.d. The version=2 upsert to key=1 closes its + // version=1 record (which predates b.d, so reads NULL) and opens a new one with b.d=2; the + // new key=3 lands as an open record with the full widened struct. + stream.addData((1, 2L, 1, 1, 2), (3, 1L, 3, 3, 3)) + runPipeline(buildCtx(includeD = true)) + + // Inline-explode flattens the array; carry the interval bounds to prove the closed + // prior record reads NULL for the newly-added nested field. + checkAnswer( + spark.table(s"$catalog.$namespace.target") + .selectExpr("key", "__START_AT", "__END_AT", "inline(vals) as (a, b)") + .select("key", "__START_AT", "__END_AT", "a", "b.c", "b.d"), + Seq( + Row(1, 1L, 2L, 1, 1, null), + Row(1, 2L, null, 1, 1, 2), + Row(3, 1L, null, 3, 3, 3) + ) + ) + } + test("broadening the column selection between runs adds the newly-included column to " + "the target") { val session = spark @@ -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. + // The failure then surfaces during microbatch reconciliation, not at table/aux creation: + // Scd2ForeachBatchHandler.reconcileMicrobatch reads that now-two-column target back and folds + // it into the affected-rows `unionByName`; ResolveUnion runs a case-insensitive + // duplicate-column check over the union's schema and reports COLUMN_ALREADY_EXISTS. SCD1 has no + // such reconcile union and instead surfaces AMBIGUOUS_REFERENCE deeper in the MERGE plan -- so + // the same user mistake maps to two different conditions across the SCD types. withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { spark.sql( s"CREATE TABLE $catalog.$namespace.target " + From 6dfe79810538120818033276eb3448e77a7b6b9b Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Mon, 3 Aug 2026 15:01:23 +0000 Subject: [PATCH 8/8] [SPARK-58409][SDP] Cite SPARK-58517 in case-only-rename test comment The COLUMN_ALREADY_EXISTS case is fallout of a case-insensitivity bug in DatasetManager's schema evolution (SPARK-58517), not intended behavior. Reword the comment to flag it as characterization of known-buggy behavior. Co-authored-by: Opus 4.8 --- .../graph/AutoCdcScd2SchemaEvolutionSuite.scala | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala index 34cfe9e88790..b50245236b7b 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SchemaEvolutionSuite.scala @@ -385,14 +385,18 @@ class AutoCdcScd2SchemaEvolutionSuite 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 target is evolved to carry both columns. - // The failure then surfaces during microbatch reconciliation, not at table/aux creation: + // This pins current (buggy) behavior, tracked by SPARK-58517: DatasetManager's schema + // evolution ignores spark.sql.caseSensitive and merges case-sensitively, so a target `value` + // and a source `Value` are wrongly treated as distinct and the target is evolved to carry both + // columns -- a schema self-inconsistent under the (default) case-insensitive resolver. The + // failure then surfaces during microbatch reconciliation, not at table/aux creation: // Scd2ForeachBatchHandler.reconcileMicrobatch reads that now-two-column target back and folds // it into the affected-rows `unionByName`; ResolveUnion runs a case-insensitive // duplicate-column check over the union's schema and reports COLUMN_ALREADY_EXISTS. SCD1 has no // such reconcile union and instead surfaces AMBIGUOUS_REFERENCE deeper in the MERGE plan -- so - // the same user mistake maps to two different conditions across the SCD types. + // the same user mistake maps to two different conditions across the SCD types. Once SPARK-58517 + // is fixed the merge should be a no-op (source `Value` maps onto target `value`) and this test + // should be updated to assert success. withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { spark.sql( s"CREATE TABLE $catalog.$namespace.target " +