jose-torres commented on code in PR #56311:
URL: https://github.com/apache/spark/pull/56311#discussion_r3530406117
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -547,3 +879,82 @@ object Scd2BatchProcessor {
F.struct(cdcMetadataFieldsInOrder.toImmutableArraySeq: _*)
}
}
+
+/**
+ * The three columns that locate a row on the SCD2 timeline: its source
record-start sequence
+ * (`recordStartAt`, null only for decomposition tails) and the bounds of its
visible interval
+ * [`startAt`, `endAt`). [[RowClassifier]] classifies a row purely from this
triple.
+ */
+private[autocdc] case class Scd2IntervalColumns(
+ recordStartAt: Column,
+ startAt: Column,
+ endAt: Column) {
+
+ /**
+ * The row's effective ordering position. Decomposition tails carry no
`recordStartAt` and
+ * fall back to their closing sequence (`endAt`), the same convention used by
+ * [[Scd2BatchProcessor.orderChronologicallyPerKeyWindow]].
+ */
+ def effectiveRecordStartAt: Column = F.coalesce(recordStartAt, endAt)
+}
+
+object RowClassifier {
+
+ /**
+ * Synthetic right boundary created by splitting a closed row, temporarily
present during
+ * microbatch reconciliation but never materializes in the target or aux
tables.
+ */
+ private[autocdc] def isDecompositionTail(row: Scd2IntervalColumns): Column =
Review Comment:
OK, this makes sense to me.
--
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]
