anew opened a new pull request, #57365: URL: https://github.com/apache/spark/pull/57365
### 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's
rows get their final `startAt` / `endAt` intervals.
- **`dropLeftoverDeletesPostReconciliation`** — drops delete-encoded rows
(tombstones and
decomposition tails) that became redundant after reconciliation. A
delete-encoded row is
redundant when the immediately preceding row's reconciled `endAt`
already equals its own
sequence, i.e. the preceding upsert already encodes the delete boundary.
Redundancy is
computed per key using a `lag` over the chronological per-key window
(`__previous_end_at` /
`__is_redundant_delete_encoding` temporary columns, dropped before
returning).
- e.g. an open upsert `[10, null)` followed by a tombstone at `15`
reconciles to a closed
upsert `[10, 15)`, making the tombstone redundant.
- e.g. an existing closed `[10, 20)` bisected by an event at `15`
reconciles the event to
`[15, 20)`, making the `[null, 20)` decomposition tail redundant.
- **`promoteDecompositionTailsToTombstones`** — converts decomposition
tails that *survive*
the deletion cleanup above into tombstones. A surviving tail is an
unmatched delete
boundary; rewriting its `recordStartAt` and `startAt` to the tail's end
sequence lets
downstream 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 yet
wired 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 `Scd2BatchProcessorSuite` covering the two new
methods (drop-vs-keep
for 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
--
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]
