AnishMahto commented on code in PR #58947:
URL: https://github.com/apache/spark/pull/58947#discussion_r4200491178
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -967,6 +968,193 @@ 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. Existing maps are
preserved, and delete-encoded
+ * rows retain null maps because their user-data authorship is not
meaningful.
Review Comment:
That's fair, this scaladoc is overreaching. Delete-encoding runs should all
have null version maps when this is called, but that's not a contract of this
function.
Rephrased to represent that version maps are only initialized for
upsert-representing rows, and no further claims are made.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -967,6 +968,193 @@ 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. Existing maps are
preserved, and delete-encoded
+ * rows retain null maps because their user-data authorship is not
meaningful.
+ */
+ 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))
+ }
+
+ /**
+ * Replaces unauthored leaf values with values inherited from the nearest
preceding authoring
+ * row for the same key in `decomposedDf` (or, absent one, from the carry-in
anchor described
+ * below), and materializes version map entries for schema-evolved leaves
that would otherwise
+ * lose their unauthored signal after inheriting a non-null value.
+ *
+ * An existing non-null version map is the row's established authorship
record. Its semantic
Review Comment:
Simplified dramatically, the scaladoc shouldn't need to expand on all the
details of version map interaction.
The scaladoc now concisely explains;
1. Ignored nulls columns are coalesced by looking at the preceding authored
value
2. The treatment of schema evolved ignore null columns
3. What a row authored vs what is going to be reconciled on this pass can be
two different things (when the ignore-null selection is changing)
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -967,6 +968,193 @@ 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. Existing maps are
preserved, and delete-encoded
+ * rows retain null maps because their user-data authorship is not
meaningful.
+ */
+ 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))
+ }
+
+ /**
+ * Replaces unauthored leaf values with values inherited from the nearest
preceding authoring
+ * row for the same key in `decomposedDf` (or, absent one, from the carry-in
anchor described
+ * below), and materializes version map entries for schema-evolved leaves
that would otherwise
+ * lose their unauthored signal after inheriting a non-null value.
+ *
+ * An existing non-null version map is the row's established authorship
record. Its semantic
+ * contents are frozen across ignore-null selection changes: the active
selection determines
+ * which unauthored leaves are coalesced in this reconciliation, but does
not reinterpret the
+ * authorship of any leaf. Removing a leaf from the selection therefore
stops coalescing it,
+ * while adding it begins consulting the authorship already recorded for
that leaf.
+ *
+ * A null version map is the absence of an authorship record, rather than an
established record
+ * that says every leaf was authored. When ignore-null is enabled, an
upsert-representing row
+ * with a null map lazily establishes one from its stored values and the
active selection before
+ * coalescing. Once established, it follows the same frozen-authorship rule.
Schema evolution may
+ * later materialize an explicit unauthored entry, but does not change a
leaf's authorship.
+ *
+ * Must run after decomposition cleanup, whose canonical row shapes expose
the persisted
+ * interval gaps and delete boundaries that terminate inheritance, and
before start/end
+ * reconciliation so inherited tracked-history values determine the final
SCD2 runs.
+ *
+ * ==== Eventual consistency of the carry-in anchor ====
+ *
+ * A leaf's reconciliation in any pass is a function of two independent
inputs:
+ *
+ * 1. The active ignore-null selection. Only leaves in the active selection
are eligible for
+ * coalescing; an unauthored leaf outside it is not reconciled in this
pass.
+ * 2. The affected window of rows pulled in for reconciliation, which is
computed independently
+ * of the ignore-null selection.
+ *
+ * The first upsert-representing row in the affected window supplies its
stored value as
+ * carry-in, even when its version map records that value as unauthored,
because its
+ * predecessor is outside the window. If the leaf was temporarily dropped
from the selection,
+ * that stored value may not have been reconciled against the latest
authoring row.
+ *
+ * The stale carry-in is eventually consistent: a later reconciliation
replaces it, and all
Review Comment:
That's fair.
Related to the earlier comment, I think the scaladoc is introducing too many
design/implementation details - dropped most of these details and/or rely on
implementation comments instead.
The concept of eventual consistency here is a consequence of:
1. Only rows pulled in for reconciliation have their ignore-null columns
coalesced
2. We intentionally don't pull in all rows when ignore-null selection changes
When ignore-null selection changes, the first row pulled in for
reconciliation may have only previously been computed against the previous
ignore-null selection. Because its the first row, it has no preceding row to
possibly correct its inheritance from (respect the current selection) in this
reconciliation run. Yet it still contributes its own values to following rows
in this run.
The correction will _eventually_ happen if a future microbatch pulls in that
row along with preceding rows.
But that's not an explicit contract of this function anyway. That's a
behavioral consequence of calling this function multiples times with different
inputs (`decomposedDf` and `ignoreNullSelection`). I'm instead relying on tests
to document this behavior.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -967,6 +968,193 @@ 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. Existing maps are
preserved, and delete-encoded
+ * rows retain null maps because their user-data authorship is not
meaningful.
+ */
+ 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))
+ }
+
+ /**
+ * Replaces unauthored leaf values with values inherited from the nearest
preceding authoring
+ * row for the same key in `decomposedDf` (or, absent one, from the carry-in
anchor described
+ * below), and materializes version map entries for schema-evolved leaves
that would otherwise
+ * lose their unauthored signal after inheriting a non-null value.
+ *
+ * An existing non-null version map is the row's established authorship
record. Its semantic
+ * contents are frozen across ignore-null selection changes: the active
selection determines
+ * which unauthored leaves are coalesced in this reconciliation, but does
not reinterpret the
+ * authorship of any leaf. Removing a leaf from the selection therefore
stops coalescing it,
+ * while adding it begins consulting the authorship already recorded for
that leaf.
+ *
+ * A null version map is the absence of an authorship record, rather than an
established record
+ * that says every leaf was authored. When ignore-null is enabled, an
upsert-representing row
+ * with a null map lazily establishes one from its stored values and the
active selection before
+ * coalescing. Once established, it follows the same frozen-authorship rule.
Schema evolution may
+ * later materialize an explicit unauthored entry, but does not change a
leaf's authorship.
+ *
+ * Must run after decomposition cleanup, whose canonical row shapes expose
the persisted
+ * interval gaps and delete boundaries that terminate inheritance, and
before start/end
+ * reconciliation so inherited tracked-history values determine the final
SCD2 runs.
+ *
+ * ==== Eventual consistency of the carry-in anchor ====
+ *
+ * A leaf's reconciliation in any pass is a function of two independent
inputs:
+ *
+ * 1. The active ignore-null selection. Only leaves in the active selection
are eligible for
+ * coalescing; an unauthored leaf outside it is not reconciled in this
pass.
+ * 2. The affected window of rows pulled in for reconciliation, which is
computed independently
+ * of the ignore-null selection.
+ *
+ * The first upsert-representing row in the affected window supplies its
stored value as
+ * carry-in, even when its version map records that value as unauthored,
because its
+ * predecessor is outside the window. If the leaf was temporarily dropped
from the selection,
+ * that stored value may not have been reconciled against the latest
authoring row.
+ *
+ * The stale carry-in is eventually consistent: a later reconciliation
replaces it, and all
+ * downstream unauthored rows inherit the correct value, once that
reconciliation's affected
+ * window includes a preceding row that authored the leaf. That is the only
condition for
+ * correction, and nothing forces it to occur; a key that only receives
in-order events can
+ * retain the stale value indefinitely.
+ */
+ 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 while preserving the original column
order.
+ // 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.
+ val field = userDataColumnSchema(colName)
+ Scd2BatchProcessor.constructCoalescedIgnoreNullColumn(
+ Seq(colName),
+ field,
+ ignoreNullContextsByTopLevelColumn(colName)
+ ).as(colName, field.metadata)
+ case colName =>
+ // Pass through keys, framework columns, and user data outside the
+ // active selection.
+ F.col(QuotingUtils.quoteIdentifier(colName))
Review Comment:
Note that `decomposedDf.columns` always returns raw column names, even if
they contain special characters and were created with quotes. With that said,
there's 3 branches here to consider:
1. **Column is the CDC metadata system column**. By construction this does
not have any special characters, hence does not need to be escaped.
2. **Column is in ignore-null selection (and therefore has a context)**. We
group the ignore-null leaf contexts to their top-level struct exactly by the
top-level struct's name present in the dataframe, so we must also retrieve it
by exact name, no quoting. Note that `constructCoalescedIgnoreNullColumn` also
expects the input path to be unquoted, and applies quoting internally instead.
3. **Column is not in the ignore-null selection**. We pass these columns
through by resolving them against their current values in the dataframe. These
column names may require escaping if they contain special characters, and Spark
will handle resolving an escaped name against any valid equivalents in the
dataframe.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -967,6 +968,193 @@ 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. Existing maps are
preserved, and delete-encoded
+ * rows retain null maps because their user-data authorship is not
meaningful.
+ */
+ 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))
+ }
+
+ /**
+ * Replaces unauthored leaf values with values inherited from the nearest
preceding authoring
+ * row for the same key in `decomposedDf` (or, absent one, from the carry-in
anchor described
+ * below), and materializes version map entries for schema-evolved leaves
that would otherwise
+ * lose their unauthored signal after inheriting a non-null value.
+ *
+ * An existing non-null version map is the row's established authorship
record. Its semantic
+ * contents are frozen across ignore-null selection changes: the active
selection determines
+ * which unauthored leaves are coalesced in this reconciliation, but does
not reinterpret the
+ * authorship of any leaf. Removing a leaf from the selection therefore
stops coalescing it,
+ * while adding it begins consulting the authorship already recorded for
that leaf.
+ *
+ * A null version map is the absence of an authorship record, rather than an
established record
+ * that says every leaf was authored. When ignore-null is enabled, an
upsert-representing row
+ * with a null map lazily establishes one from its stored values and the
active selection before
+ * coalescing. Once established, it follows the same frozen-authorship rule.
Schema evolution may
+ * later materialize an explicit unauthored entry, but does not change a
leaf's authorship.
+ *
+ * Must run after decomposition cleanup, whose canonical row shapes expose
the persisted
+ * interval gaps and delete boundaries that terminate inheritance, and
before start/end
+ * reconciliation so inherited tracked-history values determine the final
SCD2 runs.
+ *
+ * ==== Eventual consistency of the carry-in anchor ====
+ *
+ * A leaf's reconciliation in any pass is a function of two independent
inputs:
+ *
+ * 1. The active ignore-null selection. Only leaves in the active selection
are eligible for
+ * coalescing; an unauthored leaf outside it is not reconciled in this
pass.
+ * 2. The affected window of rows pulled in for reconciliation, which is
computed independently
+ * of the ignore-null selection.
+ *
+ * The first upsert-representing row in the affected window supplies its
stored value as
+ * carry-in, even when its version map records that value as unauthored,
because its
+ * predecessor is outside the window. If the leaf was temporarily dropped
from the selection,
+ * that stored value may not have been reconciled against the latest
authoring row.
+ *
+ * The stale carry-in is eventually consistent: a later reconciliation
replaces it, and all
+ * downstream unauthored rows inherit the correct value, once that
reconciliation's affected
+ * window includes a preceding row that authored the leaf. That is the only
condition for
+ * correction, and nothing forces it to occur; a key that only receives
in-order events can
+ * retain the stale value indefinitely.
+ */
+ 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 while preserving the original column
order.
Review Comment:
Column order preservation is not a requirement as far as AutoCDC or this
function is concerned, its just a nice to have property that we get for free in
this implementation, and the comment documents it. Ultimately the unions and
merges against the target/aux table are done by name, not by column order.
Dropped the comment to avoid confusion.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -1828,3 +2123,181 @@ 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) {
+
+ /** Whether the affected window begins with an upsert-representing row. */
+ def isFirstUpsertRepresentingRow: Column =
+ isFirstRowInKeyWindow && isUpsertRepresentingRow
+}
+
+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
+ * [[inherits]], [[valueToInherit]], and [[needsSchemaEvolutionEntry]] 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 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.
+ * @param needsSchemaEvolutionEntry whether this row must gain an explicit
unauthored
+ * entry for the leaf to keep reading as
unauthored
+ * after it inherits.
+ */
+private[autocdc] case class LeafInheritanceContext(
Review Comment:
I made some meaningful predicate simplifications in
https://github.com/apache/spark/pull/58947/changes#r4181817679.
As mentioned in another comment, I also decoupled
`needsSchemaEvolutionEntry` from this class.
But I don't think there is a _fundamentally_ simpler approach. At a high
level, per key we are:
1. Iterating over all rows sorted order by effective sequence
2. For every ignore-null leaf, in every row
1. Applying the current value available for inheritance, if the leaf
didn't author it
2. Updating the current value available for inheritance
The complexity mainly arises from a couple edge cases that need to be
handled when updating the current value available for inheritance (first row
AKA "carry-in", and correctly terminating an inheritable value during
iteration), and expressing this as a Spark operation.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -1828,3 +2123,181 @@ 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) {
+
+ /** Whether the affected window begins with an upsert-representing row. */
+ def isFirstUpsertRepresentingRow: Column =
+ isFirstRowInKeyWindow && isUpsertRepresentingRow
+}
+
+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
+ * [[inherits]], [[valueToInherit]], and [[needsSchemaEvolutionEntry]] 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 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.
+ * @param needsSchemaEvolutionEntry whether this row must gain an explicit
unauthored
+ * entry for the leaf to keep reading as
unauthored
+ * after it inherits.
+ */
+private[autocdc] case class LeafInheritanceContext(
+ path: Seq[String],
+ index: Int,
+ valueToInheritIfAny: Column,
+ inherits: Column,
+ needsSchemaEvolutionEntry: 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)
+}
+
+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 leafValueInRow = F.col(QuotingUtils.quoteNameParts(path))
+ val isLeafAuthoredByRow =
+ Scd2VersionMap.isAuthored(versionMap, leafValueInRow, path)
+
+ // 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. An authored
+ // leaf emits its value. If the first row in the affected window is an
upsert, it emits its
+ // stored value even when unauthored because it is the carry-in for
omitted earlier history.
+ val rowResetsInheritanceChain =
rowInheritanceContext.rowEndsInheritanceChain
+ val rowContributesAuthoredValue =
+ rowInheritanceContext.isUpsertRepresentingRow && isLeafAuthoredByRow
+
+ // When the first row is an upsert, it supplies the baseline inheritance
value for every leaf.
+ // Even if it did not author a leaf itself, its stored value may have been
coalesced from rows
+ // preceding the affected window.
+ //
+ // If this first row is an existing upsert, it cannot itself be
re-coalesced in this sweep:
+ // its predecessor is outside the affected window, so its stored value is
the only safe
+ // carry-in. With an unchanged selection that value is already correct.
After a selection
+ // change, the anchor can only be corrected by a reconciliation that
includes preceding
+ // history. See the "Eventual consistency of the carry-in anchor" section
in the
+ // [[coalesceIgnoredNulls]] scaladoc.
+ val rowEstablishesCarryIn =
rowInheritanceContext.isFirstUpsertRepresentingRow
+ val rowUpdatesInheritanceChain =
+ rowResetsInheritanceChain || rowContributesAuthoredValue ||
rowEstablishesCarryIn
+
+ // 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,
leafValueInRow)
+ F.struct(newInheritanceChainValue.as(inheritanceValueFieldName))
+ }
+ )
+
+ val valueToInheritIfAnyColName = valueToInheritIfAnyColumnName(index)
+ val valueToInheritIfAnyColRef = F.col(valueToInheritIfAnyColName)
+ val precedingRowsContributeInheritableValue =
valueToInheritIfAnyColRef.isNotNull
+
+ val valueToInherit =
+ valueToInheritIfAnyColRef.getField(inheritanceValueFieldName)
+ // Skip null over null: rebuilding a nested leaf could otherwise turn a
null parent struct
+ // into a non-null struct containing nulls.
+ val wouldInheritNullOverNull = leafValueInRow.isNull &&
valueToInherit.isNull
+
+ val rowInheritsLeaf =
Review Comment:
That's fair. I do think the condition reads at least somewhat naturally, but
agree it's long. It also does encode some redundancy between subexpressions.
In particular `precedingRowsContributeInheritableValue` is used to identify
whether this is the first row for the key, and therefore its stored values must
be treated as authoritative (there's nothing for it to inherit, and it seeds
inheritance for all following rows).
That's really the same treatment as an authored leaf, and it's currently
being deduced in a roundabout way - we instead know that the only upsert
representing row where it's not true is the first one.
Dropped `precedingRowsContributeInheritableValue` altogether, and extended
`isLeafAuthoredByRow` to naturally handle the first-row case. Also renamed it
to `storedValueIsAuthoritative` for clarity.
I also realized that since a null struct and struct of all null leaves are
semantically equivalent for an ignore-null column, we don't need to preserve
null structs exactly as present in the event with the
`wouldInheritNullOverNull` predicate - dropped that.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -1529,6 +1717,92 @@ 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 the version map entry only for eligible schema evolved leaves,
null for all other
+ // leaves.
+ val newEntryPerLeafOrNull = leafInheritanceContexts.map { context =>
+ F.when(
+ context.needsSchemaEvolutionEntry,
+ Scd2VersionMap.buildVersionMapEntry(context.path, authored = false))
+ }
+
+ // Filter out the null entries, which explicitly represent leaves that
don't need to gain a
+ // version map entry due to schema evolution.
Review Comment:
The `LeafInheritanceContext` is doing two things, which I think might be
leading to confusion here.
For all ignore-null leaves, it computes per row:
1. Whether the leaf should inherit, and what value,
2. Whether that leaf should gain an authorship entry in the version map when
it is schema evolved
It's entirely possible an ignore-null leaf does not require materializing a
version map entry due to schema evolution, while still needing to inherit a
value. That's what this case is covering.
I refactored `needsSchemaEvolutionEntry` out of the `LeafInheritanceContext`
to make this clearer.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -1529,6 +1717,92 @@ 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 the version map entry only for eligible schema evolved leaves,
null for all other
+ // leaves.
+ val newEntryPerLeafOrNull = leafInheritanceContexts.map { context =>
+ F.when(
+ context.needsSchemaEvolutionEntry,
+ Scd2VersionMap.buildVersionMapEntry(context.path, authored = false))
+ }
+
+ // Filter out the null entries, which explicitly represent leaves that
don't need to gain a
+ // version map entry due to schema evolution.
+ 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))
+ )
+ }
+
+ /**
+ * Rebuilds the column at `path` with coalesced leaf values. For struct
types the function
+ * recurses into each field, reassembling the struct from its coalesced
children. For leaf
+ * types it substitutes the inherited value when the leaf's `inherits`
predicate holds.
+ *
+ * The supplied contexts must be nonempty and cover `path`. Struct fields
without a context are
+ * passed through unchanged.
+ *
+ * A non-nullable field raises `NOT_NULL_ASSERT_VIOLATION` when coalescing
cannot supply it a
+ * non-null value: for example, when no preceding row authored it, when a
delete boundary reset
+ * its inheritance, or when its null parent struct is rebuilt because a
sibling leaf inherits.
Review Comment:
If `x` already has a non-null value, then there's no problem. But imagine
you have some column `col` holding a struct with schema
```
{
x: INT NOT NULL
}
```
And originally in the target table, you have a row with `col=null`. This is
valid.
Then lets say the schema for `col`'s struct evolves to now have `y`, and
simulataneously inherits a non-null value, for that same row already in the
target table.
Then we need to materialize `col.y`, and therefore are forced to also
materialize `col.x`, but with a null value.
It's a niche edge case that we get to cover for free, but I rephrased the
sentence to hopefully make it clearer.
There's an exact test that documents the behavior in
`Scd2CoalesceIgnoredNullsSuite`: `inheriting into a stored null struct fails
when a non-nullable field is authored null`.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -967,6 +968,193 @@ 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. Existing maps are
preserved, and delete-encoded
+ * rows retain null maps because their user-data authorship is not
meaningful.
+ */
+ 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))
+ }
+
+ /**
+ * Replaces unauthored leaf values with values inherited from the nearest
preceding authoring
+ * row for the same key in `decomposedDf` (or, absent one, from the carry-in
anchor described
+ * below), and materializes version map entries for schema-evolved leaves
that would otherwise
+ * lose their unauthored signal after inheriting a non-null value.
+ *
+ * An existing non-null version map is the row's established authorship
record. Its semantic
+ * contents are frozen across ignore-null selection changes: the active
selection determines
+ * which unauthored leaves are coalesced in this reconciliation, but does
not reinterpret the
+ * authorship of any leaf. Removing a leaf from the selection therefore
stops coalescing it,
+ * while adding it begins consulting the authorship already recorded for
that leaf.
+ *
+ * A null version map is the absence of an authorship record, rather than an
established record
+ * that says every leaf was authored. When ignore-null is enabled, an
upsert-representing row
+ * with a null map lazily establishes one from its stored values and the
active selection before
+ * coalescing. Once established, it follows the same frozen-authorship rule.
Schema evolution may
+ * later materialize an explicit unauthored entry, but does not change a
leaf's authorship.
+ *
+ * Must run after decomposition cleanup, whose canonical row shapes expose
the persisted
Review Comment:
Rephrased to be more precise. Scaladoc now concretely lists the notable
row-types decomposition produces that are bearing on value inheritance, in a
single sentence.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
