[SPARK-59183][SDP] SCD2 Ignore-null support; Populate version map per event row at ingest

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))
+      }
+    }
+}

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

@@ -123,32 +123,35 @@ 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 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
    * 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))
+      .transform(extendMicrobatchRowsWithVersionMap)
   }
 
   /**
@@ -187,21 +190,68 @@ 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.
+   *
+   * 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]] 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(alignedDf: DataFrame): DataFrame =
+    changeArgs.ignoreNullSelection match {
+      case None => alignedDf
+      case Some(ignoreNullSelection) =>
+        val cdcMetadataCol = F.col(AutoCdcReservedNames.cdcMetadataColName)
+        val resolver = alignedDf.sparkSession.sessionState.conf.resolver
+        val schemaEligibleForNullAuthorshipTracking =
+          Scd2BatchProcessor.computeUserDataSchema(
+            schema = alignedDf.schema,
+            changeArgs = changeArgs,
+            resolver = resolver
+          )
+
+        // Only upsert rows get a populated version map. By convention, delete-encoded
+        // 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,
+          resolver = resolver
+        ))
+
+        alignedDf.withColumn(
+          colName = AutoCdcReservedNames.cdcMetadataColName,
+          col = cdcMetadataCol
+            .withField(Scd2BatchProcessor.versionMapFieldName, versionMap)
+        )
+    }
+
   /**
    * 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
@@ -246,6 +296,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.
@@ -1445,23 +1508,37 @@ 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]] and then
+   * aligned to the persisted target schema. This method does not re-apply the selection.
+   */
+  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))
+    })
   }
 
   /**

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)

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

@@ -17,7 +17,13 @@
 
 package org.apache.spark.sql.pipelines.autocdc
 
-import org.apache.spark.sql.types.{BooleanType, MapType, StringType}
+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
+import org.apache.spark.sql.types.{BooleanType, MapType, StringType, StructType}
 
 /**
  * Per-row column authorship tracker for SCD2 ignore-null semantics.
@@ -77,15 +83,69 @@ 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 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. Name parts use the
+   * persisted target schema's canonical spelling.
    *
-   * 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)
+
+  /** 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
+   * 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 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 = AutoCdcSchemaUtils.extractLeafPaths(schema).map { path =>
+      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(encodedPath).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)
+    }
+  }
 }