You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by jt...@apache.org on 2026/10/15 22:03:03 UTC
(spark) branch branch-4.x updated: [SPARK-59358][SDP] SCD2 Ignore-null support; Coalesce ignored-null values
This is an automated email from the ASF dual-hosted git repository.
jose-torres pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new dd209fd8fc1b [SPARK-59358][SDP] SCD2 Ignore-null support; Coalesce ignored-null values
dd209fd8fc1b is described below
commit dd209fd8fc1bf3784e3039ec15188bbcd3e2500c
Author: AnishMahto <an...@gmail.com>
AuthorDate: Thu Oct 15 15:02:42 2026 -0700
[SPARK-59358][SDP] SCD2 Ignore-null support; Coalesce ignored-null values
### What changes were proposed in this pull request?
Once the version map is constructed per incoming row in the microbatch, we need to start actually reconciling columns in the incoming microbatch that have been deduced as unauthored - they should instead inherit the last known authored value.
While reconciling the affected existing rows from the aux/target table, it's also possible out of order received events in the current microbatch affect what authored value existing ignored-null columns inherit; these need to be updated too.
The core algorithm iterates over the full set of affected rows by the incoming microbatch in window order, coalescing unauthored columns values with the last known authored value within the window. This is intentionally run before start/end reconciliation, so that rows' column values are fully deduced before applying the track history selection and determining no-op upsert runs.
It's worth noting that only columns explicitly in the ignore-null selection in the current reconciliation are coalesced, even if existing rows have initialized version maps from a different ignore-null selection in a previous reconciliation. Version maps themselves are semantically immutable (schema evolution can lazily materialize version map entries, but that does not change the per-column authorship that the version map previously encoded), and the chosen semantic is:
1. Version maps represent per-column authorship. They are the source of truth for what SDP considered authored in the event, per upsert event. This is deduced using whatever the ignore-null selection was when the event was ingested, OR using the current selection if the event was previously ingested when ignore-null was off.
3. The current ignore-null selection represents the full set of columns whose unauthored values will be coalesced during this microbatch's reconciliation, for all rows pulled into reconciliation.
In the event that the ignore-null selection changes between runs, coalescing of all existing rows against the new ignore-null selection is eventually consistent. This is an intentionally selected semantic for performance reasons, to prevent rewriting the entire SCD2 history on an incoming microbatch. Instead, only the affected set of rows pulled in to participate in reconciliation are coalesced.
### Why are the changes needed?
Core algorithm for coalescing/reconciling ignore-null columns for rows in or affected by the incoming microbatch.
### Does this PR introduce _any_ user-facing change?
No. Ignore null reconciliation is strictly only invoked if an ignore-null selection is specified. The ignore-null API is not yet released, so all existing AutoCDC workloads necessarily do not specify an ignore-null selection.
### How was this patch tested?
`Scd2CoalesceIgnoredNullsSuite`.
### Was this patch authored or co-authored using generative AI tooling?
GPT Sol 5.6.
Closes #58947 from AnishMahto/SPARK-59358-coalesce-ignored-null-columns.
Authored-by: AnishMahto <an...@gmail.com>
Signed-off-by: Jose Torres <jt...@apache.org>
(cherry picked from commit 42dc01001f54ed355ad267a58a1eef40b792ffd6)
Signed-off-by: Jose Torres <jt...@apache.org>
---
.../autocdc/IgnoreNullColumnReconstruction.scala | 118 ++++
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 369 ++++++++++-
.../autocdc/Scd2ForeachBatchHandler.scala | 8 +-
.../sql/pipelines/autocdc/Scd2VersionMap.scala | 4 +-
.../IgnoreNullColumnReconstructionSuite.scala | 178 ++++++
.../autocdc/Scd2CoalesceIgnoredNullsSuite.scala | 695 +++++++++++++++++++++
.../autocdc/Scd2ForeachBatchHandlerSuite.scala | 102 ++-
7 files changed, 1459 insertions(+), 15 deletions(-)
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/IgnoreNullColumnReconstruction.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/IgnoreNullColumnReconstruction.scala
new file mode 100644
index 000000000000..a802d912addb
--- /dev/null
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/IgnoreNullColumnReconstruction.scala
@@ -0,0 +1,118 @@
+/*
+ * 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.SparkException
+import org.apache.spark.sql.{functions => F, Column}
+import org.apache.spark.sql.catalyst.expressions.KnownNullable
+import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
+import org.apache.spark.sql.catalyst.util.QuotingUtils
+import org.apache.spark.sql.classic.ExpressionUtils
+import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.util.ArrayImplicits._
+
+/** Rebuilds AutoCDC user-data columns selected for ignore-null from the values of their leaves. */
+private[autocdc] object IgnoreNullColumnReconstruction {
+
+ /**
+ * Rebuilds the column at `path` as `field` from the values of its leaves.
+ *
+ * If `field` is a struct, it is rebuilt from the values of the leaves beneath each of its
+ * fields. Otherwise, `field` is itself a leaf and takes its own value.
+ *
+ * A struct whose leaves are all null is rebuilt as a struct of nulls, which ignore-null treats
+ * as equivalent to a null struct. A struct without leaves is rebuilt as an empty struct. A
+ * nullable struct with a non-nullable leaf reachable through only non-nullable structs cannot
+ * be rebuilt with that leaf null, so it is instead null unless a leaf beneath it is non-null.
+ *
+ * @param path The name parts of `field`, from its top-level column down. Each part is compared
+ * exactly to a field name, so it is not quoted.
+ * @param field The schema to rebuild the column as: its name, type, nullability, and metadata.
+ * @param leafValues The path, in the same form as `path`, and value of each leaf at or beneath
+ * `path`. A leaf of `field` without a value raises an internal error.
+ * @return The rebuilt value, cast to the type of `field` and aliased to its name and metadata.
+ * Evaluating it raises `NOT_NULL_ASSERT_VIOLATION` if `field` is non-nullable and its
+ * value is null.
+ */
+ def reconstructColumn(
+ path: Seq[String],
+ field: StructField,
+ leafValues: Seq[(Seq[String], Column)]): Column = {
+ val reconstructed = field.dataType match {
+ case struct: StructType =>
+ val leafValuesByChildName = leafValues.groupBy { case (leafPath, _) =>
+ leafPath(path.length)
+ }
+ val rebuilt = F.struct(struct.fields.toImmutableArraySeq.map { childField =>
+ reconstructColumn(
+ path = path :+ childField.name,
+ field = childField,
+ leafValues = leafValuesByChildName.getOrElse(childField.name, Seq.empty))
+ }: _*)
+
+ if (field.nullable && cannotRebuildAsStructOfNulls(struct)) {
+ // The schema forbids a struct of nulls here, so the equivalent null struct is used when
+ // every leaf is null. If any leaf is non-null, the struct is rebuilt and a null
+ // non-nullable leaf fails its assertion rather than the struct dropping non-null leaves.
+ val everyLeafIsNull = leafValues.map { case (_, value) => value.isNull }.reduce(_ && _)
+ F.when(everyLeafIsNull, F.lit(null).cast(field.dataType)).otherwise(rebuilt)
+ } else {
+ rebuilt
+ }
+ case _ =>
+ leafValues.collectFirst { case (`path`, value) => value }.getOrElse {
+ throw SparkException.internalError(
+ s"Cannot reconstruct ${QuotingUtils.quoteNameParts(path)} because it has no value.")
+ }
+ }
+ conformToField(reconstructed, path, field)
+ }
+
+ /**
+ * Gives `value` the schema of `field`: its type, nullability, name, and metadata.
+ *
+ * @param path The name parts of `field`, from its top-level column down, reported if `field`
+ * is non-nullable and `value` is null.
+ * @return `value`, cast to the type of `field` and aliased to its name and metadata. Evaluating
+ * it raises `NOT_NULL_ASSERT_VIOLATION` if `field` is non-nullable and `value` is null.
+ */
+ private def conformToField(value: Column, path: Seq[String], field: StructField): Column = {
+ val withDeclaredNullability =
+ if (field.nullable) {
+ // Spark may report `value`, such as a rebuilt struct, as non-nullable; only
+ // `KnownNullable` can widen that.
+ ExpressionUtils.column(KnownNullable(ExpressionUtils.expression(value)))
+ } else {
+ ExpressionUtils.column(AssertNotNull(ExpressionUtils.expression(value), path))
+ }
+ withDeclaredNullability.cast(field.dataType).as(field.name, field.metadata)
+ }
+
+ /**
+ * Whether `struct` has a non-nullable leaf whose enclosing structs beneath `struct` are all also
+ * non-nullable. If such a leaf has no value, neither it nor any of those enclosing structs can
+ * be null, so `struct` itself cannot be non-null.
+ */
+ private def cannotRebuildAsStructOfNulls(struct: StructType): Boolean =
+ struct.fields.exists { field =>
+ !field.nullable && (field.dataType match {
+ case nested: StructType => cannotRebuildAsStructOfNulls(nested)
+ case _ => true
+ })
+ }
+}
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 0a83b03c57fe..bec882cce398 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
@@ -967,6 +967,169 @@ case class Scd2BatchProcessor(
.drop(Scd2BatchProcessor.isRedundantDeleteEncodingColName)
}
+ /**
+ * Establishes version maps from stored values and the current ignore-null selection for
+ * upsert-representing rows that do not have one. Every other row retains its existing version
+ * map unchanged.
+ */
+ private def initializeMissingVersionMaps(
+ rowsDf: DataFrame,
+ eligibleSchema: StructType,
+ ignoreNullSelection: ColumnSelection,
+ isUpsertRepresentingRow: Column,
+ resolver: Resolver): DataFrame = {
+ val metadataColumnName = AutoCdcReservedNames.cdcMetadataColName
+ val cdcMetadataCol = F.col(metadataColumnName)
+ val existingVersionMap = Scd2BatchProcessor.versionMapOf(cdcMetadataCol)
+ val versionMapToPersist = F.when(
+ isUpsertRepresentingRow && existingVersionMap.isNull,
+ Scd2VersionMap.buildVersionMap(eligibleSchema, ignoreNullSelection, resolver)
+ ).otherwise(existingVersionMap)
+
+ rowsDf.withColumn(
+ metadataColumnName,
+ cdcMetadataCol
+ .withField(Scd2BatchProcessor.versionMapFieldName, versionMapToPersist)
+ .as(metadataColumnName, rowsDf.schema(metadataColumnName).metadata))
+ }
+
+ /**
+ * Reconciles rows in `decomposedDf` against the active ignore-null selection. For each selected
+ * leaf, an unauthored value inherits from the nearest preceding row for the same key that
+ * authored the leaf. Inheritance stops at delete and interval-gap boundaries. When the first row
+ * for a key is an upsert, its stored value is available to later rows even if its map marks the
+ * leaf unauthored, because no preceding row is present.
+ *
+ * The method also materializes version map entries for schema-evolved leaves that would otherwise
+ * lose their unauthored signal after inheriting a non-null value.
+ *
+ * Each row's version map determines whether its stored value is considered authored. Before
+ * coalescing, upsert-representing rows with no map establish one from their stored values and the
+ * active selection; existing maps are preserved.
+ *
+ * Must run after decomposition cleanup so tombstones, decomposition tails, and interval gaps can
+ * terminate inheritance. Must run before start/end reconciliation so inherited tracked-history
+ * values participate in determining the final SCD2 runs.
+ */
+ private[autocdc] def coalesceIgnoredNulls(
+ decomposedDf: DataFrame): DataFrame =
+ changeArgs.ignoreNullSelection match {
+ case None => decomposedDf
+ case Some(ignoreNullSelection) =>
+ val resolver = decomposedDf.sparkSession.sessionState.conf.resolver
+ val userDataColumnSchema = AutoCdcSchemaUtils.excludeColumns(
+ schema = decomposedDf.schema,
+ columnNamesToExclude = changeArgs.keys.map(_.name) ++
+ Scd2BatchProcessor.reservedFrameworkColNames,
+ resolver = resolver
+ )
+ val activeIgnoreNullLeafPaths =
+ Scd2VersionMap.resolveIgnoreNullLeafPaths(
+ userDataColumnSchema, ignoreNullSelection, resolver)
+
+ val cdcMetadataCol = F.col(AutoCdcReservedNames.cdcMetadataColName)
+ val versionMapCol = Scd2BatchProcessor.versionMapOf(cdcMetadataCol)
+
+ val currentRow = Scd2IntervalColumns(
+ recordStartAt = Scd2BatchProcessor.recordStartAtOf(cdcMetadataCol),
+ startAt = F.col(Scd2BatchProcessor.startAtColName),
+ endAt = F.col(Scd2BatchProcessor.endAtColName)
+ )
+ val nextRow = currentRow.leadBy(1, orderChronologicallyPerKeyWindow)
+
+ val isUpsertRepresentingRow = RowClassifier.isUpsertRepresentingRow(currentRow)
+ val withInitializedVersionMaps = initializeMissingVersionMaps(
+ rowsDf = decomposedDf,
+ eligibleSchema = userDataColumnSchema,
+ ignoreNullSelection = ignoreNullSelection,
+ isUpsertRepresentingRow = isUpsertRepresentingRow,
+ resolver = resolver
+ )
+
+ // Strictly preceding rows only, as a row may never inherit from itself.
+ val precedingRowsInKeyWindow =
+ orderChronologicallyPerKeyWindow.rowsBetween(Window.unboundedPreceding, -1)
+
+ // Project row-level context used to compute the leaf-level inheritance
+ // expressions below.
+ val (withRowInheritanceContextDf, rowInheritanceContext) =
+ RowInheritanceContext.projectOn(
+ df = withInitializedVersionMaps,
+ rowEndsInheritanceChain = F.coalesce(
+ // Any row that fully closes or represents a delete event (closing a
+ // preceding upsert event) ends any running inheritance chain.
+ RowClassifier.isTombstone(currentRow) ||
+ RowClassifier.isDecompositionTail(currentRow) ||
+ RowClassifier.rowClosesStrictlyBeforeNextRowIsVisible(
+ currentRow.endAt, nextRow),
+ F.lit(false)
+ ),
+ isUpsertRepresentingRow = isUpsertRepresentingRow,
+ isFirstRowInKeyWindow =
+ F.row_number().over(orderChronologicallyPerKeyWindow) === 1
+ )
+
+ // Only leaves in the active ignore-null selection participate in
+ // reconciliation. The initialized version map determines whether each row
+ // authored its value; the selection determines whether that recorded
+ // authorship is relevant to this reconciliation.
+ val ignoreNullLeafInheritanceContexts =
+ activeIgnoreNullLeafPaths.zipWithIndex.map { case (path, index) =>
+ LeafInheritanceContext(
+ path = path,
+ index = index,
+ versionMap = versionMapCol,
+ rowInheritanceContext = rowInheritanceContext,
+ precedingRowsInKeyWindow = precedingRowsInKeyWindow
+ )
+ }
+ val ignoreNullContextsByTopLevelColumn =
+ ignoreNullLeafInheritanceContexts.groupBy(_.path.head)
+
+ // Evaluate all values available to inherit in one window pass.
+ val withValuesToInheritDf = withRowInheritanceContextDf.withColumns(
+ ignoreNullLeafInheritanceContexts.map { context =>
+ context.valueToInheritIfAnyColName -> context.valueToInheritIfAny
+ }.toMap)
+
+ // Compute updated version maps after materializing new entries due to
+ // schema evolution.
+ val schemaEvolutionUpdatedVersionMap =
+ Scd2BatchProcessor.updateVersionMapWithSchemaEvolution(
+ versionMapCol,
+ ignoreNullLeafInheritanceContexts)
+
+ // Build one expression per original column, replacing the CDC metadata and
+ // selected user-data columns.
+ // Expressions are resolved against columns in [[withValuesToInheritDf]].
+ val outputColumns = decomposedDf.columns.map {
+ case colName if colName == AutoCdcReservedNames.cdcMetadataColName =>
+ // Update the version map after materializing schema-evolution entries.
+ cdcMetadataCol
+ .withField(
+ Scd2BatchProcessor.versionMapFieldName,
+ schemaEvolutionUpdatedVersionMap
+ )
+ .as(colName, decomposedDf.schema(colName).metadata)
+ case colName
+ if ignoreNullContextsByTopLevelColumn.contains(colName) =>
+ // Rebuild only user-data columns containing an active ignore-null leaf.
+ IgnoreNullColumnReconstruction.reconstructColumn(
+ path = Seq(colName),
+ field = userDataColumnSchema(colName),
+ leafValues = ignoreNullContextsByTopLevelColumn(colName)
+ .map(context => context.path -> context.coalescedValue)
+ )
+ case colName =>
+ // Pass through keys, framework columns, and user data outside the
+ // active selection.
+ F.col(QuotingUtils.quoteIdentifier(colName))
+ }
+
+ // Evaluate the replacements and drop the temporary inheritance columns.
+ withValuesToInheritDf.select(outputColumns.toImmutableArraySeq: _*)
+ }
+
/**
* Convert surviving decomposition tails into tombstones.
*
@@ -1529,6 +1692,36 @@ object Scd2BatchProcessor {
.fieldNames
.toImmutableArraySeq
+ /** Materializes explicit unauthored entries required by retroactive schema evolution. */
+ private def updateVersionMapWithSchemaEvolution(
+ versionMap: Column,
+ leafInheritanceContexts: Seq[LeafInheritanceContext]): Column = {
+ if (leafInheritanceContexts.isEmpty) {
+ return versionMap
+ }
+
+ // Build one candidate version map entry per leaf and row: populated only when schema evolution
+ // requires an entry to be materialized for that leaf, and null otherwise.
+ val newEntryPerLeafOrNull = leafInheritanceContexts.map { context =>
+ val needsSchemaEvolutionEntry = Scd2VersionMap.needsSchemaEvolutionEntry(
+ versionMap, context.storedLeafValue, context.path, context.valueToInherit)
+ F.when(
+ needsSchemaEvolutionEntry,
+ Scd2VersionMap.buildVersionMapEntry(context.path, authored = false))
+ }
+
+ // Filter out null candidates, which represent leaves that do not need to gain a version map
+ // entry for this row.
+ val newEntries = F.filter(
+ F.array(newEntryPerLeafOrNull: _*), (entry: Column) => entry.isNotNull)
+
+ // Concat new entries mapping with existing version map.
+ F.when(
+ versionMap.isNotNull,
+ F.map_concat(versionMap, F.map_from_entries(newEntries))
+ )
+ }
+
/**
* Name of temporary column projected onto microbatch to compute the min sequencing value per
* key within the microbatch.
@@ -1669,11 +1862,9 @@ object Scd2BatchProcessor {
// decomposition tails, which are temporarily and synthetically constructed during
// reconciliation, have a null record start at.
StructField(recordStartAtFieldName, sequencingType, nullable = true),
- // The version map representing null-authorship for the row. For persisted rows:
- // If the version map is null, 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]].
+ // The version map representing null-authorship for the row. A null map on an
+ // upsert-representing row means no per-column authorship record has been established yet;
+ // [[Scd2BatchProcessor.coalesceIgnoredNulls]] may establish one when ignore-null is active.
//
// Tombstones and decomposition tails also always hold null version maps because column
// authorship is not applicable - they are delete markers.
@@ -1735,6 +1926,13 @@ private[autocdc] case class Scd2IntervalColumns(
F.lead(recordStartAt, offset).over(window),
F.lead(startAt, offset).over(window),
F.lead(endAt, offset).over(window))
+
+ /**
+ * The earliest visible timestamp of this row. For a normal upsert row `startAt` is
+ * non-null and wins; for a decomposition tail both `recordStartAt` and `startAt` are null
+ * and `endAt` surfaces via [[effectiveRecordStartAt]].
+ */
+ def effectiveStartAt: Column = F.coalesce(startAt, effectiveRecordStartAt)
}
object RowClassifier {
@@ -1807,6 +2005,22 @@ object RowClassifier {
): Column =
endAt.isNotNull && endAt < nextEffectiveRecordStartAt
+ /**
+ * Whether a row closes (`endAt`) strictly before the next row for the same key becomes
+ * visible, leaving a gap on the visible timeline that only a delete explains.
+ *
+ * Distinct from [[rowClosesStrictlyBeforeNextRow]], which compares against the next row's
+ * event sequence. Before reconciliation a target row standing for a collapsed no-op run
+ * holds the run's start in `startAt` and its last member's sequence in `recordStartAt`,
+ * while the members in between sit in the auxiliary table and need not be in the affected
+ * window. Comparing event sequences would read the interior of such a run as a gap.
+ */
+ private[autocdc] def rowClosesStrictlyBeforeNextRowIsVisible(
+ endAt: Column,
+ next: Scd2IntervalColumns
+ ): Column =
+ endAt.isNotNull && endAt < next.effectiveStartAt
+
/**
* Whether `row` carries no new information beyond its immediate successor `next` and so
* collapses into that successor's run instead of standing as its own visible interval. It is
@@ -1828,3 +2042,148 @@ object RowClassifier {
!rowClosesStrictlyBeforeNextRow(row.endAt, next.effectiveRecordStartAt) &&
areTrackedColumnsEqual
}
+
+/** Inheritance properties that apply to a row as a whole rather than to an individual leaf. */
+private[autocdc] case class RowInheritanceContext(
+ rowEndsInheritanceChain: Column,
+ isFirstRowInKeyWindow: Column,
+ isUpsertRepresentingRow: Column)
+
+private[autocdc] object RowInheritanceContext {
+
+ /**
+ * Projects the supplied expressions as temporary DataFrame columns and returns references to
+ * them. Keeping row-level window expressions out of each leaf's candidate expression prevents
+ * Catalyst from producing separate copies for every leaf.
+ */
+ def projectOn(
+ df: DataFrame,
+ rowEndsInheritanceChain: Column,
+ isUpsertRepresentingRow: Column,
+ isFirstRowInKeyWindow: Column): (DataFrame, RowInheritanceContext) = {
+ val projectedDf = df.withColumns(Map(
+ rowEndsInheritanceChainColName -> rowEndsInheritanceChain,
+ isFirstRowInKeyWindowColName -> isFirstRowInKeyWindow,
+ isUpsertRepresentingRowColName -> isUpsertRepresentingRow
+ ))
+ val projectedContext = RowInheritanceContext(
+ rowEndsInheritanceChain = F.col(rowEndsInheritanceChainColName),
+ isFirstRowInKeyWindow = F.col(isFirstRowInKeyWindowColName),
+ isUpsertRepresentingRow = F.col(isUpsertRepresentingRowColName))
+ (projectedDf, projectedContext)
+ }
+
+ private val rowEndsInheritanceChainColName =
+ s"${AutoCdcReservedNames.prefix}row_ends_inheritance_chain"
+
+ private val isFirstRowInKeyWindowColName =
+ s"${AutoCdcReservedNames.prefix}is_first_row_in_key_window"
+
+ private val isUpsertRepresentingRowColName =
+ s"${AutoCdcReservedNames.prefix}is_upsert_representing_row"
+}
+
+/**
+ * Per-leaf expressions that [[Scd2BatchProcessor.coalesceIgnoredNulls]] evaluates for one
+ * leaf of the eligible user-data schema.
+ *
+ * [[valueToInheritIfAny]] is projected into [[valueToInheritIfAnyColName]] so that
+ * [[valueToInherit]] and [[coalescedValue]] stay as plain expressions over an ordinary column.
+ *
+ * @param path the leaf's name parts within the row.
+ * @param index ordinal used to derive the temporary projected column name.
+ * @param storedLeafValue the leaf's value before this reconciliation.
+ * @param valueToInheritIfAny latest value contributed by a strictly preceding row, wrapped to
+ * distinguish no contribution from an explicit null.
+ * @param inherits whether this row should take [[valueToInherit]] over its stored value.
+ */
+private[autocdc] case class LeafInheritanceContext(
+ path: Seq[String],
+ index: Int,
+ storedLeafValue: Column,
+ valueToInheritIfAny: Column,
+ inherits: Column) {
+
+ /**
+ * Name of the temporary column that holds [[valueToInheritIfAny]], the wrapped value this row
+ * may inherit, during ignore-null coalescing.
+ */
+ val valueToInheritIfAnyColName: String =
+ LeafInheritanceContext.valueToInheritIfAnyColumnName(index)
+
+ /**
+ * Value extracted from the projected wrapper. It is null when the preceding contribution reset
+ * inheritance or no preceding contribution exists; [[inherits]] distinguishes those cases.
+ */
+ def valueToInherit: Column =
+ F.col(valueToInheritIfAnyColName)
+ .getField(LeafInheritanceContext.inheritanceValueFieldName)
+
+ /** The leaf's value after coalescing: [[valueToInherit]] if it [[inherits]], else stored. */
+ def coalescedValue: Column =
+ F.when(inherits, valueToInherit).otherwise(storedLeafValue)
+}
+
+private[autocdc] object LeafInheritanceContext {
+
+ /** Builds the per-leaf expressions used to coalesce unauthored nulls from preceding rows. */
+ def apply(
+ path: Seq[String],
+ index: Int,
+ versionMap: Column,
+ rowInheritanceContext: RowInheritanceContext,
+ precedingRowsInKeyWindow: WindowSpec): LeafInheritanceContext = {
+ val storedLeafValue = F.col(QuotingUtils.quoteNameParts(path))
+
+ // A row's stored value is authoritative when the row authored the leaf, or when the row is
+ // first in the key window. With no preceding row to recompute it from, the first row's stored
+ // value seeds inheritance for later rows even when unauthored.
+ val storedValueIsAuthoritative =
+ Scd2VersionMap.isAuthored(versionMap, storedLeafValue, path) ||
+ rowInheritanceContext.isFirstRowInKeyWindow
+
+ // Each row sees the candidate produced by the latest preceding row that updated the
+ // inheritance state. The current row can then emit one of three candidate updates for
+ // next rows: no update, a null update, or a non-null update. A nullable struct
+ // distinguishes no update from an update whose value is null.
+ //
+ // A chain end emits null so later rows cannot inherit across a delete boundary. Any other row
+ // with an authoritative stored value emits that value.
+ val rowResetsInheritanceChain = rowInheritanceContext.rowEndsInheritanceChain
+ val rowUpdatesInheritanceChain = rowResetsInheritanceChain || storedValueIsAuthoritative
+
+ // A row that does not update the inheritance chain produces a null outer column (hence
+ // "candidate"). Otherwise, the struct wraps either a null reset or a non-null value; its own
+ // nullability therefore encodes all three states.
+ val rowContributedInheritanceCandidate = F.when(
+ rowUpdatesInheritanceChain,
+ {
+ // If the row resets the inheritance chain, then it intentionally updates the inheritance
+ // chain with a null value. Otherwise it contributes its current leaf value as the new
+ // inheritance chain value.
+ val newInheritanceChainValue = F.when(!rowResetsInheritanceChain, storedLeafValue)
+ F.struct(newInheritanceChainValue.as(inheritanceValueFieldName))
+ }
+ )
+
+ LeafInheritanceContext(
+ path = path,
+ index = index,
+ storedLeafValue = storedLeafValue,
+ valueToInheritIfAny = F.last(
+ // The value to inherit for any row in the window is the last non-null inheritance
+ // candidate proposed by a preceding row. Recall if a row contributes a null candidate
+ // (different from struct{null}) then it declares it has no value to contribute to the
+ // inheritance chain.
+ rowContributedInheritanceCandidate,
+ ignoreNulls = true
+ ).over(precedingRowsInKeyWindow),
+ inherits = rowInheritanceContext.isUpsertRepresentingRow && !storedValueIsAuthoritative
+ )
+ }
+
+ private val inheritanceValueFieldName: String = "value"
+
+ private def valueToInheritIfAnyColumnName(index: Int): String =
+ s"${AutoCdcReservedNames.prefix}value_to_inherit_if_any_$index"
+}
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 124dfba9757a..30d91a97c187 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
@@ -57,6 +57,9 @@ case class Scd2ForeachBatchHandler(
* consume. Performs no writes: this is the entire pipeline up to (but excluding) the aux/target
* merges, factored out of [[execute]] so that both [[execute]] and tests exercise the exact same
* transform chain (they cannot silently desynchronize).
+ *
+ * Ignore-null coalescing applies only to rows pulled into this reconciliation. Changing the
+ * active selection does not trigger a full-history rewrite.
*/
private[autocdc] def reconcileMicrobatch(
batchDf: DataFrame,
@@ -116,7 +119,10 @@ case class Scd2ForeachBatchHandler(
.transform(d => batchProcessor.assertWellFormedRowsPostDecomposition(d, batchId))
.transform(batchProcessor.dropRedundantRowsPostDecomposition)
- val reconciledAndRoutedDf = decomposedDf
+ val withCoalescedIgnoredNullsDf =
+ batchProcessor.coalesceIgnoredNulls(decomposedDf)
+
+ val reconciledAndRoutedDf = withCoalescedIgnoredNullsDf
.transform(batchProcessor.reconcileStartAndEndAt)
.transform(batchProcessor.dropLeftoverDeletesPostReconciliation)
.transform(batchProcessor.promoteDecompositionTailsToTombstones)
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala
index e449ce32f310..61aed955710e 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala
@@ -84,8 +84,8 @@ private[pipelines] object Scd2VersionMap {
*
* @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 ignoreNullSelection The ignore-null selection under which to construct the version map
+ * for the provided schema.
* @param resolver Case-sensitivity resolver for column name matching.
* @return A [[Column]] of [[mapType]] schema.
*/
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/IgnoreNullColumnReconstructionSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/IgnoreNullColumnReconstructionSuite.scala
new file mode 100644
index 000000000000..ddebea55b5e2
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/IgnoreNullColumnReconstructionSuite.scala
@@ -0,0 +1,178 @@
+/*
+ * 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.{SparkException, SparkRuntimeException}
+import org.apache.spark.sql.{functions => F, Column, QueryTest, Row}
+import org.apache.spark.sql.classic.DataFrame
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+class IgnoreNullColumnReconstructionSuite extends QueryTest with SharedSparkSession {
+
+ /** Selects `field` from `source`, reconstructed from `leafValues`. */
+ private def reconstruct(
+ field: StructField,
+ leafValues: Seq[(Seq[String], Column)],
+ source: DataFrame = spark.range(1).toDF()): DataFrame =
+ source.select(
+ IgnoreNullColumnReconstruction.reconstructColumn(
+ path = Seq(field.name),
+ field = field,
+ leafValues = leafValues))
+
+ test("rebuilds a struct recursively from the values of its leaves") {
+ // Name parts are taken as is, so `zip.code` names a single field.
+ val locationType = new StructType().add("city", StringType).add("zip.code", StringType)
+ // An array is a leaf, even if its elements are structs.
+ val tagsType = ArrayType(new StructType().add("label", StringType))
+ val profile = StructField(
+ "profile",
+ new StructType().add("location", locationType).add("name", StringType).add("tags", tagsType))
+
+ checkAnswer(
+ reconstruct(profile, Seq(
+ Seq("profile", "location", "city") -> F.lit("SF"),
+ Seq("profile", "location", "zip.code") -> F.lit("94107"),
+ Seq("profile", "name") -> F.lit("Ada"),
+ Seq("profile", "tags") -> F.array(F.struct(F.lit("home"))))),
+ Row(Row(Row("SF", "94107"), "Ada", Seq(Row("home")))))
+ }
+
+ test("rebuilds a struct whose leaves are all null as a struct of nulls") {
+ // Every leaf is nullable, so even the non-nullable `location` can be a struct of nulls.
+ val locationType = new StructType().add("city", StringType).add("zip", StringType)
+ val profile = StructField(
+ "profile",
+ new StructType().add("location", locationType, nullable = false).add("name", StringType))
+ def leaves(city: Column, zip: Column, name: Column): Seq[(Seq[String], Column)] = Seq(
+ Seq("profile", "location", "city") -> city,
+ Seq("profile", "location", "zip") -> zip,
+ Seq("profile", "name") -> name)
+
+ checkAnswer(
+ reconstruct(profile, leaves(F.lit(null), F.lit(null), F.lit(null))),
+ Row(Row(Row(null, null), null)))
+ checkAnswer(
+ reconstruct(profile, leaves(F.lit(null), F.lit(null), F.lit("Ada"))),
+ Row(Row(Row(null, null), "Ada")))
+ }
+
+ test("nulls a struct whose leaves are all null if its schema forbids a struct of nulls") {
+ val locationType = StructType(Seq(StructField("city", StringType, nullable = false)))
+ // `profile` cannot be a struct of nulls, since `location` and `city` are non-nullable, but
+ // `profile` itself is nullable, so it is null instead.
+ val profile = StructField(
+ "profile", StructType(Seq(StructField("location", locationType, nullable = false))))
+ // `address` can be a struct of nulls around a null `home`.
+ val address = StructField(
+ "address", new StructType().add("home", locationType).add("note", StringType))
+ def addressLeaves(city: Column, note: Column): Seq[(Seq[String], Column)] = Seq(
+ Seq("address", "home", "city") -> city,
+ Seq("address", "note") -> note)
+
+ checkAnswer(
+ reconstruct(profile, Seq(Seq("profile", "location", "city") -> F.lit(null))),
+ Row(null))
+ checkAnswer(
+ reconstruct(profile, Seq(Seq("profile", "location", "city") -> F.lit("SF"))),
+ Row(Row(Row("SF"))))
+ checkAnswer(
+ reconstruct(address, addressLeaves(F.lit(null), F.lit(null))),
+ Row(Row(null, null)))
+ checkAnswer(
+ reconstruct(address, addressLeaves(F.lit(null), F.lit("note"))),
+ Row(Row(null, "note")))
+ checkAnswer(
+ reconstruct(address, addressLeaves(F.lit("LA"), F.lit(null))),
+ Row(Row(Row("LA"), null)))
+ }
+
+ test("keeps the declared type, nullability, and metadata of every field") {
+ def commentMetadata(comment: String): Metadata =
+ new MetadataBuilder().putString("comment", comment).build()
+
+ val locationType = StructType(Seq(
+ StructField("city", StringType, nullable = false, commentMetadata("city")),
+ StructField("zip", LongType, nullable = true, commentMetadata("zip"))))
+ val profileType = StructType(Seq(
+ StructField("location", locationType, nullable = true, commentMetadata("location")),
+ StructField("note", StringType, nullable = true, commentMetadata("note"))))
+ val profile = StructField("profile", profileType, nullable = true, commentMetadata("profile"))
+ val source = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row("SF"))), new StructType().add("city", StringType))
+
+ // The nullability of every value, and the type of `zip`, differ from those declared.
+ val result = reconstruct(
+ profile,
+ Seq(
+ Seq("profile", "location", "city") -> F.col("city"),
+ Seq("profile", "location", "zip") -> F.lit(94107),
+ Seq("profile", "note") -> F.lit("note")),
+ source)
+
+ assert(result.schema == StructType(Seq(profile)))
+ checkAnswer(result, Row(Row(Row("SF", 94107L), "note")))
+ }
+
+ test("raises NOT_NULL_ASSERT_VIOLATION if a non-nullable field must be reconstructed as null") {
+ def checkNotNullAssertViolation(result: DataFrame, walkedTypePath: Seq[String]): Unit =
+ checkError(
+ exception = intercept[SparkRuntimeException](result.collect()),
+ condition = "NOT_NULL_ASSERT_VIOLATION",
+ parameters = Map("walkedTypePath" -> walkedTypePath.mkString("\n", "\n", "\n")))
+
+ checkNotNullAssertViolation(
+ reconstruct(
+ StructField("name", StringType, nullable = false),
+ Seq(Seq("name") -> F.lit(null))),
+ walkedTypePath = Seq("name"))
+ // A null struct cannot replace a struct whose leaves are not all null.
+ val profileType = new StructType()
+ .add("city", StringType, nullable = false)
+ .add("note", StringType)
+ checkNotNullAssertViolation(
+ reconstruct(
+ StructField("profile", profileType),
+ Seq(Seq("profile", "city") -> F.lit(null), Seq("profile", "note") -> F.lit("note"))),
+ walkedTypePath = Seq("profile", "city"))
+ }
+
+ test("rebuilds a struct without leaves as an empty struct") {
+ val profile = StructField(
+ "profile", new StructType().add("empty", new StructType()).add("name", StringType))
+
+ val result = reconstruct(profile, Seq(Seq("profile", "name") -> F.lit("Ada")))
+
+ assert(result.schema == StructType(Seq(profile)))
+ checkAnswer(result, Row(Row(Row(), "Ada")))
+ checkAnswer(reconstruct(StructField("empty", new StructType()), Seq.empty), Row(Row()))
+ }
+
+ test("fails with an internal error if a leaf has no value") {
+ val profile = StructField(
+ "profile", new StructType().add("city", StringType).add("zip", StringType))
+
+ checkError(
+ exception = intercept[SparkException] {
+ reconstruct(profile, Seq(Seq("profile", "city") -> F.lit("SF")))
+ },
+ condition = "INTERNAL_ERROR",
+ parameters = Map("message" -> "Cannot reconstruct `profile`.`zip` because it has no value."))
+ }
+}
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2CoalesceIgnoredNullsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2CoalesceIgnoredNullsSuite.scala
new file mode 100644
index 000000000000..68e9bb2f861e
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2CoalesceIgnoredNullsSuite.scala
@@ -0,0 +1,695 @@
+/*
+ * 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.SparkRuntimeException
+import org.apache.spark.sql.{functions => F, AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.util.QuotingUtils
+import org.apache.spark.sql.classic.DataFrame
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+class Scd2CoalesceIgnoredNullsSuite extends QueryTest with SharedSparkSession {
+
+ private def includeColumns(columnNames: String*): ColumnSelection =
+ ColumnSelection.IncludeColumns(columnNames.map(UnqualifiedColumnName(_)))
+
+ private def excludeColumns(columnNames: String*): ColumnSelection =
+ ColumnSelection.ExcludeColumns(columnNames.map(UnqualifiedColumnName(_)))
+
+ private def versionMap(entries: (Seq[String], Boolean)*): Map[String, Boolean] =
+ // Construct persisted version-map states directly; Scd2VersionMapSuite tests their production.
+ entries.map { case (path, authored) =>
+ QuotingUtils.quoteNameParts(path) -> authored
+ }.toMap
+
+ private def cdcMetadata(
+ recordStartAt: java.lang.Long,
+ versionMap: Map[String, Boolean]): Row =
+ Row(recordStartAt, versionMap)
+
+ private def targetTableOf(
+ userSchema: StructType,
+ cdcColumnMetadata: Metadata = Metadata.empty)(rows: Row*): DataFrame = {
+ val schema = StructType(userSchema.fields.toSeq ++ Seq(
+ StructField(Scd2BatchProcessor.startAtColName, LongType, nullable = true),
+ StructField(Scd2BatchProcessor.endAtColName, LongType, nullable = true),
+ StructField(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(LongType),
+ nullable = false,
+ cdcColumnMetadata)
+ ))
+ spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
+ }
+
+ private def processor(ignoreNullSelection: ColumnSelection): Scd2BatchProcessor =
+ Scd2BatchProcessor(
+ changeArgs = ChangeArgs(
+ keys = Seq(UnqualifiedColumnName("id")),
+ sequencing = F.col("seq"),
+ storedAsScdType = ScdType.Type2,
+ ignoreNullSelection = Some(ignoreNullSelection)
+ ),
+ resolvedSequencingType = LongType
+ )
+
+ private def coalesce(
+ df: DataFrame,
+ ignoreNullSelection: ColumnSelection): DataFrame =
+ processor(ignoreNullSelection).coalesceIgnoredNulls(df)
+
+ test("rows without version maps establish authorship using the current selection") {
+ val selection = includeColumns("selected")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("selected", StringType)
+ .add("unselected", StringType)
+ val existingMap = versionMap(Seq("unselected") -> true)
+ val initializedMap = versionMap(
+ Seq("selected") -> false,
+ Seq("unselected") -> true)
+
+ val input = targetTableOf(schema)(
+ Row(1, "source", null, 10L, null, cdcMetadata(10L, existingMap)),
+ Row(1, null, null, 20L, null, cdcMetadata(20L, null))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "source", null, 10L, null, cdcMetadata(10L, existingMap)),
+ Row(1, "source", null, 20L, null, cdcMetadata(20L, initializedMap))
+ )
+ )
+ }
+
+ test("current selection gates reconciliation while preserving existing map authorship") {
+ // Existing maps remain the source of truth for ingestion-time authorship. The current
+ // selection only chooses which leaves to reconcile: adding a leaf consults its recorded
+ // authorship, while removing a leaf stops coalescing it without rewriting its map entry.
+ val selection = includeColumns("selectedNow")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("selectedNow", StringType)
+ .add("removedNow", StringType)
+ val emptyMap = versionMap()
+ val authoredSelectedMap = versionMap(
+ Seq("selectedNow") -> true,
+ Seq("removedNow") -> false)
+ val unauthoredSelectedMap = versionMap(
+ Seq("selectedNow") -> false,
+ Seq("removedNow") -> false)
+
+ val input = targetTableOf(schema)(
+ Row(1, "selected-1", "removed-1", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, null, 20L, null, cdcMetadata(20L, authoredSelectedMap)),
+ Row(2, "selected-2", "removed-2", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(2, null, null, 20L, null, cdcMetadata(20L, unauthoredSelectedMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "selected-1", "removed-1", 10L, null, cdcMetadata(10L, emptyMap)),
+ // The newly selected leaf retains its previously recorded authored null.
+ Row(1, null, null, 20L, null, cdcMetadata(20L, authoredSelectedMap)),
+ Row(2, "selected-2", "removed-2", 10L, null, cdcMetadata(10L, emptyMap)),
+ // The selected leaf inherits; the removed leaf is no longer reconciled.
+ Row(2, "selected-2", null, 20L, null, cdcMetadata(20L, unauthoredSelectedMap))
+ )
+ )
+ }
+
+ test("schema-evolved leaves gain entries only when they inherit non-null values") {
+ val selection = includeColumns("evolved")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("evolved", StringType)
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("evolved") -> false)
+
+ val input = targetTableOf(schema)(
+ Row(1, "source", 10L, null, cdcMetadata(10L, emptyMap)),
+ // No entry models a leaf added after this row's non-null map was established.
+ Row(1, null, 20L, null, cdcMetadata(20L, emptyMap)),
+ // Without a preceding value, the absent entry remains the sparse unauthored signal.
+ Row(2, null, 20L, null, cdcMetadata(20L, emptyMap))
+ )
+
+ val firstPass = coalesce(input, selection)
+ checkAnswer(
+ firstPass,
+ Seq(
+ Row(1, "source", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, "source", 20L, null, cdcMetadata(20L, unauthoredMap)),
+ Row(2, null, 20L, null, cdcMetadata(20L, emptyMap))
+ )
+ )
+
+ // On a later reconciliation, an authored row arrives before the previously inherited row.
+ // Its explicit false entry keeps the inherited non-null value unauthored, allowing the newer
+ // preceding value to replace it instead of treating the stored value as authoritative.
+ val previouslyInheritedRow = firstPass.filter(
+ F.col("id") === 1 &&
+ F.col(AutoCdcReservedNames.cdcMetadataColName)
+ .getField(Scd2BatchProcessor.recordStartAtFieldName) === 20L)
+ val newPrecedingRow = targetTableOf(schema)(
+ Row(1, "new-source", 15L, null, cdcMetadata(15L, emptyMap))
+ )
+
+ checkAnswer(
+ coalesce(newPrecedingRow.unionByName(previouslyInheritedRow), selection),
+ Seq(
+ Row(1, "new-source", 15L, null, cdcMetadata(15L, emptyMap)),
+ Row(1, "new-source", 20L, null, cdcMetadata(20L, unauthoredMap))
+ )
+ )
+ }
+
+ test("each post-decomposition boundary resets inheritance") {
+ val selection = includeColumns("value")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("value") -> false)
+
+ // Every boundary row carries non-null data, so a leak into the inheritance chain would
+ // surface as that value instead of the expected null.
+ val input = targetTableOf(schema)(
+ // Key 1: a leading tombstone resets values inherited before the affected suffix.
+ Row(1, "tombstone", 20L, 20L, cdcMetadata(20L, null)),
+ Row(1, "stale", 10L, null, cdcMetadata(30L, unauthoredMap)),
+ // Key 2: a decomposition tail represents the same kind of delete boundary.
+ Row(2, "tail", null, 20L, cdcMetadata(null, null)),
+ Row(2, "stale", 10L, null, cdcMetadata(30L, unauthoredMap)),
+ // Key 3: a closed interval followed by a visibility gap also ends inheritance.
+ Row(3, "before-gap", 10L, 20L, cdcMetadata(10L, emptyMap)),
+ Row(3, "stale", 30L, null, cdcMetadata(30L, unauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "tombstone", 20L, 20L, cdcMetadata(20L, null)),
+ Row(1, null, 10L, null, cdcMetadata(30L, unauthoredMap)),
+ Row(2, "tail", null, 20L, cdcMetadata(null, null)),
+ Row(2, null, 10L, null, cdcMetadata(30L, unauthoredMap)),
+ Row(3, "before-gap", 10L, 20L, cdcMetadata(10L, emptyMap)),
+ Row(3, null, 30L, null, cdcMetadata(30L, unauthoredMap))
+ )
+ )
+ }
+
+ test("an authored value restarts inheritance after a reset") {
+ val selection = includeColumns("value")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("value") -> false)
+
+ val input = targetTableOf(schema)(
+ Row(1, "old", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, 20L, 20L, cdcMetadata(20L, null)),
+ Row(1, "new", 30L, null, cdcMetadata(30L, emptyMap)),
+ Row(1, null, 40L, null, cdcMetadata(40L, unauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "old", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, 20L, 20L, cdcMetadata(20L, null)),
+ Row(1, "new", 30L, null, cdcMetadata(30L, emptyMap)),
+ Row(1, "new", 40L, null, cdcMetadata(40L, unauthoredMap))
+ )
+ )
+ }
+
+ test("unselected data and framework columns retain their values and metadata") {
+ val selection = includeColumns("selected")
+ def commentMetadata(comment: String): Metadata =
+ new MetadataBuilder().putString("comment", comment).build()
+
+ val schema = StructType(Seq(
+ StructField("id", IntegerType, nullable = true, commentMetadata("key")),
+ StructField("selected", StringType, nullable = true, commentMetadata("selected")),
+ StructField("unselected", StringType, nullable = true, commentMetadata("unselected"))
+ ))
+ val cdcColumnMetadata = commentMetadata("cdc")
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("selected") -> false)
+ val input = targetTableOf(schema, cdcColumnMetadata)(
+ Row(1, "source", "keep-1", 10L, 20L, cdcMetadata(10L, emptyMap)),
+ Row(1, null, "keep-2", 20L, 60L, cdcMetadata(20L, unauthoredMap))
+ )
+
+ val result = coalesce(input, selection)
+
+ assert(result.schema == input.schema)
+ checkAnswer(
+ result,
+ Seq(
+ Row(1, "source", "keep-1", 10L, 20L, cdcMetadata(10L, emptyMap)),
+ Row(1, "source", "keep-2", 20L, 60L, cdcMetadata(20L, unauthoredMap))
+ )
+ )
+ }
+
+ test("quoted top-level columns coalesce or pass through according to ignore-null selection") {
+ val selected = "selected.with.dot"
+ val unselected = "unselected.with.dot"
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add(selected, StringType)
+ .add(unselected, StringType)
+ val authoredMap = versionMap()
+ val mixedMap = versionMap(
+ Seq(selected) -> false,
+ Seq(unselected) -> true)
+ val input = targetTableOf(schema)(
+ Row(1, "inherit", "keep", 10L, null, cdcMetadata(10L, authoredMap)),
+ Row(1, null, null, 20L, null, cdcMetadata(20L, mixedMap)))
+
+ checkAnswer(
+ coalesce(input, includeColumns(QuotingUtils.quoteIdentifier(selected))),
+ Seq(
+ Row(1, "inherit", "keep", 10L, null, cdcMetadata(10L, authoredMap)),
+ Row(1, "inherit", null, 20L, null, cdcMetadata(20L, mixedMap))))
+ }
+
+ test("inheriting into a stored null struct fails when a non-nullable field is authored null") {
+ // Seq 20 stores profile = null, and its version map records profile.city as an authored null.
+ // profile.note was added by schema evolution, so it has no entry and inherits "note-1" from
+ // seq 10. Holding that inherited value requires a non-null profile struct, but city's
+ // authored null cannot be stored in a non-nullable field, so coalescing fails. This represents
+ // a schema misconfiguration issue, where the user is choosing a schema that simply cannot
+ // represent the coalesced result for this set and order of change events. This test locks in
+ // the error contract.
+ val profileType = new StructType()
+ .add("city", StringType, nullable = false)
+ .add("note", StringType)
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("profile", profileType)
+ val input = targetTableOf(schema)(
+ Row(1, Row("city-1", "note-1"), 10L, null, cdcMetadata(10L, versionMap())),
+ Row(1, null, 20L, null,
+ cdcMetadata(20L, versionMap(Seq("profile", "city") -> true)))
+ )
+
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ coalesce(input, includeColumns("profile")).collect()
+ },
+ condition = "NOT_NULL_ASSERT_VIOLATION",
+ parameters = Map("walkedTypePath" -> "\nprofile\ncity\n")
+ )
+ }
+
+ gridTest("column selection honors the configured resolver")(
+ Seq(
+ (false, "VALUE"),
+ (true, "Value")
+ )
+ ) { case (caseSensitive, selectedColumnName) =>
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) {
+ val selection = includeColumns(selectedColumnName)
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("Value", StringType)
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("Value") -> false)
+ val input = targetTableOf(schema)(
+ Row(1, "source", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, 20L, null, cdcMetadata(20L, unauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "source", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, "source", 20L, null, cdcMetadata(20L, unauthoredMap))
+ )
+ )
+ }
+ }
+
+ test("case-sensitive selection rejects mismatched column casing") {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ val selection = includeColumns("VALUE")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("Value", StringType)
+ val input = targetTableOf(schema)(
+ Row(1, "source", 10L, null, cdcMetadata(10L, versionMap()))
+ )
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ coalesce(input, selection)
+ },
+ condition = "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA",
+ sqlState = "42703",
+ parameters = Map(
+ "caseSensitivity" -> CaseSensitivityLabels.CaseSensitive,
+ "schemaName" -> "ignoreNullSelection",
+ "missingColumns" -> "VALUE",
+ "availableColumns" -> "Value"
+ )
+ )
+ }
+ }
+
+ test("coalescing is idempotent and a previously coalesced first row supplies carry-in") {
+ val selection = includeColumns("value")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("value") -> false)
+ val input = targetTableOf(schema)(
+ Row(1, "source", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, 20L, null, cdcMetadata(20L, unauthoredMap))
+ )
+
+ val firstPass = coalesce(input, selection)
+ val firstPassRows = Seq(
+ Row(1, "source", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, "source", 20L, null, cdcMetadata(20L, unauthoredMap))
+ )
+ checkAnswer(firstPass, firstPassRows)
+ checkAnswer(coalesce(firstPass, selection), firstPassRows)
+
+ val affectedSuffix = firstPass.filter(
+ F.col(AutoCdcReservedNames.cdcMetadataColName)
+ .getField(Scd2BatchProcessor.recordStartAtFieldName) >= 20L)
+ val incoming = targetTableOf(schema)(
+ Row(1, null, 30L, null, cdcMetadata(30L, unauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(affectedSuffix.unionByName(incoming), selection),
+ Seq(
+ Row(1, "source", 20L, null, cdcMetadata(20L, unauthoredMap)),
+ Row(1, "source", 30L, null, cdcMetadata(30L, unauthoredMap))
+ )
+ )
+ }
+
+ test("incoming authored values replace carry-in without unauthored rows interrupting it") {
+ val selection = includeColumns("value")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("value") -> false)
+ val input = targetTableOf(schema)(
+ // The affected suffix starts with an existing, previously coalesced value.
+ Row(1, "old", 10L, null, cdcMetadata(10L, unauthoredMap)),
+ Row(1, "new", 20L, null, cdcMetadata(20L, emptyMap)),
+ // Neither a previously inherited non-null nor an unauthored null contributes a candidate.
+ Row(1, "stale", 30L, null, cdcMetadata(30L, unauthoredMap)),
+ Row(1, null, 40L, null, cdcMetadata(40L, unauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "old", 10L, null, cdcMetadata(10L, unauthoredMap)),
+ Row(1, "new", 20L, null, cdcMetadata(20L, emptyMap)),
+ Row(1, "new", 30L, null, cdcMetadata(30L, unauthoredMap)),
+ Row(1, "new", 40L, null, cdcMetadata(40L, unauthoredMap))
+ )
+ )
+ }
+
+ test("an authored null replaces an older non-null inheritance candidate") {
+ val selection = includeColumns("value")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ val emptyMap = versionMap()
+ val authoredNullMap = versionMap(Seq("value") -> true)
+ val unauthoredMap = versionMap(Seq("value") -> false)
+ val input = targetTableOf(schema)(
+ Row(1, "old", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, 20L, null, cdcMetadata(20L, authoredNullMap)),
+ Row(1, "stale", 30L, null, cdcMetadata(30L, unauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "old", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, 20L, null, cdcMetadata(20L, authoredNullMap)),
+ Row(1, null, 30L, null, cdcMetadata(30L, unauthoredMap))
+ )
+ )
+ }
+
+ test("each selected leaf maintains an independent inheritance chain") {
+ val selection = includeColumns("left", "right")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("left", StringType)
+ .add("right", StringType)
+ val emptyMap = versionMap()
+ val leftUnauthoredMap = versionMap(Seq("left") -> false)
+ val rightUnauthoredMap = versionMap(Seq("right") -> false)
+ val input = targetTableOf(schema)(
+ Row(1, "left-1", "right-1", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, "right-2", 20L, null, cdcMetadata(20L, leftUnauthoredMap)),
+ Row(1, "left-3", null, 30L, null, cdcMetadata(30L, rightUnauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "left-1", "right-1", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, "left-1", "right-2", 20L, null, cdcMetadata(20L, leftUnauthoredMap)),
+ Row(1, "left-3", "right-2", 30L, null, cdcMetadata(30L, rightUnauthoredMap))
+ )
+ )
+ }
+
+ test("selected structs are rebuilt from their leaves, including null structs") {
+ val selection = includeColumns("profile")
+ val profileSchema = new StructType()
+ .add("display.name", StringType)
+ .add("age", IntegerType)
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("profile", profileSchema)
+ .add("outside", StringType)
+ val authoredAgeMap = versionMap(Seq("profile", "age") -> true)
+ val authoredProfileMap = versionMap(
+ Seq("profile", "display.name") -> true,
+ Seq("profile", "age") -> true)
+ val unauthoredProfileMap = versionMap(
+ Seq("profile", "display.name") -> false,
+ Seq("profile", "age") -> false)
+ val input = targetTableOf(schema)(
+ Row(1, Row("Alice", null), "keep-1", 10L, null, cdcMetadata(10L, authoredAgeMap)),
+ Row(1, null, "keep-2", 20L, null, cdcMetadata(20L, unauthoredProfileMap)),
+ Row(2, null, "keep-3", 10L, null, cdcMetadata(10L, authoredProfileMap)),
+ Row(2, null, "keep-4", 20L, null, cdcMetadata(20L, unauthoredProfileMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, Row("Alice", null), "keep-1", 10L, null,
+ cdcMetadata(10L, authoredAgeMap)),
+ Row(1, Row("Alice", null), "keep-2", 20L, null,
+ cdcMetadata(20L, unauthoredProfileMap)),
+ Row(2, Row(null, null), "keep-3", 10L, null, cdcMetadata(10L, authoredProfileMap)),
+ Row(2, Row(null, null), "keep-4", 20L, null, cdcMetadata(20L, unauthoredProfileMap))
+ )
+ )
+ }
+
+ test("arrays and maps are inherited as opaque leaves") {
+ val selection = includeColumns("tags", "properties")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("tags", ArrayType(StringType))
+ .add("properties", MapType(StringType, IntegerType))
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(
+ Seq("tags") -> false,
+ Seq("properties") -> false)
+ val input = targetTableOf(schema)(
+ Row(1, Seq("one", "two"), Map("x" -> 1), 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, null, null, 20L, null, cdcMetadata(20L, unauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, Seq("one", "two"), Map("x" -> 1), 10L, null,
+ cdcMetadata(10L, emptyMap)),
+ Row(1, Seq("one", "two"), Map("x" -> 1), 20L, null,
+ cdcMetadata(20L, unauthoredMap))
+ )
+ )
+ }
+
+ test("inheritance follows chronological order independently for each key") {
+ val selection = includeColumns("value")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("value") -> false)
+ val input = targetTableOf(schema)(
+ // Physical input order deliberately differs from chronological order and interleaves keys.
+ Row(1, null, 20L, null, cdcMetadata(20L, unauthoredMap)),
+ Row(2, "key-2", 5L, null, cdcMetadata(5L, emptyMap)),
+ Row(1, "key-1", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(2, null, 15L, null, cdcMetadata(15L, unauthoredMap))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, "key-1", 10L, null, cdcMetadata(10L, emptyMap)),
+ Row(1, "key-1", 20L, null, cdcMetadata(20L, unauthoredMap)),
+ Row(2, "key-2", 5L, null, cdcMetadata(5L, emptyMap)),
+ Row(2, "key-2", 15L, null, cdcMetadata(15L, unauthoredMap))
+ )
+ )
+ }
+
+ test("a selection resolving to no leaves initializes upserts but leaves deletes unchanged") {
+ val selection = excludeColumns("value", "other")
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("other", IntegerType)
+ val authoredNullsMap = versionMap(
+ Seq("value") -> true,
+ Seq("other") -> true)
+ val input = targetTableOf(schema)(
+ Row(1, null, null, 10L, null, cdcMetadata(10L, null)),
+ Row(2, null, null, 20L, 20L, cdcMetadata(20L, null))
+ )
+
+ checkAnswer(
+ coalesce(input, selection),
+ Seq(
+ Row(1, null, null, 10L, null, cdcMetadata(10L, authoredNullsMap)),
+ Row(2, null, null, 20L, 20L, cdcMetadata(20L, null))
+ )
+ )
+ }
+
+ test("a wider affected window corrects inherited values after selection changes") {
+ // Rows are reconciled only within the provided affected window. A narrow window can propagate
+ // a previously inherited value when a newer authoring row lies outside it, while a later wider
+ // window containing that row corrects the affected rows. This test constructs both windows
+ // directly; it does not assert that the wider window will be scheduled.
+
+ // "other" is authored and non-null in every row, so coalescing never changes it. It exists
+ // only so the selection can move off "value" without becoming an (invalid) empty list.
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("other", StringType)
+ val emptyMap = versionMap()
+ val unauthoredMap = versionMap(Seq("value") -> false)
+
+ // -- Batch 1 (selection = {value}) ------------------------------------------------
+ // Two rows arrive for key 1: seq 1 authors "old", seq 3 has an unauthored null.
+ // The affected window spans both rows, so seq 3 inherits "old" from seq 1.
+ val batch1Input = targetTableOf(schema)(
+ Row(1, "old", "x", 1L, null, cdcMetadata(1L, emptyMap)),
+ Row(1, null, "x", 3L, null, cdcMetadata(3L, unauthoredMap))
+ )
+ checkAnswer(
+ coalesce(batch1Input, includeColumns("value")),
+ Seq(
+ Row(1, "old", "x", 1L, null, cdcMetadata(1L, emptyMap)),
+ Row(1, "old", "x", 3L, null, cdcMetadata(3L, unauthoredMap))
+ )
+ )
+
+ // -- Batch 2 (selection = {other} -- "value" temporarily removed) -----------------
+ // An out-of-order seq 2 arrives with value = "new", slotting between seq 1 and
+ // seq 3. The decomposed target state after batch 2:
+ // seq 1: value="old", startAt=1, endAt=2, versionMap={}
+ // seq 2: value="new", startAt=2, endAt=3, versionMap={}
+ // seq 3: value="old", startAt=3, endAt=null, versionMap={value->false}
+ val batch2AffectedWindow = targetTableOf(schema)(
+ Row(1, "old", "x", 1L, 2L, cdcMetadata(1L, emptyMap)),
+ Row(1, "new", "x", 2L, 3L, cdcMetadata(2L, emptyMap)),
+ Row(1, "old", "x", 3L, null, cdcMetadata(3L, unauthoredMap))
+ )
+ // "value" is not in the active selection, so it is not reconciled. Seq 3 retains its
+ // previously inherited "old" even though seq 2 (which authors "new") now directly
+ // precedes it.
+ checkAnswer(
+ coalesce(batch2AffectedWindow, includeColumns("other")),
+ Seq(
+ Row(1, "old", "x", 1L, 2L, cdcMetadata(1L, emptyMap)),
+ Row(1, "new", "x", 2L, 3L, cdcMetadata(2L, emptyMap)),
+ Row(1, "old", "x", 3L, null, cdcMetadata(3L, unauthoredMap))
+ )
+ )
+
+ // -- Batch 3 (selection = {value} -- "value" re-added) ----------------------------
+ // Seq 4 arrives with an unauthored null. The affected window covers only the rows
+ // touched by this event: seq 3 and seq 4 (the new row). Seq 2, which
+ // holds the correct "new", is outside the window.
+ //
+ // Seq 3 still stores "old" because "value" was not reconciled in batch 2. With no preceding
+ // authoring row in this window, seq 3 supplies "old" and seq 4 inherits it instead of "new".
+ val batch3Input = targetTableOf(schema)(
+ Row(1, "old", "x", 3L, 4L, cdcMetadata(3L, unauthoredMap)),
+ Row(1, null, "x", 4L, null, cdcMetadata(4L, unauthoredMap))
+ )
+ checkAnswer(
+ coalesce(batch3Input, includeColumns("value")),
+ Seq(
+ Row(1, "old", "x", 3L, 4L, cdcMetadata(3L, unauthoredMap)),
+ Row(1, "old", "x", 4L, null, cdcMetadata(4L, unauthoredMap))
+ )
+ )
+
+ // -- Correction -------------------------------------------------------------------
+ // Suppose a later batch's affected window includes seq 2. It is now the nearest preceding row
+ // that authored "value", so "new" propagates through seq 3 and seq 4, correcting their
+ // previously inherited values.
+ val correctionInput = targetTableOf(schema)(
+ Row(1, "new", "x", 2L, 3L, cdcMetadata(2L, emptyMap)),
+ Row(1, "old", "x", 3L, 4L, cdcMetadata(3L, unauthoredMap)),
+ Row(1, "old", "x", 4L, null, cdcMetadata(4L, unauthoredMap))
+ )
+ checkAnswer(
+ coalesce(correctionInput, includeColumns("value")),
+ Seq(
+ Row(1, "new", "x", 2L, 3L, cdcMetadata(2L, emptyMap)),
+ Row(1, "new", "x", 3L, 4L, cdcMetadata(3L, unauthoredMap)),
+ Row(1, "new", "x", 4L, null, cdcMetadata(4L, unauthoredMap))
+ )
+ )
+ }
+}
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala
index 008313b1ee17..616d5b4f8424 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.pipelines.autocdc
import org.scalatest.BeforeAndAfter
import org.apache.spark.sql.{functions => F, AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.util.QuotingUtils
import org.apache.spark.sql.classic.{DataFrame, Dataset}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -86,6 +87,15 @@ class Scd2ForeachBatchHandlerSuite
resolvedSequencingType = LongType
)
+ private val ignoreNullProcessor = processor.copy(
+ changeArgs = processor.changeArgs.copy(
+ ignoreNullSelection = Some(ColumnSelection.IncludeColumns(
+ Seq(UnqualifiedColumnName("value")))))
+ )
+ private val emptyVersionMap = Map.empty[String, Boolean]
+ private val unauthoredValueVersionMap =
+ Map(QuotingUtils.quoteNameParts(Seq("value")) -> false)
+
private def createAuxTable(seedRows: Row*): Unit =
createTable(defaultAuxIdent, defaultAuxTableIdentifier, auxSchema, seedRows: _*)
@@ -120,8 +130,9 @@ class Scd2ForeachBatchHandlerSuite
value: String,
startAt: java.lang.Long,
endAt: java.lang.Long,
- recordStartAt: Long): Row =
- Row(id, value, startAt, endAt, meta(recordStartAt))
+ recordStartAt: Long,
+ versionMap: Any = null): Row =
+ Row(id, value, startAt, endAt, meta(recordStartAt, versionMap))
/** A canonical aux row `(id, value, startAt, endAt, meta(recordStartAt), deletedByBatchId)`. */
private def auxRow(
@@ -130,13 +141,18 @@ class Scd2ForeachBatchHandlerSuite
startAt: java.lang.Long,
endAt: java.lang.Long,
recordStartAt: Long,
- deletedByBatchId: java.lang.Long): Row =
- Row(id, value, startAt, endAt, meta(recordStartAt), deletedByBatchId)
+ deletedByBatchId: java.lang.Long,
+ versionMap: Any = null): Row =
+ Row(id, value, startAt, endAt, meta(recordStartAt, versionMap), deletedByBatchId)
/** Run a microbatch of source rows through the default handler. */
private def runBatch(batchId: Long)(rows: Row*): Unit =
exec.execute(microbatchOf(sourceSchema)(rows: _*), batchId)
+ /** Run a microbatch with ignore-null enabled for `value`. */
+ private def runIgnoreNullBatch(batchId: Long)(rows: Row*): Unit =
+ execWith(ignoreNullProcessor).execute(microbatchOf(sourceSchema)(rows: _*), batchId)
+
/**
* Run `rows` as batch `batchId`, capture both tables, then replay the identical batch under the
* same `batchId` and assert both tables are byte-for-byte unchanged. Models a crash/redelivery
@@ -159,11 +175,15 @@ class Scd2ForeachBatchHandlerSuite
* updated. On recovery Structured Streaming reruns the same `batchId`, which
* [[Scd2BatchProcessor.deletedByBatchIdColName]] is designed to make idempotent.
*/
- private def runBatchAuxMergeOnly(batchId: Long)(rows: Row*): Unit = {
+ private def runBatchAuxMergeOnly(
+ batchId: Long,
+ p: Scd2BatchProcessor = processor)(rows: Row*): Unit = {
// Reuse the handler's own reconciliation chain so this helper cannot drift from execute(),
// then run only the aux merge (skipping the target merge) to model the mid-batch crash.
- val reconciled = exec.reconcileMicrobatch(microbatchOf(sourceSchema)(rows: _*), batchId)
- processor.mergeRowsIntoAuxiliaryTable(
+ val reconciled = execWith(p).reconcileMicrobatch(
+ microbatchOf(sourceSchema)(rows: _*),
+ batchId)
+ p.mergeRowsIntoAuxiliaryTable(
reconciledDfWithAuxRowsTagged = reconciled.reconciledAndRoutedDf,
originalAffectedRowsFromAuxiliaryTable = reconciled.affectedRowsFromAuxiliaryTable,
auxiliaryTableIdentifier = defaultAuxTableIdentifier,
@@ -691,6 +711,74 @@ class Scd2ForeachBatchHandlerSuite
checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null))
}
+ test("recovering after a crash between merges converges with ignore-null coalescing") {
+ // The null event inherits "a", making both events one run. The first event is therefore
+ // written to aux before the crash, while the inherited tail and its unauthored map have not
+ // reached the target. Replaying the same batch must reproduce the clean single-run result.
+ createAuxTable()
+ createTargetTable()
+
+ val rows = Seq(upsert(1, "a", 10L), upsert(1, null, 20L))
+ runBatchAuxMergeOnly(1L, ignoreNullProcessor)(rows: _*)
+
+ checkAnswer(
+ auxTable,
+ auxRow(
+ 1, "a", 10L, null, 10L, deletedByBatchId = null, versionMap = emptyVersionMap))
+ assert(targetTable.collect().isEmpty)
+
+ runIgnoreNullBatch(1L)(rows: _*)
+
+ checkAnswer(targetTable, targetRow(1, "a", 10L, null, 20L, unauthoredValueVersionMap))
+ checkAnswer(
+ auxTable,
+ auxRow(
+ 1, "a", 10L, null, 10L, deletedByBatchId = null, versionMap = emptyVersionMap))
+ }
+
+ test("retry after an aux-only merge reuses a previously coalesced ignore-null carry-in") {
+ // Batch 1 establishes a coalesced run. Batch 2 then crashes after demoting its visible tail
+ // into aux but before advancing the target. The retry sees that row in both tables; it must
+ // deduplicate the copies, use the coalesced value as carry-in, and advance the run exactly
+ // once.
+ createAuxTable()
+ createTargetTable()
+
+ runIgnoreNullBatch(1L)(upsert(1, "a", 10L), upsert(1, null, 20L))
+ checkAnswer(targetTable, targetRow(1, "a", 10L, null, 20L, unauthoredValueVersionMap))
+ checkAnswer(
+ auxTable,
+ auxRow(
+ 1, "a", 10L, null, 10L, deletedByBatchId = null, versionMap = emptyVersionMap))
+
+ runBatchAuxMergeOnly(2L, ignoreNullProcessor)(upsert(1, null, 30L))
+ checkAnswer(targetTable, targetRow(1, "a", 10L, null, 20L, unauthoredValueVersionMap))
+ checkAnswer(
+ auxTable,
+ Seq(
+ auxRow(
+ 1, "a", 10L, null, 10L, deletedByBatchId = null, versionMap = emptyVersionMap),
+ auxRow(
+ 1, "a", 10L, null, 20L,
+ deletedByBatchId = null, versionMap = unauthoredValueVersionMap)
+ )
+ )
+
+ runIgnoreNullBatch(2L)(upsert(1, null, 30L))
+
+ checkAnswer(targetTable, targetRow(1, "a", 10L, null, 30L, unauthoredValueVersionMap))
+ checkAnswer(
+ auxTable,
+ Seq(
+ auxRow(
+ 1, "a", 10L, null, 10L, deletedByBatchId = null, versionMap = emptyVersionMap),
+ auxRow(
+ 1, "a", 10L, null, 20L,
+ deletedByBatchId = null, versionMap = unauthoredValueVersionMap)
+ )
+ )
+ }
+
test("recovering after a crash that logically deleted a pre-existing aux row converges") {
// This is the case the deletedByBatchId re-inclusion clause exists for: unlike the crash tests
// above (fresh inserts, deletedByBatchId = null), here the crashed attempt logically DELETES a
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org