[SPARK-59208][SDP] SCD2 Ignore-null support; Introduce ignore-null selection in ChangeArgs

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

@@ -239,6 +239,30 @@
     ],
     "sqlState" : "22023"
   },
+  "AUTOCDC_IGNORE_NULL_CANNOT_SELECT_RESERVED_COLUMN" : {
+    "message" : [
+      "In AutoCDC flow <flowName>, the ignore-null selection names column `<columnName>` which collides with the reserved column name prefix `<reservedColumnNamePrefix>` (<caseSensitivity>). Reserved columns may not appear in an ignore-null selection."
+    ],
+    "sqlState" : "42710"
+  },
+  "AUTOCDC_IGNORE_NULL_COLUMN_NOT_IN_OUTPUT_COLUMNS" : {
+    "message" : [
+      "In AutoCDC flow <flowName>, the ignore-null selection names column `<columnName>` which is not present in the flow's selected output columns (<caseSensitivity>). Only user-data columns surviving the column selection may appear in an ignore-null selection."
+    ],
+    "sqlState" : "42703"
+  },
+  "AUTOCDC_IGNORE_NULL_EMPTY_COLUMN_LIST" : {
+    "message" : [
+      "AutoCDC flow specifies an empty `ignore_null_updates_column_list`. Provide at least one column, or omit the option entirely to disable ignore-null updates. To ignore nulls for every column, specify an empty `ignore_null_updates_except_column_list` instead."
+    ],
+    "sqlState" : "22023"
+  },
+  "AUTOCDC_IGNORE_NULL_SELECTION_CONTAINS_KEY_COLUMN" : {
+    "message" : [
+      "In AutoCDC flow <flowName>, the ignore-null selection names key column `<columnName>` (<caseSensitivity>). Key columns identify rows and may not appear in an ignore-null selection. Key columns: <keyColumnNames>."
+    ],
+    "sqlState" : "22023"
+  },
   "AUTOCDC_INVALID_STATE" : {
     "message" : [
       "Detected an invalid AutoCDC state for target table <tableName>:"
@@ -6742,6 +6766,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 +6794,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/ChangeArgs.scala

@@ -64,6 +64,15 @@ object ColumnSelection {
   case class ExcludeColumns(columns: Seq[UnqualifiedColumnName])
       extends ColumnSelection
 
+  /** Extracts the explicitly named columns, or `Nil` when the selection is `None`. */
+  def namedColumns(
+      selection: Option[ColumnSelection]): Seq[UnqualifiedColumnName] =
+    selection match {
+      case None => Nil
+      case Some(IncludeColumns(cols)) => cols
+      case Some(ExcludeColumns(cols)) => cols
+    }
+
   /**
    * Applies [[ColumnSelection]] to a [[StructType]] and returns the filtered schema. Field order
    * follows the original schema; only matching fields are retained in the returned schema.
@@ -186,17 +195,26 @@ object ScdType {
  *                               which has no run concept and therefore no history-tracking columns.
  *                               See the "run of upsert events" concept in the `Scd2BatchProcessor`
  *                               scaladoc for the precise definition of a run.
+ * @param ignoreNullSelection    Selects the columns whose nulls are treated as declined
+ *                               authorship rather than authored nulls. None means ignore-null
+ *                               is off completely. An empty include list is rejected; an empty
+ *                               exclude list selects every eligible column. Eligible columns
+ *                               are those surviving the column selection that are neither keys
+ *                               nor columns whose names start with the reserved AutoCDC prefix.
+ *                               Naming a struct selects every leaf beneath it.
  */
 case class ChangeArgs(
     keys: Seq[UnqualifiedColumnName],
     sequencing: Column,
     storedAsScdType: ScdType,
     deleteCondition: Option[Column] = None,
     columnSelection: Option[ColumnSelection] = None,
-    trackHistorySelection: Option[ColumnSelection] = None
+    trackHistorySelection: Option[ColumnSelection] = None,
+    ignoreNullSelection: Option[ColumnSelection] = None
 ) {
   ChangeArgs.validateNonEmptyKeys(keys)
   ChangeArgs.validateTrackHistoryOnlyForScd2(storedAsScdType, trackHistorySelection)
+  ChangeArgs.validateNonEmptyIgnoreNullIncludeList(ignoreNullSelection)
 }
 
 object ChangeArgs {
@@ -231,4 +249,20 @@ object ChangeArgs {
       )
     }
   }
+
+  /**
+   * Rejects an empty ignore-null include list; "ignore-null off" is spelled
+   * as None.
+   */
+  private def validateNonEmptyIgnoreNullIncludeList(
+      ignoreNullSelection: Option[ColumnSelection]): Unit = {
+    ignoreNullSelection match {
+      case Some(ColumnSelection.IncludeColumns(columns)) if columns.isEmpty =>
+        throw new AnalysisException(
+          errorClass = "AUTOCDC_IGNORE_NULL_EMPTY_COLUMN_LIST",
+          messageParameters = Map.empty
+        )
+      case _ => ()
+    }
+  }
 }

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()