jose-torres commented on code in PR #57192:
URL: https://github.com/apache/spark/pull/57192#discussion_r3591965146
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala:
##########
@@ -153,21 +153,28 @@ object ScdType {
/**
* Configuration for an AutoCDC flow.
*
- * @param keys The column(s) that uniquely identify a row in the
source data.
- * @param sequencing Expression ordering CDC events to correctly resolve
out-of-order
- * arrivals. Must be a sortable type.
- * @param deleteCondition Expression that marks a source row as a DELETE. When
None, all
- * rows are treated as upserts.
- * @param storedAsScdType The SCD strategy these args should be applied to.
- * @param columnSelection Which source columns to select in the target table.
None means
- * all columns.
+ * @param keys The column(s) that uniquely identify a row in
the source data.
+ * @param sequencing Expression ordering CDC events to correctly
resolve out-of-order
+ * arrivals. Must be a sortable type.
+ * @param deleteCondition Expression that marks a source row as a
DELETE. When None, all
+ * rows are treated as upserts.
+ * @param storedAsScdType The SCD strategy these args should be applied
to.
+ * @param columnSelection Which source columns to select in the target
table. None means
+ * all columns.
+ * @param trackHistorySelection SCD2 only. Selects the selected user-data
columns whose values
+ * define a run: two consecutive upsert events
for the same key are
+ * coalesced into the same run iff they agree on
every selected
+ * tracking column. None means every eligible
selected user column
+ * (i.e. every selected source column that is
neither a key nor a
+ * framework column) is considered tracked.
Ignored under SCD1.
Review Comment:
Values other than None should fail SCD1 rather than being ignored.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -639,6 +639,193 @@ case class Scd2BatchProcessor(
.filter(!isRedundantAtSameEffectiveSequence)
.drop(Scd2BatchProcessor.nextEffectiveRecordStartAtColName)
}
+
+ /**
+ * Recompute every row's [[startAtColName]] and [[endAtColName]] over the
per-key chronological
+ * window so the dataframe reflects the canonical SCD2 timeline that the
downstream aux- and
+ * target-table merges consume.
+ *
+ * Decomposition tails and tombstones round-trip unchanged. An open upsert
may close at its
+ * successor's effective sequence (becoming closed); a closed upsert may
have its endAt cleared
+ * when absorbed into a run as a no-op continuation (becoming open).
[[recordStartAtFieldName]]
+ * is never modified.
+ *
+ * @param decomposedAndCleanedDf
+ * the output of [[dropRedundantRowsPostDecomposition]]: a dataframe
conforming to the
+ * canonical SCD2 row schema `[user_cols..., [[startAtColName]],
[[endAtColName]],
+ * [[cdcMetadataColName]]]` where every row is in one of the four
canonical post-
+ * decomposition shapes (decomposition tail, tombstone, open upsert,
closed upsert).
+ * If a row is a closed upsert in the input, it is assumed to not be
bisected by any other
+ * row in the input.
+ * @return
+ * a dataframe with the same schema and row count as the input, with each
row's
+ * [[startAtColName]] / [[endAtColName]] replaced by their reconciled
values.
+ */
+ private[autocdc] def reconcileStartAndEndAt(
+ decomposedAndCleanedDf: DataFrame): DataFrame = {
+ val trackedHistoryColumns =
computeTrackedHistoryColumns(decomposedAndCleanedDf)
+
+ val recordStartAtField =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAtCol = F.col(Scd2BatchProcessor.startAtColName)
+ val endAtCol = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Decomposition tails carry no recordStartAt of their own, so they take
the closing
+ // sequence (`endAt`) as their effective ordering position - the same
convention used by
+ // [[orderChronologicallyPerKeyWindow]] and
[[dropRedundantRowsPostDecomposition]].
+ val current = Scd2IntervalColumns(recordStartAtField, startAtCol, endAtCol)
+ val previous = current.lagBy(1, orderChronologicallyPerKeyWindow)
+ val next = current.leadBy(1, orderChronologicallyPerKeyWindow)
+
+ // A row is the last in its per-key window when `LEAD(1)` has no
successor; a constant
+ // literal is sufficient since we only care whether one exists.
+ val isLastRowInKeyWindow =
+ F.lead(F.lit(true), 1).over(orderChronologicallyPerKeyWindow).isNull
+
+ // The current row's tracked-history equality is computed against both its
predecessor and
Review Comment:
Hmmmm. I wonder whether this does the right thing when there's a sequence of
three rows that all need to collapse into each other, this seems potentially
error-prone. Please make sure there's test coverage
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -942,18 +1155,50 @@ object RowClassifier {
isOpenUpsert(row) || isClosedUpsert(row)
/**
- * Delete-representing row, encoded as an instantaneous (zero-width)
interval at `recordStartAt`
- * (`startAt == endAt == recordStartAt`). Never materializes in the target
table.
+ * Tombstone (delete-boundary) row, encoded as an instantaneous interval at
Review Comment:
I guess this comment does say all deletes are represented as instantaneous
deletes, even in the before? Maybe that was all correct then? Please do confirm
whether this is accurate.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -639,6 +639,193 @@ case class Scd2BatchProcessor(
.filter(!isRedundantAtSameEffectiveSequence)
.drop(Scd2BatchProcessor.nextEffectiveRecordStartAtColName)
}
+
+ /**
+ * Recompute every row's [[startAtColName]] and [[endAtColName]] over the
per-key chronological
+ * window so the dataframe reflects the canonical SCD2 timeline that the
downstream aux- and
+ * target-table merges consume.
+ *
+ * Decomposition tails and tombstones round-trip unchanged. An open upsert
may close at its
+ * successor's effective sequence (becoming closed); a closed upsert may
have its endAt cleared
Review Comment:
But this is not the _only_ case in which an open upsert may close, it may
close without a successor when a delete comes in. (Unless we're representing
all deletes as instantaneous deletes? That could explain a lot of the above but
I think is probably wrong.)
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -556,7 +556,7 @@ case class Scd2BatchProcessor(
val row = Scd2IntervalColumns(recordStartAtField, startAtCol, endAtCol)
val isWellFormedRow =
RowClassifier.isDecompositionTail(row) ||
- RowClassifier.isDeleteRepresentingRow(row) ||
+ RowClassifier.isTombstone(row) ||
Review Comment:
Hmm, guess it's not just in the comments either. Writing out an explicit
marker here so I remember to refer to it when I get to the classifier
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala:
##########
@@ -153,21 +153,28 @@ object ScdType {
/**
* Configuration for an AutoCDC flow.
*
- * @param keys The column(s) that uniquely identify a row in the
source data.
- * @param sequencing Expression ordering CDC events to correctly resolve
out-of-order
- * arrivals. Must be a sortable type.
- * @param deleteCondition Expression that marks a source row as a DELETE. When
None, all
- * rows are treated as upserts.
- * @param storedAsScdType The SCD strategy these args should be applied to.
- * @param columnSelection Which source columns to select in the target table.
None means
- * all columns.
+ * @param keys The column(s) that uniquely identify a row in
the source data.
+ * @param sequencing Expression ordering CDC events to correctly
resolve out-of-order
+ * arrivals. Must be a sortable type.
+ * @param deleteCondition Expression that marks a source row as a
DELETE. When None, all
+ * rows are treated as upserts.
+ * @param storedAsScdType The SCD strategy these args should be applied
to.
+ * @param columnSelection Which source columns to select in the target
table. None means
+ * all columns.
+ * @param trackHistorySelection SCD2 only. Selects the selected user-data
columns whose values
+ * define a run: two consecutive upsert events
for the same key are
Review Comment:
I think we never actually write down what a "run" is, we should make sure
this is included somewhere in the code for context.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -439,9 +439,9 @@ case class Scd2BatchProcessor(
* a dataframe with the same schema as the input. Every closed
non-tombstone row that
* was bisected has been replaced by its head + tail pair; every other row
is carried
* through as-is. Each output row can be classified as one of:
{decomposition head,
- * decomposition tail, instantaneous delete, open upsert,
closed-and-unbisected row}. It's
- * possible that some of the returned decomposition tails are logically
redundant, as
- * deletion markers that are immediately overtaken by a succeeding row.
+ * decomposition tail, tombstone, open upsert, closed-and-unbisected row}.
It's possible
Review Comment:
This change isn't right. SCD2 doesn't have tombstones as such, a deleted row
in the change feed is represented by a closed history record with nothing
further.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -639,6 +639,193 @@ case class Scd2BatchProcessor(
.filter(!isRedundantAtSameEffectiveSequence)
.drop(Scd2BatchProcessor.nextEffectiveRecordStartAtColName)
}
+
+ /**
+ * Recompute every row's [[startAtColName]] and [[endAtColName]] over the
per-key chronological
+ * window so the dataframe reflects the canonical SCD2 timeline that the
downstream aux- and
+ * target-table merges consume.
+ *
+ * Decomposition tails and tombstones round-trip unchanged. An open upsert
may close at its
+ * successor's effective sequence (becoming closed); a closed upsert may
have its endAt cleared
+ * when absorbed into a run as a no-op continuation (becoming open).
[[recordStartAtFieldName]]
+ * is never modified.
+ *
+ * @param decomposedAndCleanedDf
+ * the output of [[dropRedundantRowsPostDecomposition]]: a dataframe
conforming to the
+ * canonical SCD2 row schema `[user_cols..., [[startAtColName]],
[[endAtColName]],
+ * [[cdcMetadataColName]]]` where every row is in one of the four
canonical post-
+ * decomposition shapes (decomposition tail, tombstone, open upsert,
closed upsert).
+ * If a row is a closed upsert in the input, it is assumed to not be
bisected by any other
+ * row in the input.
+ * @return
+ * a dataframe with the same schema and row count as the input, with each
row's
+ * [[startAtColName]] / [[endAtColName]] replaced by their reconciled
values.
+ */
+ private[autocdc] def reconcileStartAndEndAt(
+ decomposedAndCleanedDf: DataFrame): DataFrame = {
+ val trackedHistoryColumns =
computeTrackedHistoryColumns(decomposedAndCleanedDf)
+
+ val recordStartAtField =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAtCol = F.col(Scd2BatchProcessor.startAtColName)
+ val endAtCol = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Decomposition tails carry no recordStartAt of their own, so they take
the closing
+ // sequence (`endAt`) as their effective ordering position - the same
convention used by
+ // [[orderChronologicallyPerKeyWindow]] and
[[dropRedundantRowsPostDecomposition]].
+ val current = Scd2IntervalColumns(recordStartAtField, startAtCol, endAtCol)
+ val previous = current.lagBy(1, orderChronologicallyPerKeyWindow)
+ val next = current.leadBy(1, orderChronologicallyPerKeyWindow)
+
+ // A row is the last in its per-key window when `LEAD(1)` has no
successor; a constant
+ // literal is sufficient since we only care whether one exists.
+ val isLastRowInKeyWindow =
+ F.lead(F.lit(true), 1).over(orderChronologicallyPerKeyWindow).isNull
+
+ // The current row's tracked-history equality is computed against both its
predecessor and
+ // its successor so the same window scan can decide both run-head start
(LAG-side) and no-op
+ // continuation closure (LEAD-side) without an extra pass. The comparison
is null-safe
+ // (`<=>`), so two rows with matching null values in the same tracked
column register as
+ // equal. An empty tracked-history column set collapses to a constant
`true`, which makes
+ // every consecutive upsert pair a no-op continuation - the correct
degenerate behavior when
+ // the user tracks nothing.
+ val areTrackedColumnsEqualInPreviousRow = trackedHistoryColumns
+ .map { c =>
+ val col = F.col(QuotingUtils.quoteIdentifier(c))
+ col <=> F.lag(col, 1).over(orderChronologicallyPerKeyWindow)
+ }
+ .reduceOption(_ && _)
+ .getOrElse(F.lit(true))
+
+ val areTrackedColumnsEqualInNextRow = trackedHistoryColumns
+ .map { c =>
+ val col = F.col(QuotingUtils.quoteIdentifier(c))
+ col <=> F.lead(col, 1).over(orderChronologicallyPerKeyWindow)
+ }
+ .reduceOption(_ && _)
+ .getOrElse(F.lit(true))
+
+ // Reconciliation of start/end at is dependent on the class of row being
reconciled. Build
+ // row classification predicates.
+ val isDecompositionTail = RowClassifier.isDecompositionTail(current)
+ val isUpsertRepresentingRow =
RowClassifier.isUpsertRepresentingRow(current)
+
+ // From the previous row's perspective, the current row is its successor.
+ val previousIsNoOpUpsertWithCurrent =
+ RowClassifier.isNoOpUpsertContinuation(
+ row = previous,
+ next = current,
+ areTrackedColumnsEqualInNextRow = areTrackedColumnsEqualInPreviousRow
+ )
+
+ // "Window-local run head" means the current row begins a new run within
the affected
+ // window. The first row in the window is automatically considered
local-run-head since
+ // there's no predecessor to coalesce with. A non-first row is a local run
head iff its
+ // predecessor is not a no-op continuation that absorbs it.
+ val isWindowLocalUpsertRunHead =
+ isUpsertRepresentingRow && !previousIsNoOpUpsertWithCurrent
+ val isFirstRowInKeyWindow = previous.effectiveRecordStartAt.isNull
+ val runHeadStartAt =
+ F.when(
+ isWindowLocalUpsertRunHead,
+ // The first row in the window may be a window-local run head but not
a global run
+ // head (e.g., an aux anchor row pulled in for left context). In that
case, `startAt`
+ // may be strictly less than `recordStartAt`, encoding the true global
run start, and
+ // we propagate it forward to later in-window continuations of the
same run.
+ // For every later window-local upsert run head, `recordStartAt` is
the run start.
+ F.when(isFirstRowInKeyWindow, startAtCol).otherwise(recordStartAtField)
+ )
+
+ // Propagate the run head's `startAt` forward to every row in the run via
a running
+ // `last(...)` over `[unboundedPreceding, currentRow]`. `runHeadStartAt`
is non-null
+ // only on run heads, and `ignoreNulls = true` makes intermediate rows
inherit the most
+ // recent head's value.
+ val runStartAt =
+ F.last(runHeadStartAt, ignoreNulls = true).over(
+ orderChronologicallyPerKeyWindow.rowsBetween(
+ Window.unboundedPreceding,
+ Window.currentRow
+ )
+ )
+
+ val currentIsNoOpUpsertWithNext =
+ RowClassifier.isNoOpUpsertContinuation(
+ row = current,
+ next = next,
+ areTrackedColumnsEqualInNextRow = areTrackedColumnsEqualInNextRow
+ )
+
+ val finalStartAt =
+ F.when(isDecompositionTail, F.lit(null).cast(resolvedSequencingType))
+ .when(isUpsertRepresentingRow, runStartAt)
+ .otherwise(startAtCol)
+
+ val finalEndAt =
+ F.when(isDecompositionTail, endAtCol)
+ .when(isLastRowInKeyWindow, endAtCol)
+ // A no-op continuation collapses into its run head, so the row's
visible interval
+ // disappears and `endAt` is reset to null to route the row to the aux
table.
Review Comment:
I don't understand this last part, why does endAt = null cause the row to be
routed to the aux table? In the test coverage we end up with some cases where
all rows for a particular version range end up with endAt = null, if they all
go to the aux table there'll be nothing left to represent the run to the user.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -639,6 +639,193 @@ case class Scd2BatchProcessor(
.filter(!isRedundantAtSameEffectiveSequence)
.drop(Scd2BatchProcessor.nextEffectiveRecordStartAtColName)
}
+
+ /**
+ * Recompute every row's [[startAtColName]] and [[endAtColName]] over the
per-key chronological
+ * window so the dataframe reflects the canonical SCD2 timeline that the
downstream aux- and
+ * target-table merges consume.
+ *
+ * Decomposition tails and tombstones round-trip unchanged. An open upsert
may close at its
+ * successor's effective sequence (becoming closed); a closed upsert may
have its endAt cleared
+ * when absorbed into a run as a no-op continuation (becoming open).
[[recordStartAtFieldName]]
+ * is never modified.
+ *
+ * @param decomposedAndCleanedDf
+ * the output of [[dropRedundantRowsPostDecomposition]]: a dataframe
conforming to the
+ * canonical SCD2 row schema `[user_cols..., [[startAtColName]],
[[endAtColName]],
+ * [[cdcMetadataColName]]]` where every row is in one of the four
canonical post-
+ * decomposition shapes (decomposition tail, tombstone, open upsert,
closed upsert).
+ * If a row is a closed upsert in the input, it is assumed to not be
bisected by any other
+ * row in the input.
+ * @return
+ * a dataframe with the same schema and row count as the input, with each
row's
+ * [[startAtColName]] / [[endAtColName]] replaced by their reconciled
values.
+ */
+ private[autocdc] def reconcileStartAndEndAt(
+ decomposedAndCleanedDf: DataFrame): DataFrame = {
+ val trackedHistoryColumns =
computeTrackedHistoryColumns(decomposedAndCleanedDf)
+
+ val recordStartAtField =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAtCol = F.col(Scd2BatchProcessor.startAtColName)
+ val endAtCol = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Decomposition tails carry no recordStartAt of their own, so they take
the closing
+ // sequence (`endAt`) as their effective ordering position - the same
convention used by
+ // [[orderChronologicallyPerKeyWindow]] and
[[dropRedundantRowsPostDecomposition]].
+ val current = Scd2IntervalColumns(recordStartAtField, startAtCol, endAtCol)
+ val previous = current.lagBy(1, orderChronologicallyPerKeyWindow)
+ val next = current.leadBy(1, orderChronologicallyPerKeyWindow)
+
+ // A row is the last in its per-key window when `LEAD(1)` has no
successor; a constant
+ // literal is sufficient since we only care whether one exists.
+ val isLastRowInKeyWindow =
+ F.lead(F.lit(true), 1).over(orderChronologicallyPerKeyWindow).isNull
+
+ // The current row's tracked-history equality is computed against both its
predecessor and
+ // its successor so the same window scan can decide both run-head start
(LAG-side) and no-op
+ // continuation closure (LEAD-side) without an extra pass. The comparison
is null-safe
+ // (`<=>`), so two rows with matching null values in the same tracked
column register as
+ // equal. An empty tracked-history column set collapses to a constant
`true`, which makes
+ // every consecutive upsert pair a no-op continuation - the correct
degenerate behavior when
+ // the user tracks nothing.
+ val areTrackedColumnsEqualInPreviousRow = trackedHistoryColumns
+ .map { c =>
+ val col = F.col(QuotingUtils.quoteIdentifier(c))
+ col <=> F.lag(col, 1).over(orderChronologicallyPerKeyWindow)
+ }
+ .reduceOption(_ && _)
+ .getOrElse(F.lit(true))
+
+ val areTrackedColumnsEqualInNextRow = trackedHistoryColumns
+ .map { c =>
+ val col = F.col(QuotingUtils.quoteIdentifier(c))
+ col <=> F.lead(col, 1).over(orderChronologicallyPerKeyWindow)
+ }
+ .reduceOption(_ && _)
+ .getOrElse(F.lit(true))
+
+ // Reconciliation of start/end at is dependent on the class of row being
reconciled. Build
+ // row classification predicates.
+ val isDecompositionTail = RowClassifier.isDecompositionTail(current)
+ val isUpsertRepresentingRow =
RowClassifier.isUpsertRepresentingRow(current)
+
+ // From the previous row's perspective, the current row is its successor.
+ val previousIsNoOpUpsertWithCurrent =
+ RowClassifier.isNoOpUpsertContinuation(
+ row = previous,
+ next = current,
+ areTrackedColumnsEqualInNextRow = areTrackedColumnsEqualInPreviousRow
Review Comment:
Extremely skeptical of this line.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -639,6 +639,193 @@ case class Scd2BatchProcessor(
.filter(!isRedundantAtSameEffectiveSequence)
.drop(Scd2BatchProcessor.nextEffectiveRecordStartAtColName)
}
+
+ /**
+ * Recompute every row's [[startAtColName]] and [[endAtColName]] over the
per-key chronological
+ * window so the dataframe reflects the canonical SCD2 timeline that the
downstream aux- and
+ * target-table merges consume.
+ *
+ * Decomposition tails and tombstones round-trip unchanged. An open upsert
may close at its
Review Comment:
I think it's still right to say that decomposition tails and instantaneous
deletes round-trip unchanged.
--
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]
