[SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation#57365
Open
anew wants to merge 3 commits into
Open
[SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation#57365anew wants to merge 3 commits into
anew wants to merge 3 commits into
Conversation
anew
commented
Jul 20, 2026
| val metadata = reconciledDf.schema(c).metadata | ||
| F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata) | ||
| case c => | ||
| F.col(c) |
Contributor
Author
There was a problem hiding this comment.
That will mis-resolve user columns whose literal names contain dots or other identifier-sensitive characters, e.g. user.name, as nested-field paths. The rest of this class already handles these cases with QuotingUtils.quoteIdentifier, and there is existing test coverage for dotted tracked-history columns. This should use F.col(QuotingUtils.quoteIdentifier(c)) for pass-through columns.
Contributor
Author
There was a problem hiding this comment.
fixed and added a test.
The default pass-through branch used F.col(c), which parses column names containing dots (e.g. `user.name`) as nested-field paths and fails to resolve them. Quote the name with QuotingUtils.quoteIdentifier, matching how the rest of the class handles user column names, and add a test with a dotted column.
anew
commented
Jul 20, 2026
| // `name`) and fail to resolve, so the select would throw. A decomposition tail (recordStartAt | ||
| // and startAt null, endAt=20) is promoted to a tombstone at 20 while the dotted user column | ||
| // survives verbatim. | ||
| val df = targetTableOf(userSchema)( |
Contributor
Author
There was a problem hiding this comment.
- Without the fix in this last commit, it fails with
[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column ... with name `user.name` cannot be resolved
- With the fix, the full suite passes
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This PR adds two post-reconciliation cleanup stages to the SCD2 batch processor
(
Scd2BatchProcessor), used by Declarative Pipelines' AUTO CDC / SCD Type 2 processing.Both operate on the output of
reconcileStartAndEndAt, which is where an affected key'srows get their final
startAt/endAtintervals.dropLeftoverDeletesPostReconciliation— drops delete-encoded rows (tombstones anddecomposition tails) that became redundant after reconciliation. A delete-encoded row is
redundant when the immediately preceding row's reconciled
endAtalready equals its ownsequence, i.e. the preceding upsert already encodes the delete boundary. Redundancy is
computed per key using a
lagover the chronological per-key window (__previous_end_at/__is_redundant_delete_encodingtemporary columns, dropped before returning).[10, null)followed by a tombstone at15reconciles to a closedupsert
[10, 15), making the tombstone redundant.[10, 20)bisected by an event at15reconciles the event to[15, 20), making the[null, 20)decomposition tail redundant.promoteDecompositionTailsToTombstones— converts decomposition tails that survivethe deletion cleanup above into tombstones. A surviving tail is an unmatched delete
boundary; rewriting its
recordStartAtandstartAtto the tail's end sequence letsdownstream auxiliary-table handling preserve it as a tombstone. Column metadata and the
input schema are preserved.
Both methods are
private[autocdc]building blocks with unit coverage; they are not yetwired into the top-level processing flow (a follow-up composes them into the pipeline).
Why are the changes needed?
After start/end reconciliation, some delete-encoded rows carry information that the
reconciled upserts already express, and some decomposition tails remain as standalone delete
boundaries. Without this cleanup, redundant tombstones/tails would be routed to the auxiliary
table needlessly, and unmatched tails would not be represented in the form downstream aux
handling expects. These two stages normalize the reconciled output so that each key's history
is encoded once and correctly.
Does this PR introduce any user-facing change?
No, SCD2 is an unreleased feature.
How was this patch tested?
Added 13 unit tests to
Scd2BatchProcessorSuitecovering the two new methods (drop-vs-keepfor tombstones and decomposition tails, no-window-predecessor cases, upserts never dropped,
per-key independence, combined redundant drops; and tail-to-tombstone rewriting, pass-through
of tombstones/upserts, user-column and schema/metadata preservation, per-key promotion).
Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7 and 4.8