anew commented on code in PR #57192:
URL: https://github.com/apache/spark/pull/57192#discussion_r3599240609
##########
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:
Good catch @szehon-ho — you're right, in a pure open no-op run every row
including the tail has endAt = null, so my earlier "tail keeps a non-null
endAt" was wrong. Corrected the comment (2ea22dbd36c): clearing endAt here only
erases the collapsed interior row's interval and is not itself the aux-routing
signal (the tail can also be open). Target-vs-aux routing lands in follow-up
SPARK-57378, keyed off the reconciled run shape.
--
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]
