jose-torres commented on code in PR #58947:
URL: https://github.com/apache/spark/pull/58947#discussion_r4181665327
##########
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:
This last part doesn't seem to match the behavior, delete-encoded rows
retain existing maps regardless of whether they're null. (I'm also a bit
worried the inconsistent second word between "upsert-representing" and
"delete-encoded" may be hiding a secret third option.)
##########
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:
I am not convinced this sentence is coherent, although the sequencing
requirement is correct AFAICT.
##########
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:
I feel like design requirements are leaking into the implementation here,
it's hard to understand what about this specific component is related to this
semantic.
##########
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:
Why do they have a LeafInheritanceContext at all then?
##########
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:
Why do we have to quote it here but not in the other branches?
##########
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:
This last condition doesn't sound right to me. If I have struct(x NOT
NULL=1, y=null), and x's sibling y inherits 2 or something, why would that
cause x to no longer have a non-null value?
##########
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:
This seems _much_ more complex than any of its usages require, is there
really no way to simplify it?
##########
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:
This in particular is an architecture smell to me. This boolean just has so
many clauses tucked inside it.
##########
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:
Hmm, I'm a bit worried about the idea that the column order is something
which must be preserved. Ideally it should be possible to change the schema
order of the input table, table definition, etc. without breaking the pipeline,
does that work?
##########
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:
This is the first reference to a carry-in being "stale", and I'm not sure
what that means nor what eventual consistency refers to in this context.
--
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]
