[SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation by anew · Pull Request #57365 · apache/spark

anew · 2026-07-20T06:06:35Z

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

anew · 2026-07-20T18:27:20Z

+        val metadata = reconciledDf.schema(c).metadata
+        F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata)
+      case c =>
+        F.col(c)


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.

fixed and added a test.

anew · 2026-07-20T19:23:17Z

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


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

jose-torres

As with the prior PR, the behavior here makes sense on its own terms but raises questions we'll want to go back and review on a full-algorithm basis. Please do clarify the one example, and I'll wait a bit for comments from @szehon-ho before merging.

jose-torres · 2026-07-22T01:31:19Z

+   * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+   * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+   *
+   * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`


The way this is stated is confusing to me. Why does the history record identity follow the prefix of the range when closing an active row but the suffix when closing an inactive row?

Good point — the asymmetry was only in the wording, not the logic. Rewrote the doc to state the single rule both cases share: a delete-encoded row is redundant when the immediately preceding reconciled upsert's endAt equals the row's own boundary (its startAt/identity is irrelevant to the check). I also spelled out where the [null, 20) decomposition tail comes from — the open-head + tail split produced when a closed row is bisected by an out-of-order event.

jose-torres · 2026-07-22T01:33:26Z

+        isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+      )
+
+    withWindowCols


I suppose the implementation of this method doesn't end up depending on the row identity confusion above, but I still think it's worth clearing up.

I suppose. this still relates to the previous comment? If so, I hope the updated doc makes it clear

jose-torres · 2026-07-22T01:35:00Z

+  }
+
+  /**
+   * Convert surviving decomposition tails into tombstones.


Another thing to handle in a post-hoc pass after implementation: we should make sure that these two are actually handled in a different enough way to require two different representations. It kinda sounds to me like a decomposition tail is just the within-batch version of a tombstone.

Agreed this is worth a look on the full-algorithm pass. They're handled identically here (both are just delete-boundary encodings), and a decomposition tail is essentially the within-batch analog of a tombstone — the main distinction today is provenance (a tail is synthesized during bisection and carries a null recordStartAt, a tombstone comes from an actual delete event). Whether they can be collapsed into one representation is a good question to settle when we review the algorithm end-to-end; but I think it s outside of this PR's scope.

jose-torres · 2026-07-22T01:44:39Z

+        val metadata = reconciledDf.schema(c).metadata
+        F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata)
+      case c =>
+        F.col(QuotingUtils.quoteIdentifier(c))


I guess we only need the quote here because we know the exact names of the other cases and they don't contain special characters?

Exactly. The other case branches match framework columns. I added a comment to clarify that

szehon-ho · 2026-07-22T17:33:26Z

+      )
+      .withColumn(
+        Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+        isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)


Could we defensively verify that the preceding row is an upsert? Adjacent delete-encoded rows with the same boundary should not occur because upstream cleanup prevents them, but checking here would make this method enforce its documented invariant directly. For example:

val previous = row.lagBy(1, orderChronologicallyPerKeyWindow) val isRedundantDeleteEncoding = isDeleteEncodedRow && // Defensive: the predecessor should always be an upsert because upstream cleanup prevents // adjacent delete-encoded rows with the same boundary. RowClassifier.isUpsertRepresentingRow(previous) && (previous.endAt <=> endAt)

Good idea — done. Adapted it to the existing window setup: I build the lagged interval via row.lagBy(1, orderChronologicallyPerKeyWindow) and gate on RowClassifier.isUpsertRepresentingRow(previous) before the endAt comparison, which also let me drop the separate previousEndAt temp column. Added a test with two adjacent tombstones on the same boundary — the second now correctly survives (its predecessor isn't an upsert), which the old check would have dropped.

(cherry picked from commit 75ce170)

(cherry picked from commit 0d5d7d3)

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.

- Rewrite the dropLeftoverDeletesPostReconciliation scaladoc so both examples are stated as the same rule (a delete-encoded row is redundant when the immediately preceding reconciled upsert's endAt equals the row's boundary), and spell out where the [null, 20) decomposition tail in the second example comes from (the head + tail split produced when a closed row is bisected). - Add a comment in promoteDecompositionTailsToTombstones explaining that only the default (user-column) case needs quoting, since the other cases match framework columns with known, identifier-safe names.

dropLeftoverDeletesPostReconciliation only checked that the immediately preceding row's endAt equals the delete-encoded row's boundary, but the method's documented rule is that the preceding *upsert* must close on that boundary. Add RowClassifier.isUpsertRepresentingRow(previous) to the redundancy predicate so the method enforces its invariant directly, rather than dropping a delete-encoded row that merely abuts another delete-encoded row on the same boundary. Build the lagged interval via row.lagBy(1, ...) and read previous.endAt from it, which removes the separate previousEndAtColName temp column. Add a test covering two adjacent tombstones on the same boundary: the first is redundant and drops, the second survives because its predecessor is not an upsert. Co-authored-by: Isaac

anew · 2026-07-22T20:28:49Z

had to force-push this in order to rebase on latest master (and fix the broken CI caused by the previous master).

…oding Rows Post-Reconciliation ### 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 Closes #57365 from anew/SPARK-57356-post-reconciliation-cleanup. Lead-authored-by: Andreas Neumann <anew@apache.org> Co-authored-by: Anish Mahto <anish.mahto99@gmail.com> Signed-off-by: Szehon Ho <szehon.apache@gmail.com> (cherry picked from commit e350bb0) Signed-off-by: Szehon Ho <szehon.apache@gmail.com>

szehon-ho · 2026-07-22T22:09:12Z

Merge Summary:

merged into master e350bb0
merged into branch-4.x cebd378

Posted by merge_spark_pr.py

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

anew commented Jul 20, 2026

View reviewed changes

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

jose-torres approved these changes Jul 22, 2026

View reviewed changes

szehon-ho reviewed Jul 22, 2026

View reviewed changes

szehon-ho approved these changes Jul 22, 2026

View reviewed changes

AnishMahto and others added 5 commits July 22, 2026 20:24

implement post-reconciliation cleanup

787d705

(cherry picked from commit 75ce170)

cleanup

cbf50d9

(cherry picked from commit 0d5d7d3)

anew force-pushed the SPARK-57356-post-reconciliation-cleanup branch from e7480d8 to 0803968 Compare July 22, 2026 20:28

szehon-ho closed this in e350bb0 Jul 22, 2026

Uh oh!

Conversation

anew commented Jul 20, 2026

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jose-torres left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anew Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anew commented Jul 22, 2026

Uh oh!

szehon-ho commented Jul 22, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

anew Jul 22, 2026 •

edited

Loading