anew commented on code in PR #57192:
URL: https://github.com/apache/spark/pull/57192#discussion_r3599245673
##########
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:
Added: a three-row bisection (head@5 alice + event@15 bob + tail end=30)
asserting the head closes at 15, not 30. The bisecting event is
tracked-different so the head actually closes there.
--
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]
