From bd07e659de1974ac44d1f82cf4a28fde91e6398b Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 9 Sep 2026 04:55:30 +0000 Subject: [PATCH 01/15] [SPARK-59358][SDP] Project missing columns during SCD2 preprocessing --- .../autocdc/Scd2BatchProcessor.scala | 37 +++++-- .../autocdc/Scd2ForeachBatchHandler.scala | 17 +-- .../autocdc/Scd2BatchProcessorSuite.scala | 102 ++++++++++++++---- 3 files changed, 117 insertions(+), 39 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index 2910d0664bc22..908402cbd1b09 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala @@ -123,9 +123,10 @@ case class Scd2BatchProcessor( * consume. * * Step ordering is load-bearing: the row-extension steps reference user data columns that - * target-column selection is allowed to drop, so selection runs last. Unlike SCD1, no per-key - * deduplication step is performed here - SCD2 preserves every event as part of the row's - * history, including byte-identical full-event duplicates. + * target-column selection is allowed to drop. Selection therefore runs after those extensions, + * followed by target-schema alignment. Unlike SCD1, no per-key deduplication step is performed + * here - SCD2 preserves every event as part of the row's history, including byte-identical + * full-event duplicates. * * Duplicate event elimination (e.g., collapsing two identical events at the same sequence), * whether across microbatches or within the same microbatch, is the responsibility of @@ -133,22 +134,24 @@ case class Scd2BatchProcessor( * * @param microbatchDf * the incoming CDC microbatch. + * @param targetTableDf + * the current persisted target. The selected microbatch is aligned to its schema so target + * columns omitted by the source receive explicit nulls before reconciliation. * @return * a dataframe that retains every input row 1:1 - no rows added, dropped, reordered, or - * merged - with the following schema, in column order: - * 1. The user columns of `microbatchDf` that survive [[ChangeArgs.columnSelection]], in - * the order they appeared in the input. - * 2. [[startAtColName]], populated with the sequence value of the row. - * 3. [[endAtColName]], populated with the sequence value of the row IFF it's a delete - * event, null otherwise. - * 4. [[cdcMetadataColName]], conforming to [[cdcMetadataColSchema]]. + * merged. Its fields use `targetTableDf` as the authority for order and spelling. + * [[startAtColName]], [[endAtColName]], and [[cdcMetadataColName]] are populated according + * to their documented contracts. */ - private[autocdc] def preprocessMicrobatch(microbatchDf: DataFrame): DataFrame = { + private[autocdc] def preprocessMicrobatch( + microbatchDf: DataFrame, + targetTableDf: DataFrame): DataFrame = { microbatchDf .transform(extendMicrobatchRowsWithStartAt) .transform(extendMicrobatchRowsWithEndAt) .transform(extendMicrobatchRowsWithCdcMetadata) .transform(projectTargetColumnsOntoMicrobatch) + .transform(alignMicrobatchToTargetSchema(_, targetTableDf)) } /** @@ -246,6 +249,18 @@ case class Scd2BatchProcessor( microbatch.select(finalColumnsToSelect: _*) } + /** + * Align the selected incoming rows to the current persisted target schema. The target-shaped + * side is empty, so this retains exactly the microbatch's rows while [[DataFrame.unionByName]] + * supplies nulls for target columns omitted by the source, including nested struct/array fields. + * + * Keeping the target as the left schema authority also preserves its column order and spelling. + */ + private def alignMicrobatchToTargetSchema( + projectedDf: DataFrame, + targetTableDf: DataFrame): DataFrame = + targetTableDf.limit(0).unionByName(projectedDf, allowMissingColumns = true) + /** * For each key in the preprocessed microbatch, compute the earliest [[recordStartAtFieldName]] * across the key's events. diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala index e4adfcb516be7..3aabc4b19b845 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala @@ -68,14 +68,17 @@ case class Scd2ForeachBatchHandler( batchId = batchId ).validateMicrobatch() - val preprocessedBatchDf = batchProcessor.preprocessMicrobatch(batchDf) + val targetTableDf = batchDf.sparkSession.read.table(targetTableIdentifier.quotedString) + val preprocessedBatchDf = batchProcessor.preprocessMicrobatch( + microbatchDf = batchDf, + targetTableDf = targetTableDf + ) val perKeyMinimumSequenceInMicrobatchDf = batchProcessor.computeMinimumSequencePerKey( preprocessedBatchDf ) val auxTableDf = batchDf.sparkSession.read.table(auxiliaryTableIdentifier.quotedString) - val targetTableDf = batchDf.sparkSession.read.table(targetTableIdentifier.quotedString) val perKeyAffectedSequenceCutoffDf = batchProcessor.computePerKeyAffectedSequenceCutoff( rawAuxiliaryTableDf = auxTableDf, @@ -95,12 +98,10 @@ case class Scd2ForeachBatchHandler( perKeyAffectedSequenceCutoffDf = perKeyAffectedSequenceCutoffDf ) - // The three inputs share the canonical SCD2 row schema by name, but not necessarily by column - // set: after cross-run schema evolution the target (and the aux table, which mirrors it) can - // carry user columns that the current microbatch no longer emits. `allowMissingColumns` pads - // such columns with null on the side that lacks them (recursing into structs and arrays; map - // types are not supported) instead of failing the union. (findAffectedRowsFromAuxiliaryTable - // drops the aux-only deletedByBatchId column.) + // Preprocessing has already aligned the microbatch to the target schema. Keep + // allowMissingColumns here as a safeguard for nested differences between persisted target and + // auxiliary rows; findAffectedRowsFromAuxiliaryTable also drops the aux-only + // deletedByBatchId column. val microbatchAndAffectedRows = preprocessedBatchDf .unionByName(affectedRowsFromAuxiliaryTable, allowMissingColumns = true) .unionByName(affectedRowsFromTargetTable, allowMissingColumns = true) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala index 71d837740321d..093461a96f386 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala @@ -30,6 +30,28 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { private def microbatchOf(schema: StructType)(rows: Row*): DataFrame = spark.createDataFrame(spark.sparkContext.parallelize(rows), schema) + /** + * Preprocess a microbatch against either an explicit persisted user schema or, by default, the + * microbatch's user-selected schema. The latter keeps tests unrelated to cross-run evolution + * focused on their existing axis. + */ + private def preprocessMicrobatch( + processor: Scd2BatchProcessor, + microbatch: DataFrame, + targetUserSchema: Option[StructType] = None): DataFrame = { + val selectedMicrobatchSchema = ColumnSelection.applyToSchema( + schemaName = "microbatch", + schema = microbatch.schema, + columnSelection = processor.changeArgs.columnSelection, + resolver = spark.sessionState.conf.resolver + ) + val targetTableDf = targetTableOf( + targetUserSchema.getOrElse(selectedMicrobatchSchema), + processor.resolvedSequencingType + )() + processor.preprocessMicrobatch(microbatch, targetTableDf) + } + /** * Build an mock aux-table [[DataFrame]] from explicit user rows + framework column values. */ @@ -270,7 +292,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) assert(result.schema.fieldNames.toSeq == Seq( "id", "seq", "value", @@ -309,7 +331,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) assert(result.collect().isEmpty) assert(result.schema.fieldNames.toSeq == Seq( @@ -353,7 +375,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // - __RECORD_START_AT = sequencing for every row, regardless of delete vs upsert // (lineage preserved into the merge step) checkAnswer( - df = processor.preprocessMicrobatch(batch), + df = preprocessMicrobatch(processor, batch), expectedAnswer = Seq( Row(1, 10L, "first-upsert", false, 10L, null, Row(10L, null)), Row(1, 20L, "second-upsert", false, 20L, null, Row(20L, null)), @@ -391,7 +413,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // Both rows must survive verbatim. checkAnswer( - df = processor.preprocessMicrobatch(batch), + df = preprocessMicrobatch(processor, batch), expectedAnswer = Seq( Row(1, 10L, "alice", false, 10L, null, Row(10L, null)), Row(1, 10L, "alice", false, 10L, null, Row(10L, null)) @@ -421,7 +443,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) checkAnswer( - df = processor.preprocessMicrobatch(batch).select( + df = preprocessMicrobatch(processor, batch).select( F.col(Scd2BatchProcessor.endAtColName) ), expectedAnswer = Seq(Row(null), Row(null)) @@ -452,7 +474,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) checkAnswer( - df = processor.preprocessMicrobatch(batch).select( + df = preprocessMicrobatch(processor, batch).select( F.col(Scd2BatchProcessor.endAtColName) ), expectedAnswer = Row(null) @@ -485,7 +507,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) checkAnswer( df = result.select( @@ -522,7 +544,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) assert(result.schema.fieldNames.toSeq == Seq( "id", "name", "age", "seq", @@ -549,7 +571,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) assert(result.schema.fieldNames.toSeq == Seq( "id", "age", @@ -581,7 +603,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) assert(result.schema.fieldNames.toSeq == Seq( "id", "age", "seq", @@ -614,7 +636,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) // Output column order follows the microbatch schema (id before age), not the user's listing // order in IncludeColumns. Framework columns are always appended last. @@ -646,7 +668,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) // Output column names follow the microbatch schema's casing, not the user's casing. assert(result.schema.fieldNames.toSeq == Seq( @@ -685,7 +707,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { resolvedSequencingType = LongType ) - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) assert(result.schema.fieldNames.toSeq == Seq( "id", "user.id", @@ -729,7 +751,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // The orchestrator runs row-extension steps before column selection, so the framework // columns reference seq / is_delete fully even though the final projection drops them. - val result = processor.preprocessMicrobatch(batch) + val result = preprocessMicrobatch(processor, batch) assert(result.schema.fieldNames.toSeq == Seq( "id", "value", @@ -746,6 +768,46 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) } + test("preprocessMicrobatch aligns selected rows to the persisted target schema") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + val batchSchema = new StructType() + .add("id", IntegerType) + .add("Value", new StructType().add("a", IntegerType)) + .add("seq", LongType) + val targetUserSchema = new StructType() + .add("id", IntegerType) + .add("value", new StructType() + .add("a", IntegerType) + .add("removedNested", StringType)) + .add("removedTopLevel", StringType) + val batch = microbatchOf(batchSchema)(Row(1, Row(2), 10L)) + val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + columnSelection = Some(ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq")))) + ), + resolvedSequencingType = LongType + ) + + val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema)) + assert(result.schema.fieldNames.toSeq == Seq( + "id", + "value", + "removedTopLevel", + Scd2BatchProcessor.startAtColName, + Scd2BatchProcessor.endAtColName, + AutoCdcReservedNames.cdcMetadataColName + )) + checkAnswer( + df = result, + expectedAnswer = Row(1, Row(2, null), null, 10L, null, Row(10L, null)) + ) + } + } + // =============== computeMinimumSequencePerKey tests =============== test("computeMinimumSequencePerKey returns one row per distinct key and aggregates across " + @@ -774,7 +836,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(2, 40L, true) // delete - smallest sequence for key=2 ) - val preprocessed = processor.preprocessMicrobatch(raw) + val preprocessed = preprocessMicrobatch(processor, raw) val result = processor.computeMinimumSequencePerKey(preprocessed) assert(result.schema.fieldNames.toSeq == Seq( @@ -808,7 +870,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { Row("EU", 1, 30L) ) - val preprocessed = processor.preprocessMicrobatch(raw) + val preprocessed = preprocessMicrobatch(processor, raw) val result = processor.computeMinimumSequencePerKey(preprocessed) assert(result.schema.fieldNames.toSeq == Seq( @@ -832,7 +894,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { val processor = processorWithKeys(keys = Seq("id")) val raw = microbatchOf(schema)() - val preprocessed = processor.preprocessMicrobatch(raw) + val preprocessed = preprocessMicrobatch(processor, raw) val result = processor.computeMinimumSequencePerKey(preprocessed) assert(result.collect().isEmpty) @@ -853,7 +915,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 30L), Row(1, 10L) ) - val preprocessed = processor.preprocessMicrobatch(raw) + val preprocessed = preprocessMicrobatch(processor, raw) val result = processor.computeMinimumSequencePerKey(preprocessed) assert(result.schema.fieldNames.toSeq == Seq( @@ -2175,7 +2237,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 10L, "alice", "inactive") ) - val result = processor.reconcileStartAndEndAt(processor.preprocessMicrobatch(df)) + val result = processor.reconcileStartAndEndAt(preprocessMicrobatch(processor, df)) checkAnswer( df = result, @@ -2209,7 +2271,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) val ex = intercept[AnalysisException] { - processor.reconcileStartAndEndAt(processor.preprocessMicrobatch(df)) + processor.reconcileStartAndEndAt(preprocessMicrobatch(processor, df)) } assert(ex.getCondition == "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA") } From 8a0ec2b28d03f597a711c05579f291c3b57638b6 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 9 Sep 2026 17:55:37 +0000 Subject: [PATCH 02/15] [SPARK-59358][SDP] Test preprocessing column casing --- .../autocdc/Scd2BatchProcessorSuite.scala | 68 ++++++++++++++++++- 1 file changed, 65 insertions(+), 3 deletions(-) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala index 093461a96f386..0c8d2cb6c5166 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala @@ -768,16 +768,76 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) } - test("preprocessMicrobatch aligns selected rows to the persisted target schema") { + gridTest("preprocessMicrobatch uses target spelling under case-insensitive analysis")( + Seq( + ("id", "Value", "ID", "value"), + ("ID", "value", "id", "Value") + ) + ) { case (sourceKey, sourceValue, targetKey, targetValue) => withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { val batchSchema = new StructType() + .add(sourceKey, IntegerType) + .add(sourceValue, StringType) + .add("seq", LongType) + val targetUserSchema = new StructType() + .add(targetKey, IntegerType) + .add(targetValue, StringType) + val batch = microbatchOf(batchSchema)(Row(1, "a", 10L)) + val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName(sourceKey)), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + columnSelection = Some(ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq")))) + ), + resolvedSequencingType = LongType + ) + + val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema)) + assert(result.schema.fieldNames.take(2).toSeq == Seq(targetKey, targetValue)) + checkAnswer(result.select(F.col(targetKey), F.col(targetValue)), Row(1, "a")) + } + } + + test("preprocessMicrobatch keeps distinct case-sensitive columns") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + val batchSchema = new StructType() + .add("id", IntegerType) + .add("Value", StringType) + .add("seq", LongType) + val targetUserSchema = new StructType() .add("id", IntegerType) - .add("Value", new StructType().add("a", IntegerType)) + .add("value", StringType) + .add("Value", StringType) + val batch = microbatchOf(batchSchema)(Row(1, "a", 10L)) + val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + columnSelection = Some(ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq")))) + ), + resolvedSequencingType = LongType + ) + + val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema)) + assert(result.schema.fieldNames.take(3).toSeq == Seq("id", "value", "Value")) + checkAnswer(result.select(F.col("value"), F.col("Value")), Row(null, "a")) + } + } + + test("preprocessMicrobatch uses target spelling recursively under case-insensitive analysis") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + val batchSchema = new StructType() + .add("ID", IntegerType) + .add("Value", new StructType().add("City", IntegerType)) .add("seq", LongType) val targetUserSchema = new StructType() .add("id", IntegerType) .add("value", new StructType() - .add("a", IntegerType) + .add("city", IntegerType) .add("removedNested", StringType)) .add("removedTopLevel", StringType) val batch = microbatchOf(batchSchema)(Row(1, Row(2), 10L)) @@ -801,6 +861,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { Scd2BatchProcessor.endAtColName, AutoCdcReservedNames.cdcMetadataColName )) + assert(result.schema("value").dataType.asInstanceOf[StructType].fieldNames.toSeq == + Seq("city", "removedNested")) checkAnswer( df = result, expectedAnswer = Row(1, Row(2, null), null, 10L, null, Row(10L, null)) From ded99f144b0ff2206229d0171ec6435cb2b4091f Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Thu, 10 Sep 2026 17:40:50 +0000 Subject: [PATCH 03/15] cleanup --- .../spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala | 6 +++--- .../sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala | 4 ++-- .../sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala | 8 +++++--- 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index 908402cbd1b09..de06cdc61c442 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala @@ -135,8 +135,7 @@ case class Scd2BatchProcessor( * @param microbatchDf * the incoming CDC microbatch. * @param targetTableDf - * the current persisted target. The selected microbatch is aligned to its schema so target - * columns omitted by the source receive explicit nulls before reconciliation. + * the current persisted target table. * @return * a dataframe that retains every input row 1:1 - no rows added, dropped, reordered, or * merged. Its fields use `targetTableDf` as the authority for order and spelling. @@ -254,7 +253,8 @@ case class Scd2BatchProcessor( * side is empty, so this retains exactly the microbatch's rows while [[DataFrame.unionByName]] * supplies nulls for target columns omitted by the source, including nested struct/array fields. * - * Keeping the target as the left schema authority also preserves its column order and spelling. + * Keeping the target as the left schema authority also preserves its column order and exact + * case-spelling. */ private def alignMicrobatchToTargetSchema( projectedDf: DataFrame, diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala index 3aabc4b19b845..36060e831e32b 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala @@ -69,6 +69,8 @@ case class Scd2ForeachBatchHandler( ).validateMicrobatch() val targetTableDf = batchDf.sparkSession.read.table(targetTableIdentifier.quotedString) + val auxTableDf = batchDf.sparkSession.read.table(auxiliaryTableIdentifier.quotedString) + val preprocessedBatchDf = batchProcessor.preprocessMicrobatch( microbatchDf = batchDf, targetTableDf = targetTableDf @@ -78,8 +80,6 @@ case class Scd2ForeachBatchHandler( preprocessedBatchDf ) - val auxTableDf = batchDf.sparkSession.read.table(auxiliaryTableIdentifier.quotedString) - val perKeyAffectedSequenceCutoffDf = batchProcessor.computePerKeyAffectedSequenceCutoff( rawAuxiliaryTableDf = auxTableDf, targetTableDf = targetTableDf, diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala index 0c8d2cb6c5166..6b0f2d3d866cf 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala @@ -768,7 +768,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) } - gridTest("preprocessMicrobatch uses target spelling under case-insensitive analysis")( + gridTest("preprocessMicrobatch keeps divergent target spelling under case-insensitive analysis")( Seq( ("id", "Value", "ID", "value"), ("ID", "value", "id", "Value") @@ -800,7 +800,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { } } - test("preprocessMicrobatch keeps distinct case-sensitive columns") { + test("preprocessMicrobatch keeps distinct case-sensitive columns between target " + + "and microbatch") { withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { val batchSchema = new StructType() .add("id", IntegerType) @@ -828,7 +829,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { } } - test("preprocessMicrobatch uses target spelling recursively under case-insensitive analysis") { + test("preprocessMicrobatch keeps nested target field spelling recursively under " + + "case-insensitive analysis") { withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { val batchSchema = new StructType() .add("ID", IntegerType) From a563f15468a0fbd0e71d9fd6c08cf0eb56eb1a9e Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Thu, 10 Sep 2026 22:27:20 +0000 Subject: [PATCH 04/15] cleanup whitespace --- .../spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala index 36060e831e32b..124dfba9757ab 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala @@ -70,7 +70,7 @@ case class Scd2ForeachBatchHandler( val targetTableDf = batchDf.sparkSession.read.table(targetTableIdentifier.quotedString) val auxTableDf = batchDf.sparkSession.read.table(auxiliaryTableIdentifier.quotedString) - + val preprocessedBatchDf = batchProcessor.preprocessMicrobatch( microbatchDf = batchDf, targetTableDf = targetTableDf From 7761b72ee10fd3ed544825d91dab35e066531bf2 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 8 Sep 2026 04:33:55 +0000 Subject: [PATCH 05/15] cleanup and unify helpers to Scd2VersionMap --- .../autocdc/Scd2BatchProcessor.scala | 87 +++- .../pipelines/autocdc/Scd2VersionMap.scala | 77 +++- .../autocdc/Scd2VersionMapSuite.scala | 419 ++++++++++++++++++ 3 files changed, 566 insertions(+), 17 deletions(-) create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index de06cdc61c442..5bc856e676299 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala @@ -124,9 +124,9 @@ case class Scd2BatchProcessor( * * Step ordering is load-bearing: the row-extension steps reference user data columns that * target-column selection is allowed to drop. Selection therefore runs after those extensions, - * followed by target-schema alignment. Unlike SCD1, no per-key deduplication step is performed - * here - SCD2 preserves every event as part of the row's history, including byte-identical - * full-event duplicates. + * followed by target-schema alignment and version-map population. Unlike SCD1, no per-key + * deduplication step is performed here - SCD2 preserves every event as part of the row's history, + * including byte-identical full-event duplicates. * * Duplicate event elimination (e.g., collapsing two identical events at the same sequence), * whether across microbatches or within the same microbatch, is the responsibility of @@ -151,6 +151,7 @@ case class Scd2BatchProcessor( .transform(extendMicrobatchRowsWithCdcMetadata) .transform(projectTargetColumnsOntoMicrobatch) .transform(alignMicrobatchToTargetSchema(_, targetTableDf)) + .transform(extendMicrobatchRowsWithVersionMap) } /** @@ -189,21 +190,62 @@ case class Scd2BatchProcessor( /** * Project the operational CDC metadata column carrying the literal event sequence. Downstream * merges rely on it to preserve original event lineage regardless of how rows start/end-at are - * coalesced. + * coalesced. The version map is initially null; [[extendMicrobatchRowsWithVersionMap]] populates + * it after column selection runs. */ private def extendMicrobatchRowsWithCdcMetadata(microbatchDf: DataFrame): DataFrame = { microbatchDf.withColumn( colName = AutoCdcReservedNames.cdcMetadataColName, col = Scd2BatchProcessor.constructCdcMetadataCol( recordStartAt = changeArgs.sequencing, - // TODO (SPARK-59183): actually populate version map according to ignore-null selection and - // actual authorship in microbatch. versionMap = F.lit(null), sequencingType = resolvedSequencingType ) ) } + /** + * Populates the version map on each microbatch row, recording which leaves the event authored. + * Null leaves in ignore-null columns get a `false` entry (declined); null leaves in other + * columns get `true` (authored null); non-null leaves need no entry. No-op when ignore-null + * is off. + * + * Must run after [[projectTargetColumnsOntoMicrobatch]], because the eligible schema is + * computed from the post-selection schema. + */ + private def extendMicrobatchRowsWithVersionMap(projectedDf: DataFrame): DataFrame = + changeArgs.ignoreNullSelection match { + case None => projectedDf + case Some(ignoreNullSelection) => + val cdcMetadataCol = F.col(AutoCdcReservedNames.cdcMetadataColName) + val resolver = projectedDf.sparkSession.sessionState.conf.resolver + val schemaEligibleForNullAuthorshipTracking = + Scd2BatchProcessor.computeUserDataSchema( + schema = projectedDf.schema, + changeArgs = changeArgs, + resolver = resolver + ) + + projectedDf.withColumn( + // Update the existing CDC metadata column via replace semantics when projecting a column + // with the same name. + colName = AutoCdcReservedNames.cdcMetadataColName, + col = Scd2BatchProcessor.constructCdcMetadataCol( + // Copy the same recordStartAt already computed from when the CDC metadata column was + // first projected. + recordStartAt = Scd2BatchProcessor.recordStartAtOf(cdcMetadataCol), + // Construct the version map for this row since ignore-null is being used. + versionMap = Scd2VersionMap.buildVersionMap( + schema = schemaEligibleForNullAuthorshipTracking, + ignoreNullSelection = ignoreNullSelection, + resolver = resolver + ), + // Sequencing type is unchanged from when the CDC metadata column was first projected. + sequencingType = resolvedSequencingType + ) + ) + } + /** * Apply the user's target column selection while preserving the SCD2 framework columns; the * latter are required by downstream merges and persisted to both the auxiliary and target @@ -1460,23 +1502,38 @@ object Scd2BatchProcessor { private[pipelines] def computeTrackedHistoryColumns( schema: StructType, changeArgs: ChangeArgs, - resolver: Resolver): Seq[String] = { - val keyColNames = changeArgs.keys.map(_.name) - - val eligibleSchema = StructType(schema.fields.filterNot { field => - reservedFrameworkColNames.exists(resolver(_, field.name)) || - keyColNames.exists(resolver(_, field.name)) - }) - + resolver: Resolver): Seq[String] = ColumnSelection .applyToSchema( schemaName = "trackHistorySelection", - schema = eligibleSchema, + schema = computeUserDataSchema(schema, changeArgs, resolver), columnSelection = changeArgs.trackHistorySelection, resolver = resolver ) .fieldNames .toImmutableArraySeq + + /** + * The subset of `schema` that is user data a column selection may act on: every field that is + * neither a framework reserved column nor one of [[ChangeArgs.keys]]. Field order is preserved. + * + * Both [[ChangeArgs.trackHistorySelection]] and [[ChangeArgs.ignoreNullSelection]] resolve + * against this, so an exclude-list in either cannot pick up a key or a framework column, and + * an include-list naming one fails as not found. + * + * `schema` is expected to have already been narrowed by [[ChangeArgs.columnSelection]], which + * happens once per microbatch in [[Scd2BatchProcessor.projectTargetColumnsOntoMicrobatch]]; + * this method does not re-apply it. + */ + private[pipelines] def computeUserDataSchema( + schema: StructType, + changeArgs: ChangeArgs, + resolver: Resolver): StructType = { + val keyColNames = changeArgs.keys.map(_.name) + StructType(schema.fields.filterNot { field => + reservedFrameworkColNames.exists(resolver(_, field.name)) || + keyColNames.exists(resolver(_, field.name)) + }) } /** diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala index 5926d9e108ad0..752915eab512a 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala @@ -17,7 +17,10 @@ package org.apache.spark.sql.pipelines.autocdc -import org.apache.spark.sql.types.{BooleanType, MapType, StringType} +import org.apache.spark.sql.{functions => F, Column} +import org.apache.spark.sql.catalyst.analysis.Resolver +import org.apache.spark.sql.catalyst.util.QuotingUtils +import org.apache.spark.sql.types.{BooleanType, MapType, StringType, StructType} /** * Per-row column authorship tracker for SCD2 ignore-null semantics. @@ -82,10 +85,80 @@ private[pipelines] object Scd2VersionMap { * must be formatted by [[org.apache.spark.sql.catalyst.util.QuotingUtils.quoted]] to * ensure segments that need quoting are back-tick escaped. * - * Values indicate authorship. I.e, `true` => authored-null, `false` => unauthored-null. + * Values indicate authorship: `true` means authored-null, `false` means unauthored-null. + * Null values never appear in the map. * * Lack of entry in the map for a null-valued leaf column implies the column was * schema-evolved with an unauthored-null. */ def mapType: MapType = MapType(StringType, BooleanType, valueContainsNull = false) + + /** + * Enumerates every leaf path in `schema`, in schema order, as its sequence of name parts. + * Structs unfold recursively; every other type (including arrays and maps) is an opaque leaf. + */ + private[autocdc] def extractLeafPaths(schema: StructType): Seq[Seq[String]] = + schema.fields.toSeq.flatMap { field => + field.dataType match { + case nested: StructType => + extractLeafPaths(nested).map(field.name +: _) + case _ => Seq(Seq(field.name)) + } + } + + /** + * Joins a multi-part name into a single dot-delimited string, backtick-quoting any segment + * that contains special characters, via [[QuotingUtils.quoted]]. + */ + private[autocdc] def quotedPath(path: Seq[String]): String = + QuotingUtils.quoted(path.toArray) + + /** + * Builds the ingest-time version map column for a microbatch. Each row's map records which + * null leaves are authored vs declined, based on the active ignore-null selection. + * + * @param schema The schema whose leaves the version map covers. Null-authorship is tracked + * for every leaf column in this schema, as per the version map contract. + * @param ignoreNullSelection The ignore-null column selection this schema is being ingested + * under. + * @param resolver Case-sensitivity resolver for column name matching. + * @return A [[Column]] of [[mapType]] schema. + */ + def buildVersionMap( + schema: StructType, + ignoreNullSelection: ColumnSelection, + resolver: Resolver): Column = { + val ignoreNullColumns = ColumnSelection.applyToSchema( + schemaName = "ignoreNullSelection", + schema = schema, + columnSelection = Some(ignoreNullSelection), + resolver = resolver + ) + val ignoreNullLeafPathsQuoted = + extractLeafPaths(ignoreNullColumns).map(quotedPath).toSet + + // For each leaf, build a nullable struct (key, value). The struct is non-null only when + // the leaf column's runtime value is null (meaning the leaf needs a version map entry). + // The value is a non-nullable BooleanType literal indicating authorship: true if the null + // is authored, false if declined. + val candidateEntries = extractLeafPaths(schema).map { path => + val leafPathQuoted = quotedPath(path) + val isIgnoreNullLeaf = ignoreNullLeafPathsQuoted.contains(leafPathQuoted) + val leafIsNull = F.col(leafPathQuoted).isNull + + // If the leaf is not null, this candidate entry will simply resolve to null and will not be + // added to the version map during construction below. + F.when(leafIsNull, F.struct( + F.lit(leafPathQuoted).as("key"), + F.lit(!isIgnoreNullLeaf).as("value") + )) + } + + if (candidateEntries.isEmpty) { + F.map().cast(mapType) + } else { + val nonNullEntries = F.filter(F.array(candidateEntries: _*), (e: Column) => e.isNotNull) + F.map_from_entries(nonNullEntries) + } + } } diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala new file mode 100644 index 0000000000000..d7f4ec61cfe44 --- /dev/null +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala @@ -0,0 +1,419 @@ +/* + * 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.autocdc + +import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types._ + +class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { + + private def resolver = spark.sessionState.conf.resolver + + // ---- Schema helpers ---- + + private val flatSchema = new StructType() + .add("a", IntegerType) + .add("b", StringType) + .add("c", DoubleType) + + private val nestedSchema = new StructType() + .add("x", IntegerType) + .add("address", new StructType() + .add("city", StringType) + .add("zip", IntegerType)) + + private val deeplyNestedSchema = new StructType() + .add("top", new StructType() + .add("mid", new StructType() + .add("leaf", StringType))) + + private val arrayAndMapSchema = new StructType() + .add("tags", ArrayType(StringType)) + .add("props", MapType(StringType, IntegerType)) + .add("plain", IntegerType) + + // Schemas with special-char leaves nested inside a normal top-level struct, so that + // ColumnSelection (which operates on top-level field names) can still select them. + private val specialCharSchema = new StructType() + .add("normal", IntegerType) + .add("wrapper", new StructType() + .add("has space", StringType)) + + private val periodInNameSchema = new StructType() + .add("wrapper", new StructType() + .add("a.b", IntegerType)) + .add("c", StringType) + + private val hyphenInNameSchema = new StructType() + .add("wrapper", new StructType() + .add("col-one", IntegerType)) + .add("col_two", StringType) + + // ---- Row helper ---- + + private def singleRow(schema: StructType)(values: Any*) = + spark.createDataFrame( + spark.sparkContext.parallelize(Seq(Row.fromSeq(values))), schema) + + // ========================================================================= + // extractLeafPaths + // ========================================================================= + + test("extractLeafPaths - flat columns produce single-element paths") { + assert(Scd2VersionMap.extractLeafPaths(flatSchema) === + Seq(Seq("a"), Seq("b"), Seq("c"))) + } + + test("extractLeafPaths - nested struct produces only leaf paths, not intermediaries") { + assert(Scd2VersionMap.extractLeafPaths(nestedSchema) === + Seq(Seq("x"), Seq("address", "city"), Seq("address", "zip"))) + } + + test("extractLeafPaths - deeply nested struct produces full multi-part paths") { + assert(Scd2VersionMap.extractLeafPaths(deeplyNestedSchema) === + Seq(Seq("top", "mid", "leaf"))) + } + + test("extractLeafPaths - arrays and maps are opaque leaves") { + assert(Scd2VersionMap.extractLeafPaths(arrayAndMapSchema) === + Seq(Seq("tags"), Seq("props"), Seq("plain"))) + } + + test("extractLeafPaths - empty schema produces empty seq") { + assert(Scd2VersionMap.extractLeafPaths(new StructType()) === Seq.empty) + } + + // ========================================================================= + // quotedPath + // ========================================================================= + + test("quotedPath - simple names are not quoted") { + assert(Scd2VersionMap.quotedPath(Seq("a")) === "a") + assert(Scd2VersionMap.quotedPath(Seq("address", "city")) === "address.city") + } + + test("quotedPath - name containing a period is backtick-quoted") { + assert(Scd2VersionMap.quotedPath(Seq("a.b")) === "`a.b`") + assert(Scd2VersionMap.quotedPath(Seq("x", "a.b")) === "x.`a.b`") + } + + test("quotedPath - name containing a space is backtick-quoted") { + assert(Scd2VersionMap.quotedPath(Seq("has space")) === "`has space`") + } + + test("quotedPath - name containing a hyphen is backtick-quoted") { + assert(Scd2VersionMap.quotedPath(Seq("col-one")) === "`col-one`") + } + + // ========================================================================= + // mapType + // ========================================================================= + + test("mapType is Map(String, Boolean)") { + assert(Scd2VersionMap.mapType === + MapType(StringType, BooleanType, valueContainsNull = false)) + } + + // ========================================================================= + // buildVersionMap: version map contract cases + // ========================================================================= + + // Contract case 1: null in event + not part of ignore-null -> (column, true) = authored null + test("contract case 1 - null leaf not in ignore-null selection is authored (true)") { + val df = singleRow(flatSchema)(null, "hello", 2.0) + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("b"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + // a is null + not in ignore-null -> true (authored null) + checkAnswer(result, Row(Map("a" -> true))) + } + + // Contract case 2: null in event + part of ignore-null -> (column, false) = unauthored null + test("contract case 2 - null leaf in ignore-null selection is declined (false)") { + val df = singleRow(flatSchema)(null, "hello", 2.0) + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + // a is null + in ignore-null -> false (declined) + checkAnswer(result, Row(Map("a" -> false))) + } + + // Contract case 3: column not in the event schema (schema evolution later adds it with null). + // At ingest time this manifests as no entry for the column. We verify absence by constructing + // a narrower schema that omits the column. + test("contract case 3 - column absent from schema has no entry (schema evolution)") { + // Simulate ingest with a narrower schema that does not yet include column "b". + val narrowSchema = new StructType().add("a", IntegerType) + val df = singleRow(narrowSchema)(null) + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(narrowSchema, selection, resolver).as("vm")) + // Only "a" appears (declined); a future column "b" added by schema evolution has no entry. + checkAnswer(result, Row(Map("a" -> false))) + } + + // Non-null values are always considered authored and produce no entry. + test("contract - non-null values produce no entry regardless of ignore-null membership") { + val df = singleRow(flatSchema)(1, "hello", 2.0) + val selection = ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("a"), UnqualifiedColumnName("b"), UnqualifiedColumnName("c"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map.empty[String, Boolean])) + } + + // ========================================================================= + // buildVersionMap: flat schema variations + // ========================================================================= + + test("flat schema - all null, none in ignore-null -> all authored (true)") { + val df = singleRow(flatSchema)(null, null, null) + val selection = ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("a"), UnqualifiedColumnName("b"), UnqualifiedColumnName("c"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map("a" -> true, "b" -> true, "c" -> true))) + } + + test("flat schema - all null, all in ignore-null -> all declined (false)") { + val df = singleRow(flatSchema)(null, null, null) + val selection = ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("a"), UnqualifiedColumnName("b"), UnqualifiedColumnName("c"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map("a" -> false, "b" -> false, "c" -> false))) + } + + test("flat schema - mixed nulls with partial ignore-null") { + // a=null (ignore-null), b="hi" (non-null), c=null (not ignore-null) + val df = singleRow(flatSchema)(null, "hi", null) + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map("a" -> false, "c" -> true))) + } + + test("flat schema - multiple rows produce independent per-row maps") { + val df = spark.createDataFrame( + spark.sparkContext.parallelize(Seq( + Row(1, null, 3.0), + Row(null, "x", null))), + flatSchema) + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("b"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + checkAnswer(result, Seq( + Row(Map("b" -> false)), // b=null + ignore-null + Row(Map("a" -> true, "c" -> true)))) // a,c=null + not ignore-null + } + + // ========================================================================= + // buildVersionMap: nested schemas + // ========================================================================= + + test("nested schema - null nested leaf tracked with dotted key") { + // x=1, address.city=null, address.zip=100 + // ColumnSelection operates on top-level fields; include "address" struct. + val df = singleRow(nestedSchema)(1, Row(null, 100)) + val selection = ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("address"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(nestedSchema, selection, resolver).as("vm")) + // address.city is null + ignore-null leaf -> false + // address.zip is null-checked but non-null -> no entry + checkAnswer(result, Row(Map("address.city" -> false))) + } + + test("nested schema - entire struct null makes all nested leaves null") { + val df = singleRow(nestedSchema)(null, null) + // Include "address" in ignore-null; "x" is not included. + val selection = ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("address"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(nestedSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map( + "x" -> true, // null + not ignore-null -> authored + "address.city" -> false, // null + ignore-null -> declined + "address.zip" -> false))) // null + ignore-null -> declined + } + + test("deeply nested schema - three-level path tracked correctly") { + val df = singleRow(deeplyNestedSchema)(Row(Row(null))) + // Include the top-level "top" struct -> all leaves under it are ignore-null. + val selection = ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("top"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(deeplyNestedSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map("top.mid.leaf" -> false))) + } + + // ========================================================================= + // buildVersionMap: arrays and maps treated as opaque leaves + // ========================================================================= + + test("array and map columns are tracked as opaque leaves") { + // tags=null, props=null, plain=null + val df = singleRow(arrayAndMapSchema)(null, null, null) + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("tags"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(arrayAndMapSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map( + "tags" -> false, // null + ignore-null -> declined + "props" -> true, // null + not ignore-null -> authored + "plain" -> true))) // null + not ignore-null -> authored + } + + // ========================================================================= + // buildVersionMap: special character column names (backtick quoting) + // ========================================================================= + + test("column name with space is backtick-quoted in version map key") { + // normal=null, wrapper."has space"=null + val df = singleRow(specialCharSchema)(null, Row(null)) + // Include "wrapper" struct -> its leaf "has space" is ignore-null. + val selection = ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("wrapper"))) + val result = df.select( + Scd2VersionMap.buildVersionMap( + specialCharSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map( + "normal" -> true, + "wrapper.`has space`" -> false))) + } + + test("column name with period is backtick-quoted in version map key") { + // wrapper."a.b"=null, c=null + val df = singleRow(periodInNameSchema)(Row(null), null) + val selection = ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("wrapper"))) + val result = df.select( + Scd2VersionMap.buildVersionMap( + periodInNameSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map( + "wrapper.`a.b`" -> false, + "c" -> true))) + } + + test("column name with hyphen is backtick-quoted in version map key") { + // wrapper."col-one"=null, col_two=null + val df = singleRow(hyphenInNameSchema)(Row(null), null) + val selection = ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("wrapper"))) + val result = df.select( + Scd2VersionMap.buildVersionMap( + hyphenInNameSchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map( + "wrapper.`col-one`" -> false, + "col_two" -> true))) + } + + // ========================================================================= + // buildVersionMap: empty schema + // ========================================================================= + + test("empty schema produces an empty version map") { + val emptySchema = new StructType() + val df = singleRow(emptySchema)() + val selection = ColumnSelection.ExcludeColumns(Seq.empty) + val result = df.select( + Scd2VersionMap.buildVersionMap(emptySchema, selection, resolver).as("vm")) + checkAnswer(result, Row(Map.empty[String, Boolean])) + } + + // ========================================================================= + // buildVersionMap: ExcludeColumns variations + // ========================================================================= + + test("ExcludeColumns - excluded column is NOT ignore-null, others are") { + val df = singleRow(flatSchema)(null, null, null) + // ExcludeColumns("b") means ignore-null = everything except b. + val selection = ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("b"))) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + // a,c are in ignore-null -> declined (false); b is NOT -> authored (true). + checkAnswer(result, Row(Map("a" -> false, "b" -> true, "c" -> false))) + } + + test("ExcludeColumns - empty exclude list -> ignore-null covers all columns") { + val df = singleRow(flatSchema)(null, null, null) + val selection = ColumnSelection.ExcludeColumns(Seq.empty) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + // Exclude nothing -> ignore-null applies to all columns -> all nulls are declined. + checkAnswer(result, Row(Map("a" -> false, "b" -> false, "c" -> false))) + } + + // ========================================================================= + // buildVersionMap: IncludeColumns variations + // ========================================================================= + + test("IncludeColumns - empty include list means no columns are ignore-null") { + val df = singleRow(flatSchema)(null, null, null) + val selection = ColumnSelection.IncludeColumns(Seq.empty) + val result = df.select( + Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) + // No columns included in ignore-null -> all nulls are authored. + checkAnswer(result, Row(Map("a" -> true, "b" -> true, "c" -> true))) + } + + // ========================================================================= + // buildVersionMap: case sensitivity + // ========================================================================= + + test("case-insensitive resolver matches ignore-null columns regardless of case") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + val df = singleRow(flatSchema)(null, null, null) + // Selection uses uppercase "A" but schema has lowercase "a". + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("A"))) + val caseInsensitiveResolver = spark.sessionState.conf.resolver + val result = df.select( + Scd2VersionMap.buildVersionMap( + flatSchema, selection, caseInsensitiveResolver).as("vm")) + checkAnswer(result, Row(Map("a" -> false, "b" -> true, "c" -> true))) + } + } + + test("case-sensitive resolver does not match differently-cased column names") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + val caseSensitiveResolver = spark.sessionState.conf.resolver + val df = singleRow(flatSchema)(null, null, null) + // Selection uses uppercase "A" but schema has lowercase "a" -> "A" is not found. + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("A"))) + val e = intercept[Exception] { + df.select( + Scd2VersionMap.buildVersionMap( + flatSchema, selection, caseSensitiveResolver).as("vm")).collect() + } + assert(e.getMessage.contains("A")) + } + } + + test("case-sensitive resolver matches exact-case column names") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + val caseSensitiveResolver = spark.sessionState.conf.resolver + val df = singleRow(flatSchema)(null, null, null) + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a"))) + val result = df.select( + Scd2VersionMap.buildVersionMap( + flatSchema, selection, caseSensitiveResolver).as("vm")) + checkAnswer(result, Row(Map("a" -> false, "b" -> true, "c" -> true))) + } + } +} From 668f2d9c4efc2f441c8def77ba68410287392c2a Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 8 Sep 2026 05:12:16 +0000 Subject: [PATCH 06/15] delete rows in microbatch receive null version maps --- .../autocdc/Scd2BatchProcessor.scala | 26 ++--- .../autocdc/Scd2BatchProcessorSuite.scala | 95 +++++++++++++++++++ 2 files changed, 110 insertions(+), 11 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index 5bc856e676299..7c464dbbcdebb 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala @@ -210,6 +210,10 @@ case class Scd2BatchProcessor( * columns get `true` (authored null); non-null leaves need no entry. No-op when ignore-null * is off. * + * Delete-encoded rows (those matching [[ChangeArgs.deleteCondition]]) receive a null version + * map: their data-column values are not part of the SCD2 contract, so authorship tracking + * is not applicable. + * * Must run after [[projectTargetColumnsOntoMicrobatch]], because the eligible schema is * computed from the post-selection schema. */ @@ -219,28 +223,28 @@ case class Scd2BatchProcessor( case Some(ignoreNullSelection) => val cdcMetadataCol = F.col(AutoCdcReservedNames.cdcMetadataColName) val resolver = projectedDf.sparkSession.sessionState.conf.resolver - val schemaEligibleForNullAuthorshipTracking = + val schemaEligibleForNullAuthorshipTracking = Scd2BatchProcessor.computeUserDataSchema( schema = projectedDf.schema, changeArgs = changeArgs, resolver = resolver ) + val isUpsertRow = !changeArgs.deleteCondition.getOrElse(F.lit(false)) projectedDf.withColumn( - // Update the existing CDC metadata column via replace semantics when projecting a column - // with the same name. colName = AutoCdcReservedNames.cdcMetadataColName, col = Scd2BatchProcessor.constructCdcMetadataCol( - // Copy the same recordStartAt already computed from when the CDC metadata column was - // first projected. recordStartAt = Scd2BatchProcessor.recordStartAtOf(cdcMetadataCol), - // Construct the version map for this row since ignore-null is being used. - versionMap = Scd2VersionMap.buildVersionMap( - schema = schemaEligibleForNullAuthorshipTracking, - ignoreNullSelection = ignoreNullSelection, - resolver = resolver + // Only upsert rows get a populated version map. By convention, delete-encoded rows + // always maintain a null version map. + versionMap = F.when( + isUpsertRow, + Scd2VersionMap.buildVersionMap( + schema = schemaEligibleForNullAuthorshipTracking, + ignoreNullSelection = ignoreNullSelection, + resolver = resolver + ) ), - // Sequencing type is unchanged from when the CDC metadata column was first projected. sequencingType = resolvedSequencingType ) ) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala index 6b0f2d3d866cf..ee667d5c6d0cd 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala @@ -872,6 +872,101 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { } } + gridTest("preprocessMicrobatch leaves delete-representing rows with a null version map")( + Seq( + // Delete condition is specified; all non-matching rows should be treated as upsert. + Some(F.col("is_delete")), + // Delete condition is unspecified; all microbatch rows should be treated as upsert. + None + ) + ) { case (deleteCondition) => + val schema = new StructType() + .add("id", IntegerType) + .add("value", StringType) + .add("seq", LongType) + .add("is_delete", BooleanType) + + val batch = microbatchOf(schema)( + Row(1, null, 10L, false), // upsert with null value + Row(1, "a", 20L, false), // upsert with non-null value + Row(1, null, 30L, true) // delete iff deleteCondition is set + ) + + val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + deleteCondition = deleteCondition, + ignoreNullSelection = + Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("value")))) + ), + resolvedSequencingType = LongType + ) + + val result = preprocessMicrobatch(processor, batch) + + val versionMaps = result.select( + F.col("seq"), + Scd2BatchProcessor.versionMapOf( + F.col(AutoCdcReservedNames.cdcMetadataColName) + ).as("vm") + ) + + // Upsert rows always get a populated version map. The third row is a delete (null + // version map) only when deleteCondition is set; otherwise it is an upsert too. + val expectedDeleteRowMap: Any = + if (deleteCondition.isDefined) null else Map("value" -> false) + + checkAnswer( + df = versionMaps, + expectedAnswer = Seq( + Row(10L, Map("value" -> false)), + Row(20L, Map.empty[String, Boolean]), + Row(30L, expectedDeleteRowMap) + ) + ) + } + + test("preprocessMicrobatch leaves version map null for all rows when ignore null is off") { + val schema = new StructType() + .add("id", IntegerType) + .add("value", StringType) + .add("seq", LongType) + .add("is_delete", BooleanType) + + val batch = microbatchOf(schema)( + Row(1, null, 10L, false), + Row(1, null, 20L, true) + ) + + val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + deleteCondition = Some(F.col("is_delete")), + // None ignore-null selection should be treated as ignore-null off. + ignoreNullSelection = None + ), + resolvedSequencingType = LongType + ) + + val result = preprocessMicrobatch(processor, batch) + + val versionMaps = result.select( + Scd2BatchProcessor.versionMapOf( + F.col(AutoCdcReservedNames.cdcMetadataColName) + ).as("vm") + ) + + // ignoreNullSelection is None -> version map is null on every row. + checkAnswer( + df = versionMaps, + expectedAnswer = Seq(Row(null), Row(null)) + ) + } + // =============== computeMinimumSequencePerKey tests =============== test("computeMinimumSequencePerKey returns one row per distinct key and aggregates across " + From b15794df629e7beb1e026ceb8112f543f283e960 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 8 Sep 2026 16:35:27 +0000 Subject: [PATCH 07/15] document risk for changing ignore-null selection between microbatch retries --- .../spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala | 3 +++ 1 file changed, 3 insertions(+) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index 7c464dbbcdebb..22977d3f4100e 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala @@ -216,6 +216,9 @@ case class Scd2BatchProcessor( * * Must run after [[projectTargetColumnsOntoMicrobatch]], because the eligible schema is * computed from the post-selection schema. + * + * TODO(SPARK-59343): decide how to handle the ignore-null selection changing between + * partial-retry attempts of the same microbatch. */ private def extendMicrobatchRowsWithVersionMap(projectedDf: DataFrame): DataFrame = changeArgs.ignoreNullSelection match { From f796137d242b1e85168e5755924961c2a94831a9 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 8 Sep 2026 19:18:02 +0000 Subject: [PATCH 08/15] use withField --- .../autocdc/Scd2BatchProcessor.scala | 24 ++++++++----------- 1 file changed, 10 insertions(+), 14 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index 22977d3f4100e..8aa68c7ce1e6c 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala @@ -232,24 +232,20 @@ case class Scd2BatchProcessor( changeArgs = changeArgs, resolver = resolver ) + + // Only upsert rows get a populated version map. By convention, delete-encoded + // rows always maintain a null version map. val isUpsertRow = !changeArgs.deleteCondition.getOrElse(F.lit(false)) + val versionMap = F.when(isUpsertRow, Scd2VersionMap.buildVersionMap( + schema = schemaEligibleForNullAuthorshipTracking, + ignoreNullSelection = ignoreNullSelection, + resolver = resolver + )) projectedDf.withColumn( colName = AutoCdcReservedNames.cdcMetadataColName, - col = Scd2BatchProcessor.constructCdcMetadataCol( - recordStartAt = Scd2BatchProcessor.recordStartAtOf(cdcMetadataCol), - // Only upsert rows get a populated version map. By convention, delete-encoded rows - // always maintain a null version map. - versionMap = F.when( - isUpsertRow, - Scd2VersionMap.buildVersionMap( - schema = schemaEligibleForNullAuthorshipTracking, - ignoreNullSelection = ignoreNullSelection, - resolver = resolver - ) - ), - sequencingType = resolvedSequencingType - ) + col = cdcMetadataCol + .withField(Scd2BatchProcessor.versionMapFieldName, versionMap) ) } From 7a9e4acbacd4ea05ea6b6100f6895b95948eb93e Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 8 Sep 2026 20:19:46 +0000 Subject: [PATCH 09/15] use JSON serialization for version map keys --- .../pipelines/autocdc/Scd2VersionMap.scala | 31 ++-- .../autocdc/Scd2BatchProcessorSuite.scala | 5 +- .../autocdc/Scd2VersionMapSuite.scala | 148 ++++++++++++------ 3 files changed, 122 insertions(+), 62 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala index 752915eab512a..19bd0dadea2ff 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala @@ -17,6 +17,9 @@ package org.apache.spark.sql.pipelines.autocdc +import org.json4s.JsonAST.{JArray, JString} +import org.json4s.jackson.JsonMethods.compact + import org.apache.spark.sql.{functions => F, Column} import org.apache.spark.sql.catalyst.analysis.Resolver import org.apache.spark.sql.catalyst.util.QuotingUtils @@ -80,10 +83,10 @@ private[pipelines] object Scd2VersionMap { /** * Schema of the version map: `Map(String, Boolean)`. * - * Keys are dot-delimited paths to *leaf* columns that received a null value in their - * corresponding upsert event (e.g. `"address.city"`, `` "`has space`.city" ``). Paths - * must be formatted by [[org.apache.spark.sql.catalyst.util.QuotingUtils.quoted]] to - * ensure segments that need quoting are back-tick escaped. + * Keys are compact JSON arrays containing the canonical name parts of *leaf* columns that + * received a null value in their corresponding upsert event (e.g. `["address","city"]`). + * Keeping name parts separate distinguishes a nested path from a column whose name contains + * dots and keeps persisted keys independent of SQL identifier quoting rules. * * Values indicate authorship: `true` means authored-null, `false` means unauthored-null. * Null values never appear in the map. @@ -106,12 +109,9 @@ private[pipelines] object Scd2VersionMap { } } - /** - * Joins a multi-part name into a single dot-delimited string, backtick-quoting any segment - * that contains special characters, via [[QuotingUtils.quoted]]. - */ - private[autocdc] def quotedPath(path: Seq[String]): String = - QuotingUtils.quoted(path.toArray) + /** Encodes a leaf path as the compact JSON string persisted as its version map key. */ + private[autocdc] def encodePath(path: Seq[String]): String = + compact(JArray(path.map(JString(_)).toList)) /** * Builds the ingest-time version map column for a microbatch. Each row's map records which @@ -134,22 +134,21 @@ private[pipelines] object Scd2VersionMap { columnSelection = Some(ignoreNullSelection), resolver = resolver ) - val ignoreNullLeafPathsQuoted = - extractLeafPaths(ignoreNullColumns).map(quotedPath).toSet + val ignoreNullLeafPaths = extractLeafPaths(ignoreNullColumns).toSet // For each leaf, build a nullable struct (key, value). The struct is non-null only when // the leaf column's runtime value is null (meaning the leaf needs a version map entry). // The value is a non-nullable BooleanType literal indicating authorship: true if the null // is authored, false if declined. val candidateEntries = extractLeafPaths(schema).map { path => - val leafPathQuoted = quotedPath(path) - val isIgnoreNullLeaf = ignoreNullLeafPathsQuoted.contains(leafPathQuoted) - val leafIsNull = F.col(leafPathQuoted).isNull + val encodedPath = encodePath(path) + val isIgnoreNullLeaf = ignoreNullLeafPaths.contains(path) + val leafIsNull = F.col(QuotingUtils.quoteNameParts(path)).isNull // If the leaf is not null, this candidate entry will simply resolve to null and will not be // added to the version map during construction below. F.when(leafIsNull, F.struct( - F.lit(leafPathQuoted).as("key"), + F.lit(encodedPath).as("key"), F.lit(!isIgnoreNullLeaf).as("value") )) } diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala index ee667d5c6d0cd..4d4c63403a61a 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala @@ -915,13 +915,14 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // Upsert rows always get a populated version map. The third row is a delete (null // version map) only when deleteCondition is set; otherwise it is an upsert too. + val valueVersionMapKey = Scd2VersionMap.encodePath(Seq("value")) val expectedDeleteRowMap: Any = - if (deleteCondition.isDefined) null else Map("value" -> false) + if (deleteCondition.isDefined) null else Map(valueVersionMapKey -> false) checkAnswer( df = versionMaps, expectedAnswer = Seq( - Row(10L, Map("value" -> false)), + Row(10L, Map(valueVersionMapKey -> false)), Row(20L, Map.empty[String, Boolean]), Row(30L, expectedDeleteRowMap) ) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala index d7f4ec61cfe44..d49401582ec83 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala @@ -17,7 +17,10 @@ package org.apache.spark.sql.pipelines.autocdc -import org.apache.spark.sql.{QueryTest, Row} +import org.json4s.JsonAST.{JArray, JString} +import org.json4s.jackson.JsonMethods.parse + +import org.apache.spark.sql.{functions => F, QueryTest, Row} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types._ @@ -72,6 +75,8 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { spark.createDataFrame( spark.sparkContext.parallelize(Seq(Row.fromSeq(values))), schema) + private def encodedPath(path: String*): String = Scd2VersionMap.encodePath(path) + // ========================================================================= // extractLeafPaths // ========================================================================= @@ -101,25 +106,29 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { } // ========================================================================= - // quotedPath + // encodePath // ========================================================================= - test("quotedPath - simple names are not quoted") { - assert(Scd2VersionMap.quotedPath(Seq("a")) === "a") - assert(Scd2VersionMap.quotedPath(Seq("address", "city")) === "address.city") + test("encodePath - writes name parts as a compact JSON array") { + assert(Scd2VersionMap.encodePath(Seq("a")) === """["a"]""") + assert(Scd2VersionMap.encodePath(Seq("address", "city")) === + """["address","city"]""") } - test("quotedPath - name containing a period is backtick-quoted") { - assert(Scd2VersionMap.quotedPath(Seq("a.b")) === "`a.b`") - assert(Scd2VersionMap.quotedPath(Seq("x", "a.b")) === "x.`a.b`") + test("encodePath - distinguishes nested paths from names containing periods") { + assert(Scd2VersionMap.encodePath(Seq("a", "b")) === """["a","b"]""") + assert(Scd2VersionMap.encodePath(Seq("a.b")) === """["a.b"]""") } - test("quotedPath - name containing a space is backtick-quoted") { - assert(Scd2VersionMap.quotedPath(Seq("has space")) === "`has space`") + test("encodePath - escapes JSON delimiters and control characters") { + val encoded = Scd2VersionMap.encodePath( + Seq("quote\"", "back\\slash", "null" + 0.toChar + "byte")) + assert(encoded === "[\"quote\\\"\",\"back\\\\slash\",\"null\\" + "u0000byte\"]") } - test("quotedPath - name containing a hyphen is backtick-quoted") { - assert(Scd2VersionMap.quotedPath(Seq("col-one")) === "`col-one`") + test("encodePath - round trips arbitrary name parts") { + val path = Seq("", "a.b", "has space", "back`tick", "quote\"", "back\\slash") + assert(parse(Scd2VersionMap.encodePath(path)) === JArray(path.map(JString(_)).toList)) } // ========================================================================= @@ -142,7 +151,7 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) // a is null + not in ignore-null -> true (authored null) - checkAnswer(result, Row(Map("a" -> true))) + checkAnswer(result, Row(Map(encodedPath("a") -> true))) } // Contract case 2: null in event + part of ignore-null -> (column, false) = unauthored null @@ -152,7 +161,7 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) // a is null + in ignore-null -> false (declined) - checkAnswer(result, Row(Map("a" -> false))) + checkAnswer(result, Row(Map(encodedPath("a") -> false))) } // Contract case 3: column not in the event schema (schema evolution later adds it with null). @@ -166,7 +175,7 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(narrowSchema, selection, resolver).as("vm")) // Only "a" appears (declined); a future column "b" added by schema evolution has no entry. - checkAnswer(result, Row(Map("a" -> false))) + checkAnswer(result, Row(Map(encodedPath("a") -> false))) } // Non-null values are always considered authored and produce no entry. @@ -189,7 +198,10 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { Seq(UnqualifiedColumnName("a"), UnqualifiedColumnName("b"), UnqualifiedColumnName("c"))) val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) - checkAnswer(result, Row(Map("a" -> true, "b" -> true, "c" -> true))) + checkAnswer(result, Row(Map( + encodedPath("a") -> true, + encodedPath("b") -> true, + encodedPath("c") -> true))) } test("flat schema - all null, all in ignore-null -> all declined (false)") { @@ -198,7 +210,10 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { Seq(UnqualifiedColumnName("a"), UnqualifiedColumnName("b"), UnqualifiedColumnName("c"))) val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) - checkAnswer(result, Row(Map("a" -> false, "b" -> false, "c" -> false))) + checkAnswer(result, Row(Map( + encodedPath("a") -> false, + encodedPath("b") -> false, + encodedPath("c") -> false))) } test("flat schema - mixed nulls with partial ignore-null") { @@ -207,7 +222,7 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a"))) val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) - checkAnswer(result, Row(Map("a" -> false, "c" -> true))) + checkAnswer(result, Row(Map(encodedPath("a") -> false, encodedPath("c") -> true))) } test("flat schema - multiple rows produce independent per-row maps") { @@ -220,15 +235,18 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) checkAnswer(result, Seq( - Row(Map("b" -> false)), // b=null + ignore-null - Row(Map("a" -> true, "c" -> true)))) // a,c=null + not ignore-null + Row(Map(encodedPath("b") -> false)), // b=null + ignore-null + // a,c=null + not ignore-null + Row(Map( + encodedPath("a") -> true, + encodedPath("c") -> true)))) } // ========================================================================= // buildVersionMap: nested schemas // ========================================================================= - test("nested schema - null nested leaf tracked with dotted key") { + test("nested schema - null nested leaf tracked with multipart key") { // x=1, address.city=null, address.zip=100 // ColumnSelection operates on top-level fields; include "address" struct. val df = singleRow(nestedSchema)(1, Row(null, 100)) @@ -238,7 +256,7 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { Scd2VersionMap.buildVersionMap(nestedSchema, selection, resolver).as("vm")) // address.city is null + ignore-null leaf -> false // address.zip is null-checked but non-null -> no entry - checkAnswer(result, Row(Map("address.city" -> false))) + checkAnswer(result, Row(Map(encodedPath("address", "city") -> false))) } test("nested schema - entire struct null makes all nested leaves null") { @@ -249,9 +267,9 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(nestedSchema, selection, resolver).as("vm")) checkAnswer(result, Row(Map( - "x" -> true, // null + not ignore-null -> authored - "address.city" -> false, // null + ignore-null -> declined - "address.zip" -> false))) // null + ignore-null -> declined + encodedPath("x") -> true, // null + not ignore-null -> authored + encodedPath("address", "city") -> false, // null + ignore-null -> declined + encodedPath("address", "zip") -> false))) // null + ignore-null -> declined } test("deeply nested schema - three-level path tracked correctly") { @@ -261,7 +279,7 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { Seq(UnqualifiedColumnName("top"))) val result = df.select( Scd2VersionMap.buildVersionMap(deeplyNestedSchema, selection, resolver).as("vm")) - checkAnswer(result, Row(Map("top.mid.leaf" -> false))) + checkAnswer(result, Row(Map(encodedPath("top", "mid", "leaf") -> false))) } // ========================================================================= @@ -275,16 +293,16 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(arrayAndMapSchema, selection, resolver).as("vm")) checkAnswer(result, Row(Map( - "tags" -> false, // null + ignore-null -> declined - "props" -> true, // null + not ignore-null -> authored - "plain" -> true))) // null + not ignore-null -> authored + encodedPath("tags") -> false, // null + ignore-null -> declined + encodedPath("props") -> true, // null + not ignore-null -> authored + encodedPath("plain") -> true))) // null + not ignore-null -> authored } // ========================================================================= - // buildVersionMap: special character column names (backtick quoting) + // buildVersionMap: special character column names // ========================================================================= - test("column name with space is backtick-quoted in version map key") { + test("column name with space is encoded in version map key") { // normal=null, wrapper."has space"=null val df = singleRow(specialCharSchema)(null, Row(null)) // Include "wrapper" struct -> its leaf "has space" is ignore-null. @@ -294,11 +312,11 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { Scd2VersionMap.buildVersionMap( specialCharSchema, selection, resolver).as("vm")) checkAnswer(result, Row(Map( - "normal" -> true, - "wrapper.`has space`" -> false))) + encodedPath("normal") -> true, + encodedPath("wrapper", "has space") -> false))) } - test("column name with period is backtick-quoted in version map key") { + test("column name with period is encoded in version map key") { // wrapper."a.b"=null, c=null val df = singleRow(periodInNameSchema)(Row(null), null) val selection = ColumnSelection.IncludeColumns( @@ -307,11 +325,11 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { Scd2VersionMap.buildVersionMap( periodInNameSchema, selection, resolver).as("vm")) checkAnswer(result, Row(Map( - "wrapper.`a.b`" -> false, - "c" -> true))) + encodedPath("wrapper", "a.b") -> false, + encodedPath("c") -> true))) } - test("column name with hyphen is backtick-quoted in version map key") { + test("column name with hyphen is encoded in version map key") { // wrapper."col-one"=null, col_two=null val df = singleRow(hyphenInNameSchema)(Row(null), null) val selection = ColumnSelection.IncludeColumns( @@ -320,8 +338,35 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { Scd2VersionMap.buildVersionMap( hyphenInNameSchema, selection, resolver).as("vm")) checkAnswer(result, Row(Map( - "wrapper.`col-one`" -> false, - "col_two" -> true))) + encodedPath("wrapper", "col-one") -> false, + encodedPath("col_two") -> true))) + } + + test("special-character path parts survive version map encoding") { + val specialNames = + Seq("a.b", "has space", "back`tick", "quote\"", "back\\slash", "null" + 0.toChar + "byte") + val schema = new StructType() + .add("wrapper", StructType(specialNames.map(StructField(_, StringType)))) + val df = singleRow(schema)(Row.fromSeq(Seq.fill[Any](specialNames.size)(null))) + val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("wrapper"))) + + val encodedKeys = df + .select(F.map_keys(Scd2VersionMap.buildVersionMap(schema, selection, resolver))) + .head() + .getSeq[String](0) + val decodedPaths = encodedKeys.map { encodedKey => + parse(encodedKey) match { + case JArray(parts) => + parts.map { + case JString(part) => part + case other => fail(s"Expected a JSON string path part, but found $other") + } + case other => fail(s"Expected a JSON array version map key, but found $other") + } + } + + val expectedPaths = specialNames.map(name => Seq("wrapper", name)).toSet + assert(decodedPaths.map(_.toSeq).toSet === expectedPaths) } // ========================================================================= @@ -348,7 +393,10 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) // a,c are in ignore-null -> declined (false); b is NOT -> authored (true). - checkAnswer(result, Row(Map("a" -> false, "b" -> true, "c" -> false))) + checkAnswer(result, Row(Map( + encodedPath("a") -> false, + encodedPath("b") -> true, + encodedPath("c") -> false))) } test("ExcludeColumns - empty exclude list -> ignore-null covers all columns") { @@ -357,7 +405,10 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) // Exclude nothing -> ignore-null applies to all columns -> all nulls are declined. - checkAnswer(result, Row(Map("a" -> false, "b" -> false, "c" -> false))) + checkAnswer(result, Row(Map( + encodedPath("a") -> false, + encodedPath("b") -> false, + encodedPath("c") -> false))) } // ========================================================================= @@ -370,7 +421,10 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) // No columns included in ignore-null -> all nulls are authored. - checkAnswer(result, Row(Map("a" -> true, "b" -> true, "c" -> true))) + checkAnswer(result, Row(Map( + encodedPath("a") -> true, + encodedPath("b") -> true, + encodedPath("c") -> true))) } // ========================================================================= @@ -386,7 +440,10 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap( flatSchema, selection, caseInsensitiveResolver).as("vm")) - checkAnswer(result, Row(Map("a" -> false, "b" -> true, "c" -> true))) + checkAnswer(result, Row(Map( + encodedPath("a") -> false, + encodedPath("b") -> true, + encodedPath("c") -> true))) } } @@ -413,7 +470,10 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val result = df.select( Scd2VersionMap.buildVersionMap( flatSchema, selection, caseSensitiveResolver).as("vm")) - checkAnswer(result, Row(Map("a" -> false, "b" -> true, "c" -> true))) + checkAnswer(result, Row(Map( + encodedPath("a") -> false, + encodedPath("b") -> true, + encodedPath("c") -> true))) } } } From b642a2500b35ecbab779b27663c79b21677c942a Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 8 Sep 2026 22:05:33 +0000 Subject: [PATCH 10/15] cleanup --- .../autocdc/Scd2VersionMapSuite.scala | 42 +++++++++---------- 1 file changed, 19 insertions(+), 23 deletions(-) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala index d49401582ec83..660f184874d2d 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala @@ -20,7 +20,7 @@ package org.apache.spark.sql.pipelines.autocdc import org.json4s.JsonAST.{JArray, JString} import org.json4s.jackson.JsonMethods.parse -import org.apache.spark.sql.{functions => F, QueryTest, Row} +import org.apache.spark.sql.{functions => F, AnalysisException, QueryTest, Row} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types._ @@ -411,21 +411,6 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { encodedPath("c") -> false))) } - // ========================================================================= - // buildVersionMap: IncludeColumns variations - // ========================================================================= - - test("IncludeColumns - empty include list means no columns are ignore-null") { - val df = singleRow(flatSchema)(null, null, null) - val selection = ColumnSelection.IncludeColumns(Seq.empty) - val result = df.select( - Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm")) - // No columns included in ignore-null -> all nulls are authored. - checkAnswer(result, Row(Map( - encodedPath("a") -> true, - encodedPath("b") -> true, - encodedPath("c") -> true))) - } // ========================================================================= // buildVersionMap: case sensitivity @@ -452,13 +437,24 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { val caseSensitiveResolver = spark.sessionState.conf.resolver val df = singleRow(flatSchema)(null, null, null) // Selection uses uppercase "A" but schema has lowercase "a" -> "A" is not found. - val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("A"))) - val e = intercept[Exception] { - df.select( - Scd2VersionMap.buildVersionMap( - flatSchema, selection, caseSensitiveResolver).as("vm")).collect() - } - assert(e.getMessage.contains("A")) + val selection = + ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("A"))) + checkError( + exception = intercept[AnalysisException] { + df.select( + Scd2VersionMap.buildVersionMap( + flatSchema, selection, caseSensitiveResolver).as("vm") + ).collect() + }, + condition = "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA", + sqlState = "42703", + parameters = Map( + "caseSensitivity" -> "case-sensitive", + "schemaName" -> "ignoreNullSelection", + "missingColumns" -> "A", + "availableColumns" -> "a, b, c" + ) + ) } } From 1800e18fe8dfd511ad4c9ed6219d895bbb226633 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 8 Sep 2026 22:24:27 +0000 Subject: [PATCH 11/15] fix delete detection --- .../spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index 8aa68c7ce1e6c..263b6040d08ac 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala @@ -234,8 +234,10 @@ case class Scd2BatchProcessor( ) // Only upsert rows get a populated version map. By convention, delete-encoded - // rows always maintain a null version map. - val isUpsertRow = !changeArgs.deleteCondition.getOrElse(F.lit(false)) + // rows always maintain a null version map. We detect deletes via endAt rather + // than changeArgs.deleteCondition because column selection may have already + // dropped the column the delete condition references. + val isUpsertRow = F.col(Scd2BatchProcessor.endAtColName).isNull val versionMap = F.when(isUpsertRow, Scd2VersionMap.buildVersionMap( schema = schemaEligibleForNullAuthorshipTracking, ignoreNullSelection = ignoreNullSelection, From a463835a66769a14b1660b5c97d766798b28e36d Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 8 Sep 2026 23:21:14 +0000 Subject: [PATCH 12/15] self-review --- .../spark/sql/pipelines/autocdc/Scd2VersionMap.scala | 10 ++++++---- .../pipelines/autocdc/Scd2BatchProcessorSuite.scala | 4 ++++ 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala index 19bd0dadea2ff..6c7e344a92e6d 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala @@ -83,10 +83,12 @@ private[pipelines] object Scd2VersionMap { /** * Schema of the version map: `Map(String, Boolean)`. * - * Keys are compact JSON arrays containing the canonical name parts of *leaf* columns that - * received a null value in their corresponding upsert event (e.g. `["address","city"]`). - * Keeping name parts separate distinguishes a nested path from a column whose name contains - * dots and keeps persisted keys independent of SQL identifier quoting rules. + * Keys are compact JSON arrays of the name parts of *leaf* columns that received a null + * value in their upsert event (e.g. `["address","city"]`). Keeping + * name parts separate distinguishes a nested path from a column whose name contains dots + * and keeps persisted keys independent of SQL identifier quoting rules. Note: until + * SPARK-59347, these are microbatch-schema spellings, not necessarily canonical target + * spellings. * * Values indicate authorship: `true` means authored-null, `false` means unauthored-null. * Null values never appear in the map. diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala index 4d4c63403a61a..4f493dc9fc123 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala @@ -898,6 +898,10 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { sequencing = F.col("seq"), storedAsScdType = ScdType.Type2, deleteCondition = deleteCondition, + // Even if we drop `is_delete` from the output schema, delete-row detection should still + // work and the row should still receive a null version map. + columnSelection = Some(ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("is_delete")))), ignoreNullSelection = Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("value")))) ), From 1bfd55dd5dcf874fede721f10dbfd648da58d21b Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 9 Sep 2026 01:22:36 +0000 Subject: [PATCH 13/15] refactor out generic extractLeafPaths --- .../autocdc/AutoCdcSchemaUtils.scala | 36 ++++++++++ .../pipelines/autocdc/Scd2VersionMap.scala | 17 +---- .../autocdc/AutoCdcSchemaUtilsSuite.scala | 69 +++++++++++++++++++ .../autocdc/Scd2VersionMapSuite.scala | 28 -------- 4 files changed, 107 insertions(+), 43 deletions(-) create mode 100644 sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtils.scala create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtilsSuite.scala diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtils.scala new file mode 100644 index 0000000000000..e5015610cca23 --- /dev/null +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtils.scala @@ -0,0 +1,36 @@ +/* + * 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.autocdc + +import org.apache.spark.sql.types.StructType + +private[autocdc] object AutoCdcSchemaUtils { + + /** + * Enumerates every leaf path in `schema`, in schema order, as its sequence of name parts. + * Structs unfold recursively; every other type (including arrays and maps) is an opaque leaf. + */ + def extractLeafPaths(schema: StructType): Seq[Seq[String]] = + schema.fields.toSeq.flatMap { field => + field.dataType match { + case nested: StructType => + extractLeafPaths(nested).map(field.name +: _) + case _ => Seq(Seq(field.name)) + } + } +} diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala index 6c7e344a92e6d..9e1b7c3fac5a8 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala @@ -98,19 +98,6 @@ private[pipelines] object Scd2VersionMap { */ def mapType: MapType = MapType(StringType, BooleanType, valueContainsNull = false) - /** - * Enumerates every leaf path in `schema`, in schema order, as its sequence of name parts. - * Structs unfold recursively; every other type (including arrays and maps) is an opaque leaf. - */ - private[autocdc] def extractLeafPaths(schema: StructType): Seq[Seq[String]] = - schema.fields.toSeq.flatMap { field => - field.dataType match { - case nested: StructType => - extractLeafPaths(nested).map(field.name +: _) - case _ => Seq(Seq(field.name)) - } - } - /** Encodes a leaf path as the compact JSON string persisted as its version map key. */ private[autocdc] def encodePath(path: Seq[String]): String = compact(JArray(path.map(JString(_)).toList)) @@ -136,13 +123,13 @@ private[pipelines] object Scd2VersionMap { columnSelection = Some(ignoreNullSelection), resolver = resolver ) - val ignoreNullLeafPaths = extractLeafPaths(ignoreNullColumns).toSet + val ignoreNullLeafPaths = AutoCdcSchemaUtils.extractLeafPaths(ignoreNullColumns).toSet // For each leaf, build a nullable struct (key, value). The struct is non-null only when // the leaf column's runtime value is null (meaning the leaf needs a version map entry). // The value is a non-nullable BooleanType literal indicating authorship: true if the null // is authored, false if declined. - val candidateEntries = extractLeafPaths(schema).map { path => + val candidateEntries = AutoCdcSchemaUtils.extractLeafPaths(schema).map { path => val encodedPath = encodePath(path) val isIgnoreNullLeaf = ignoreNullLeafPaths.contains(path) val leafIsNull = F.col(QuotingUtils.quoteNameParts(path)).isNull diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtilsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtilsSuite.scala new file mode 100644 index 0000000000000..d5ce59d60056c --- /dev/null +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtilsSuite.scala @@ -0,0 +1,69 @@ +/* + * 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.autocdc + +import org.apache.spark.SparkFunSuite +import org.apache.spark.sql.types._ + +class AutoCdcSchemaUtilsSuite extends SparkFunSuite { + + test("extractLeafPaths returns single-element paths for flat columns") { + val schema = new StructType() + .add("a", IntegerType) + .add("b", StringType) + .add("c", DoubleType) + + assert(AutoCdcSchemaUtils.extractLeafPaths(schema) === + Seq(Seq("a"), Seq("b"), Seq("c"))) + } + + test("extractLeafPaths returns leaves rather than intermediate structs") { + val schema = new StructType() + .add("x", IntegerType) + .add("address", new StructType() + .add("city", StringType) + .add("zip", IntegerType)) + + assert(AutoCdcSchemaUtils.extractLeafPaths(schema) === + Seq(Seq("x"), Seq("address", "city"), Seq("address", "zip"))) + } + + test("extractLeafPaths returns full paths for deeply nested structs") { + val schema = new StructType() + .add("top", new StructType() + .add("mid", new StructType() + .add("leaf", StringType))) + + assert(AutoCdcSchemaUtils.extractLeafPaths(schema) === + Seq(Seq("top", "mid", "leaf"))) + } + + test("extractLeafPaths treats arrays and maps as opaque leaves") { + val schema = new StructType() + .add("tags", ArrayType(StringType)) + .add("props", MapType(StringType, IntegerType)) + .add("plain", IntegerType) + + assert(AutoCdcSchemaUtils.extractLeafPaths(schema) === + Seq(Seq("tags"), Seq("props"), Seq("plain"))) + } + + test("extractLeafPaths returns an empty sequence for an empty schema") { + assert(AutoCdcSchemaUtils.extractLeafPaths(new StructType()) === Seq.empty) + } +} diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala index 660f184874d2d..fa4a606e2fa21 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala @@ -77,34 +77,6 @@ class Scd2VersionMapSuite extends QueryTest with SharedSparkSession { private def encodedPath(path: String*): String = Scd2VersionMap.encodePath(path) - // ========================================================================= - // extractLeafPaths - // ========================================================================= - - test("extractLeafPaths - flat columns produce single-element paths") { - assert(Scd2VersionMap.extractLeafPaths(flatSchema) === - Seq(Seq("a"), Seq("b"), Seq("c"))) - } - - test("extractLeafPaths - nested struct produces only leaf paths, not intermediaries") { - assert(Scd2VersionMap.extractLeafPaths(nestedSchema) === - Seq(Seq("x"), Seq("address", "city"), Seq("address", "zip"))) - } - - test("extractLeafPaths - deeply nested struct produces full multi-part paths") { - assert(Scd2VersionMap.extractLeafPaths(deeplyNestedSchema) === - Seq(Seq("top", "mid", "leaf"))) - } - - test("extractLeafPaths - arrays and maps are opaque leaves") { - assert(Scd2VersionMap.extractLeafPaths(arrayAndMapSchema) === - Seq(Seq("tags"), Seq("props"), Seq("plain"))) - } - - test("extractLeafPaths - empty schema produces empty seq") { - assert(Scd2VersionMap.extractLeafPaths(new StructType()) === Seq.empty) - } - // ========================================================================= // encodePath // ========================================================================= From 7e61ec0d39c27988357e00abb18f43ab2b05aedc Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 9 Sep 2026 04:59:44 +0000 Subject: [PATCH 14/15] [SPARK-59183][SDP] Handle reductive evolution in version maps --- .../autocdc/Scd2BatchProcessor.scala | 20 ++-- .../pipelines/autocdc/Scd2VersionMap.scala | 5 +- .../autocdc/Scd2BatchProcessorSuite.scala | 110 ++++++++++++++++++ 3 files changed, 122 insertions(+), 13 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index 263b6040d08ac..b8e82343c3297 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala @@ -214,21 +214,22 @@ case class Scd2BatchProcessor( * map: their data-column values are not part of the SCD2 contract, so authorship tracking * is not applicable. * - * Must run after [[projectTargetColumnsOntoMicrobatch]], because the eligible schema is - * computed from the post-selection schema. + * Must run after [[projectTargetColumnsOntoMicrobatch]] and + * [[alignMicrobatchToTargetSchema]], because the eligible schema is computed from the selected, + * target-aligned schema. * * TODO(SPARK-59343): decide how to handle the ignore-null selection changing between * partial-retry attempts of the same microbatch. */ - private def extendMicrobatchRowsWithVersionMap(projectedDf: DataFrame): DataFrame = + private def extendMicrobatchRowsWithVersionMap(alignedDf: DataFrame): DataFrame = changeArgs.ignoreNullSelection match { - case None => projectedDf + case None => alignedDf case Some(ignoreNullSelection) => val cdcMetadataCol = F.col(AutoCdcReservedNames.cdcMetadataColName) - val resolver = projectedDf.sparkSession.sessionState.conf.resolver + val resolver = alignedDf.sparkSession.sessionState.conf.resolver val schemaEligibleForNullAuthorshipTracking = Scd2BatchProcessor.computeUserDataSchema( - schema = projectedDf.schema, + schema = alignedDf.schema, changeArgs = changeArgs, resolver = resolver ) @@ -244,7 +245,7 @@ case class Scd2BatchProcessor( resolver = resolver )) - projectedDf.withColumn( + alignedDf.withColumn( colName = AutoCdcReservedNames.cdcMetadataColName, col = cdcMetadataCol .withField(Scd2BatchProcessor.versionMapFieldName, versionMap) @@ -1526,9 +1527,8 @@ object Scd2BatchProcessor { * against this, so an exclude-list in either cannot pick up a key or a framework column, and * an include-list naming one fails as not found. * - * `schema` is expected to have already been narrowed by [[ChangeArgs.columnSelection]], which - * happens once per microbatch in [[Scd2BatchProcessor.projectTargetColumnsOntoMicrobatch]]; - * this method does not re-apply it. + * `schema` is expected to have already been narrowed by [[ChangeArgs.columnSelection]] and then + * aligned to the persisted target schema. This method does not re-apply the selection. */ private[pipelines] def computeUserDataSchema( schema: StructType, diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala index 9e1b7c3fac5a8..d06a85ac7b14f 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala @@ -86,9 +86,8 @@ private[pipelines] object Scd2VersionMap { * Keys are compact JSON arrays of the name parts of *leaf* columns that received a null * value in their upsert event (e.g. `["address","city"]`). Keeping * name parts separate distinguishes a nested path from a column whose name contains dots - * and keeps persisted keys independent of SQL identifier quoting rules. Note: until - * SPARK-59347, these are microbatch-schema spellings, not necessarily canonical target - * spellings. + * and keeps persisted keys independent of SQL identifier quoting rules. Name parts use the + * persisted target schema's canonical spelling. * * Values indicate authorship: `true` means authored-null, `false` means unauthored-null. * Null values never appear in the map. diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala index 4f493dc9fc123..f57d0102501d0 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala @@ -933,6 +933,116 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) } + gridTest("preprocessMicrobatch applies ignore-null selection to reductively removed columns")( + Seq( + // Include only value: removed is outside ignore-null and its padded null is authored. + (ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("value"))), true), + // Exclude value: removed is inside ignore-null and its padded null is declined. + (ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("value"))), false) + ) + ) { case (ignoreNullSelection, expectedAuthorship) => + val batchSchema = new StructType() + .add("id", IntegerType) + .add("value", StringType) + .add("seq", LongType) + val targetUserSchema = new StructType() + .add("id", IntegerType) + .add("value", StringType) + .add("removed", StringType) + val batch = microbatchOf(batchSchema)(Row(1, "a", 10L)) + val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + columnSelection = Some(ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq")))), + ignoreNullSelection = Some(ignoreNullSelection) + ), + resolvedSequencingType = LongType + ) + + val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema)) + checkAnswer( + df = result.select( + F.col("removed"), + Scd2BatchProcessor.versionMapOf( + F.col(AutoCdcReservedNames.cdcMetadataColName)).as("vm")), + expectedAnswer = Row( + null, + Map(Scd2VersionMap.encodePath(Seq("removed")) -> expectedAuthorship)) + ) + } + + test("preprocessMicrobatch applies ignore-null to a reductively removed nested field") { + val batchSchema = new StructType() + .add("id", IntegerType) + .add("value", new StructType().add("a", IntegerType)) + .add("seq", LongType) + val targetUserSchema = new StructType() + .add("id", IntegerType) + .add("value", new StructType() + .add("a", IntegerType) + .add("removed", StringType)) + val batch = microbatchOf(batchSchema)(Row(1, Row(1), 10L)) + val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + columnSelection = Some(ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq")))), + ignoreNullSelection = + Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("value")))) + ), + resolvedSequencingType = LongType + ) + + val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema)) + checkAnswer( + df = result.select( + F.col("value"), + Scd2BatchProcessor.versionMapOf( + F.col(AutoCdcReservedNames.cdcMetadataColName)).as("vm")), + expectedAnswer = Row( + Row(1, null), + Map(Scd2VersionMap.encodePath(Seq("value", "removed")) -> false)) + ) + } + + test("preprocessMicrobatch uses target spelling for version-map keys") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + val batchSchema = new StructType() + .add("id", IntegerType) + .add("Value", StringType) + .add("seq", LongType) + val targetUserSchema = new StructType() + .add("id", IntegerType) + .add("value", StringType) + val batch = microbatchOf(batchSchema)(Row(1, null, 10L)) + val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + columnSelection = Some(ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq")))), + ignoreNullSelection = + Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("Value")))) + ), + resolvedSequencingType = LongType + ) + + val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema)) + assert(result.schema.fieldNames.take(2).toSeq == Seq("id", "value")) + checkAnswer( + df = result.select(Scd2BatchProcessor.versionMapOf( + F.col(AutoCdcReservedNames.cdcMetadataColName)).as("vm")), + expectedAnswer = Row(Map(Scd2VersionMap.encodePath(Seq("value")) -> false)) + ) + } + } + test("preprocessMicrobatch leaves version map null for all rows when ignore null is off") { val schema = new StructType() .add("id", IntegerType) From eb76dc43e7fc3ecf2343df782b6213b05a999e3e Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 9 Sep 2026 17:56:36 +0000 Subject: [PATCH 15/15] [SPARK-59183][SDP] Pin version map key casing --- .../pipelines/autocdc/Scd2BatchProcessorSuite.scala | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala index f57d0102501d0..9fd1b1dff406d 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala @@ -1010,7 +1010,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) } - test("preprocessMicrobatch uses target spelling for version-map keys") { + test("version-map key spelling matches the preprocessed microbatch and target") { withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { val batchSchema = new StructType() .add("id", IntegerType) @@ -1034,11 +1034,18 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema)) - assert(result.schema.fieldNames.take(2).toSeq == Seq("id", "value")) + val sourceValueName = batchSchema.fields(1).name + val preprocessedValueName = result.schema.fields(1).name + val targetValueName = targetUserSchema.fields(1).name + assert(sourceValueName == "Value") + assert(preprocessedValueName == targetValueName) + assert(preprocessedValueName == "value") + + val expectedVersionMapKey = Scd2VersionMap.encodePath(Seq(preprocessedValueName)) checkAnswer( df = result.select(Scd2BatchProcessor.versionMapOf( F.col(AutoCdcReservedNames.cdcMetadataColName)).as("vm")), - expectedAnswer = Row(Map(Scd2VersionMap.encodePath(Seq("value")) -> false)) + expectedAnswer = Row(Map(expectedVersionMapKey -> false)) ) } }