Skip to content

[SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation#57365

Open
anew wants to merge 3 commits into
apache:masterfrom
anew:SPARK-57356-post-reconciliation-cleanup
Open

[SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation#57365
anew wants to merge 3 commits into
apache:masterfrom
anew:SPARK-57356-post-reconciliation-cleanup

Conversation

@anew

@anew anew commented Jul 20, 2026

Copy link
Copy Markdown
Contributor

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

(cherry picked from commit 0d5d7d3)
@anew anew changed the title [PSARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [WIP][SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation Jul 20, 2026
val metadata = reconciledDf.schema(c).metadata
F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata)
case c =>
F.col(c)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.
// `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)(

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • 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

@anew anew changed the title [WIP][SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation Jul 20, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants