[SPARK-59165][SDP] SCD2 Ignore-null support; Introduce version map in _cdc_metadata

common/utils/src/main/resources/error/error-conditions.json

@@ -6742,6 +6742,14 @@
     ],
     "sqlState" : "0A000"
   },
+  "PIPELINE_NESTED_COMPLEX_TYPE_SCHEMA_EVOLUTION_UNSUPPORTED" : {
+    "message" : [
+      "Schema evolution within a complex type (array or map) whose element is itself a complex type is not supported.",
+      "Column path <columnPath> has element type <currentType> but the target element type is <targetType>.",
+      "Flatten the nesting or apply the change manually."
+    ],
+    "sqlState" : "0A000"
+  },
   "PIPELINE_RUN_FAILED" : {
     "message" : [
       "<message>"
@@ -6762,6 +6770,12 @@
     ],
     "sqlState" : "42K03"
   },
+  "PIPELINE_TIGHTEN_NULLABILITY_UNSUPPORTED" : {
+    "message" : [
+      "Cannot tighten nullability of <columnPath> from nullable to non-nullable. Existing data may already contain nulls."
+    ],
+    "sqlState" : "0A000"
+  },
   "PIPE_OPERATOR_AGGREGATE_EXPRESSION_CONTAINS_NO_AGGREGATE_FUNCTION" : {
     "message" : [
       "Non-grouping expression <expr> is provided as an argument to the |> AGGREGATE pipe operator but does not contain any aggregate function; please update it to include an aggregate function and then retry the query again."

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

@@ -194,6 +194,9 @@ case class Scd2BatchProcessor(
       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
       )
     )
@@ -514,6 +517,9 @@ case class Scd2BatchProcessor(
             colName,
             Scd2BatchProcessor.constructCdcMetadataCol(
               recordStartAt = F.lit(null).cast(resolvedSequencingType),
+              // Decomposition tails are synthetic rows that carry no column-level authorship; they
+              // are instead row wide delete markers. They should always hold a null version map.
+              versionMap = F.lit(null),
               sequencingType = resolvedSequencingType
             )
           )
@@ -922,6 +928,9 @@ case class Scd2BatchProcessor(
           isDecompositionTail,
           Scd2BatchProcessor.constructCdcMetadataCol(
             recordStartAt = endAt,
+            // Tombstone rows carry no column-level authorship; they are instead row wide delete
+            // markers. They should always hold a null version map.
+            versionMap = F.lit(null),
             sequencingType = resolvedSequencingType
           )
         ).otherwise(F.col(c)).as(c, metadata)
@@ -1357,6 +1366,9 @@ object Scd2BatchProcessor {
    */
   private[pipelines] val recordStartAtFieldName: String = "__RECORD_START_AT"
 
+  /** CDC metadata field for the ignore-null version map. */
+  private[pipelines] val versionMapFieldName: String = "__VERSION_MAP"
+
   /**
    * Aux-table only column that holds the microbatch id by which a row was logically
    * deleted (null if the row is still live). Future microbatches must treat any row with a
@@ -1567,6 +1579,10 @@ object Scd2BatchProcessor {
   private def recordStartAtOf(cdcMetadataCol: Column): Column =
     cdcMetadataCol.getField(recordStartAtFieldName)
 
+  /** Project the [[versionMapFieldName]] out of an SCD2 CDC metadata column. */
+  private[autocdc] def versionMapOf(cdcMetadataCol: Column): Column =
+    cdcMetadataCol.getField(versionMapFieldName)
+
   /**
    * The [[Scd2IntervalColumns]] of a row read from either the auxiliary or the target table, in
    * the canonical SCD2 row schema. The columns are unresolved name references, so they read from
@@ -1587,7 +1603,16 @@ object Scd2BatchProcessor {
         // The sequence value of the originating CDC event for this row. Nullable because
         // decomposition tails, which are temporarily and synthetically constructed during
         // reconciliation, have a null record start at.
-        StructField(recordStartAtFieldName, sequencingType, nullable = true)
+        StructField(recordStartAtFieldName, sequencingType, nullable = true),
+        // The version map representing null-authorship for the row. If the version map is null for
+        // a row, that row was ingested with ignore-null off, and all columns are considered
+        // explicitly authored (null or not). If the version map is non-null, the row was ingested
+        // with ignore-null on, and contents of the map comply with the contract defined in
+        // [[Scd2VersionMap]].
+        //
+        // Tombstones and decomposition tails also always hold null version maps because column
+        // authorship is not applicable - they are delete markers.
+        StructField(versionMapFieldName, Scd2VersionMap.mapType, nullable = true)
       )
     )
 
@@ -1597,11 +1622,13 @@ object Scd2BatchProcessor {
    */
   private[pipelines] def constructCdcMetadataCol(
       recordStartAt: Column,
+      versionMap: Column,
       sequencingType: DataType
   ): Column = {
     val cdcMetadataFieldsInOrder = cdcMetadataColSchema(sequencingType).fields.map { field =>
       val value = field.name match {
         case `recordStartAtFieldName` => recordStartAt
+        case `versionMapFieldName` => versionMap
         case other =>
           throw SparkException.internalError(
             s"Unable to construct SCD2 CDC metadata column due to unknown `${other}` field."

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

@@ -0,0 +1,91 @@
+/*
+ * 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.{BooleanType, MapType, StringType}
+
+/**
+ * Per-row column authorship tracker for SCD2 ignore-null semantics.
+ *
+ * Recall in SCD2, every materialized row traces back to an upsert event that created it (and
+ * if the row is closed then also a delete/succession event that closed it, but that's not
+ * relevant here). Every data column in the row is at least partially derived by the
+ * corresponding data column in the upsert event that spawned the row.
+ *
+ * For columns where ignore-null was not applied, the data column in the row is fully derived
+ * (authored) by the corresponding data column in the upsert event. For columns where
+ * ignore-null was applied however, if the data column in the upsert event was null
+ * (unauthored), then we need to look backwards to deduce the corresponding inherited data
+ * column for the row.
+ *
+ * Non-null values in an event are always considered authored, regardless of whether the column
+ * in the event was included in the ignore-null configuration or not. Null values however, as
+ * mentioned above, may or may not be considered authored -- it depends on whether they are
+ * specified for a column that was included in the ignore-null configuration.
+ *
+ * In SCD2 the version map helps us answer per row: for all the columns that received a null
+ * value in the upsert event that created this row, which nulls are considered authored vs
+ * unauthored?
+ *
+ * Concretely, the contract of the version map is as follows.
+ * 1. Every column that received a null in the event but is considered authored (i.e. not part
+ *    of ignore-null selection at ingestion), receives an entry of (column name, true) in the
+ *    version map.
+ * 2. Every column that received a null in the event but is considered unauthored (i.e. part
+ *    of the ignore-null selection at ingestion), receives an entry of (column name, false) in
+ *    the version map.
+ * 3. Every column that was not present in the event, but schema evolved in later with a null
+ *    value, will be treated as unauthored BUT does not yet have any entry in the version map.
+ *    An entry will be added as per (2).
+ *
+ * In a single sentence: if a null column in the SCD2 row is either absent from the version
+ * map or has a false value in the version map, the null is considered unauthored by the
+ * upsert event that spawned this row. Otherwise the null value was explicitly authored by
+ * the row.
+ *
+ * As mentioned above, authorship is dependent on the configured ignore-null selection, which
+ * is free to change between pipeline runs for the same AutoCDC flow. As such, we choose that
+ * the version map strictly reflects authorship as of the ignore-null selection that was active
+ * when the upsert event that produced this row was ingested. This means the authorship
+ * information the version map encoded at creation time is invariant/frozen -- even if the
+ * ignore-null selection changes on a future run, the version map is not rewritten (unless the
+ * table is full refreshed).
+ *
+ * It's worth noting that while contract case (3) materializes new entries in the version map
+ * after creation, it does not change the set of columns whose null values are considered
+ * authored/unauthored. Therefore authorship information encoded by the mutated version map is
+ * still invariant, and independent of a changing ignore-null configuration. New rows materialized
+ * in the map are still compliant with whatever the ignore-null selection was at ingestion time.
+ */
+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.
+   *
+   * Values indicate authorship. I.e, `true` => authored-null, `false` => unauthored-null.
+   *
+   * 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)
+}

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala

@@ -690,7 +690,11 @@ object DatasetManager extends Logging {
     // above has already folded a case-only-differing incoming field onto the persisted one. On the
     // non-merging paths (materialized views, full refresh), `targetSchema` is the declared schema
     // as-is, where exact-name matching keeps a case-only rename visible as a schema change.
-    val columnChanges = diffSchemas(currentSchema, targetSchema)
+    val columnChanges = diffSchemas(
+      currentSchema,
+      targetSchema,
+      rejectNullabilityTightening = mergeWithExistingSchema
+    )
 
     val existingProperties = existingTable.properties()
 

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala

@@ -402,6 +402,7 @@ class AutoCdcMergeFlow(
           F.lit(null).cast(sequencingType).as(Scd2BatchProcessor.endAtColName)
         val emptyCdcMetadataCol: Column = Scd2BatchProcessor.constructCdcMetadataCol(
           recordStartAt = F.lit(null),
+          versionMap = F.lit(null),
           sequencingType = sequencingType
         ).as(AutoCdcReservedNames.cdcMetadataColName)