[SPARK-59347][SDP] SCD2 Ignore-null support; union microbatch with target table during preprocessing

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala

@@ -123,32 +123,34 @@ 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
    * downstream reconciliation - not preprocessing.
    *
    * @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.

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)