From bd07e659de1974ac44d1f82cf4a28fde91e6398b Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 9 Sep 2026 04:55:30 +0000 Subject: [PATCH 1/4] [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 2/4] [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 3/4] 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 4/4] 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