szehon-ho commented on code in PR #57192:
URL: https://github.com/apache/spark/pull/57192#discussion_r3598982361
##########
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:
Ping — still named `areTrackedColumnsEqualInNextRow` on `a791172`, including
the call site that passes `areTrackedColumnsEqualInPreviousRow`. Renaming to
`areTrackedColumnsEqual` would remove the LEAD/LAG confusion.
--
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]
