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..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 @@ -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,23 @@ case class Scd2BatchProcessor( * * @param microbatchDf * the incoming CDC microbatch. + * @param targetTableDf + * the current persisted target table. * @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 +248,19 @@ 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 exact + * case-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..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 @@ -68,15 +68,18 @@ case class Scd2ForeachBatchHandler( batchId = batchId ).validateMicrobatch() - val preprocessedBatchDf = batchProcessor.preprocessMicrobatch(batchDf) + val targetTableDf = batchDf.sparkSession.read.table(targetTableIdentifier.quotedString) + val auxTableDf = batchDf.sparkSession.read.table(auxiliaryTableIdentifier.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, targetTableDf = targetTableDf, @@ -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..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 @@ -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,110 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) } + gridTest("preprocessMicrobatch keeps divergent 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 between target " + + "and microbatch") { + 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", 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 keeps nested target field 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("city", 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 + )) + 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)) + ) + } + } + // =============== computeMinimumSequencePerKey tests =============== test("computeMinimumSequencePerKey returns one row per distinct key and aggregates across " + @@ -774,7 +900,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 +934,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 +958,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 +979,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 +2301,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 +2335,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") }