[SPARK-57322][SDP] Implement SCD2 Batch Processor; Reconcile StartAt/EndAt#57192
[SPARK-57322][SDP] Implement SCD2 Batch Processor; Reconcile StartAt/EndAt#57192anew wants to merge 3 commits into
Conversation
…d-history run coalescing Add reconcileStartAndEndAt to Scd2BatchProcessor, which recomputes every row's startAt/endAt over the per-key chronological window so the dataframe reflects the canonical SCD2 timeline consumed by the aux- and target-table merges. Introduce ChangeArgs.trackHistorySelection to control which user-data columns define a run: consecutive upserts for the same key coalesce into one run iff they agree on every tracked column. Supporting changes: rename RowClassifier.isDeleteRepresentingRow to isTombstone for consistency, add RowClassifier.rowClosesStrictlyBeforeNextRow and isNoOpUpsertContinuation, add Scd2IntervalColumns.lagBy/leadBy window helpers, and add computeTrackedHistoryColumns. Includes reconcileStartAndEndAt tests. Co-authored-by: Isaac
Tracked history columns are resolved after columnSelection because SCD2 intervals only encode changes visible in the target table. Columns excluded from the target cannot produce observable target history, so they are not valid tracking columns. Add tests covering the default behavior and the error raised when trackHistorySelection explicitly includes an excluded column.
anew
left a comment
anew
left a comment
There was a problem hiding this comment.
One concern:
trackHistorySelection is applied after columnSelection, so dropped columns cannot define history. preprocessMicrobatch ends by applying projectTargetColumnsOntoMicrobatch, which removes columns excluded from the target. The new reconcileStartAndEndAt step then calls computeTrackedHistoryColumns on that already-projected dataframe, so trackHistorySelection only sees retained target columns. That conflicts with the new ChangeArgs docs, which say None tracks every eligible source column and that trackHistorySelection selects the user-data columns defining a run. Example: if status is excluded from the target but should still be tracked, IncludeColumns(status) will fail as not found; with None, a status change is silently ignored and two events may be coalesced into one run.
Relevant code:
- sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:133, sql/
- pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:196, sql/pipelines/src/main/scala/org/apache/
spark/sql/pipelines/autocdc/ChangeArgs.scala:162.
Does it actually make sense to track columns that are not in the column selection?
- Usually, no:
- If columnSelection defines the user-visible target row values, then tracking a column that is not persisted in the target is hard to justify: the target would split SCD2 history into separate intervals even though every selected column value is identical across those intervals. From a consumer’s perspective that looks like redundant history.
So the cleaner contract is probably:
- trackHistorySelection is applied to the post-columnSelection user-data schema, excluding keys and framework columns.
|
Tracked history columns are resolved after columnSelection because SCD2 intervals only encode changes visible in the target table. Columns excluded from the target cannot produce observable target history, so they are not valid tracking columns. Added tests covering the default behavior and the error raised when trackHistorySelection explicitly includes an excluded column. |
1 participant
Approved AutoCDC SPIP: https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7
This is a stacked PR. Review incremental diff here: AnishMahto/spark@SPARK-57222-SCD2-decompose-affected-rows...SPARK-57322-reconcile-start-end-at
What changes were proposed in this pull request?
Preamble:
The SCD type 2 flow is a foreachBatch streaming query on an input change-data-feed, and is responsible for reconciling the incoming change data onto some target table that follows SCD2 replication semantics.
SCD2 flows also maintain an "auxiliary" table to keep track of early-arriving out-of-order received events state. Each microbatch will need to reconcile against this auxiliary table as well, and update the auxiliary table's state appropriately for future microbatches.
Reconcile StartAt/EndAt:
Once we have all of the relevant rows for microbatch SCD2 reconciliation decomposed, we can begin setting the start/end-at for incoming microbatch rows, and rewriting existing start/end-at of existing rows in the aux/target tables to account for late arriving events.
When iterating on the affected microbatch rows in chronologically sorted order after decomposition, we can reconcile a row's end-at by looking at its next neighbor, which naturally succeeds the row.
Similarly we may need to rewrite both existing and new upsert-representing rows' start-at depending on whether the upsert row is a now a run-head or not. This can be done by looking backwards to identify the upsert run's head in chronological sorted order.
Why are the changes needed?
SCD2 core implementation.
Does this PR introduce any user-facing change?
No, this is core implementation of an unreleased feature (AutoCDC SCD2).
How was this patch tested?
Unit tested in
Scd2BatchProcessorSuite.Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7/4.8