[SPARK-57378][SDP] Implement SCD2 Batch Processor; Merge Reconciled Rows into Aux and Target Tables#57444
Closed
anew wants to merge 5 commits into
Closed
[SPARK-57378][SDP] Implement SCD2 Batch Processor; Merge Reconciled Rows into Aux and Target Tables#57444anew wants to merge 5 commits into
anew wants to merge 5 commits into
Conversation
(cherry picked from commit eac615f)
(cherry picked from commit 48f6408)
Cleanups on the merge-into-aux-and-target implementation:
- Remove 4 trailing-whitespace lines flagged by scalastyle in
mergeRowsIntoAuxiliaryTable.
- Fix typo auxTableDeleyedByBatchId -> auxTableDeletedByBatchId.
- Rename misleading auxTableRecordStartAt to targetTableRecordStartAt in
mergeRowsIntoTargetTable (it reads from the target, not the aux, table).
- Fix grammar in the isVisibleTargetRow comment ("its" -> "it's",
"Only op upsert" -> "Only non-no-op upsert").
- Turn the aux-table GC full-scan caveat into an explicit TODO.
Fill coverage gaps in the merge/tag paths: - identifyAndTagAuxRows: a two-key case asserting the per-key window never lets one key's run influence another's tagging. - identifyAndTagAuxRows / mergeRowsIntoAuxiliaryTable / mergeRowsIntoTargetTable: user columns whose names contain a dot (e.g. `user.name`) must be quoted at every pass-through/assignment site, matching the coverage reconcileStartAndEndAt already has. Each test fails with UNRESOLVED_COLUMN if the corresponding QuotingUtils.quoteIdentifier call is dropped.
Fill the remaining merge coverage gaps: - mergeRowsIntoAuxiliaryTable: one merge that simultaneously inserts a new routed row, updates a surviving row, logically deletes a non-surviving affected row, and garbage-collects an older-batch row - exercising the whenMatched/whenNotMatched/whenNotMatchedBySource clauses together. - mergeRowsIntoAuxiliaryTable and mergeRowsIntoTargetTable: an in-place update over a schema with several non-key columns (name, status, score), asserting every non-key column is refreshed - guards the non-key update assignment map against dropping columns. Each test fails under the corresponding mutation (GC clause removed, or the assignment map truncated to one column).
anew
changed the title
szehon-ho
reviewed
Jul 24, 2026
| * Temporary in that the column has no observable side effect or persistence across microbatches. | ||
| */ | ||
| private[autocdc] val shouldRouteToAuxTableColName: String = | ||
| s"${AutoCdcReservedNames.prefix}should_route_to_aux_table" |
Member
There was a problem hiding this comment.
These temporary names are not covered by reservedFrameworkColNames, and input validation does not reject user columns in the AutoCDC internal namespace. A user column named __spark_autocdc_should_route_to_aux_table is overwritten here and later dropped; the sibling __spark_autocdc_should_delete_target_row is also overwritten while constructing the target MERGE source. This can lose user data or make the merge source ambiguous. Please reserve/reject these names (ideally the full AutoCDC-owned namespace) and add collision tests.
Contributor
Author
There was a problem hiding this comment.
Yes, this is true for all of the reserved metadata column names, already in previous PRs. We should make sure at the time of graph registration (or flow analysis) that no user columns collide with the autocdc metadata prefix. Do you mind if we do that in a follow-up PR? I created https://issues.apache.org/jira/browse/SPARK-58313
szehon-ho
approved these changes
Jul 24, 2026
szehon-ho
left a comment
szehon-ho
left a comment
Member
There was a problem hiding this comment.
ok lets do the follow up in another pr as discussed
jose-torres
approved these changes
Jul 24, 2026
| // collection now, proactively hard-delete it. | ||
| // TODO: This GC triggers a full scan of the aux table; revisit whether to GC only | ||
| // periodically rather than on every microbatch. | ||
| .whenNotMatchedBySource(isGarbageCollectableAuxRow) |
Contributor
There was a problem hiding this comment.
Noting for the record that this notMatchedBySource is safe, even though the incoming batch built from originalAffectedRowsFromAuxiliaryTable may not have included all target rows in the first place, because isGarbageCollectableAuxRow will exclude anything that's not already marked as ready for deletion.
jose-torres
pushed a commit
that referenced
this pull request
Jul 24, 2026
…ows into Aux and Target Tables Note: This continues AnishMahto's #56457, opened from my fork while the original author is out of office. It carries his implement merge into aux and target + self-review commits and adds review fixes and additional test coverage on top. #56457 should be closed in favor of this PR to keep review in one place. ### What changes were proposed in this pull request? This is the next step in the SCD2 AutoCDC batch-processor series (following SPARK-57356, "Cleanup Delete Encoding Rows Post-Reconciliation"). It adds the final merge stage that writes a reconciled microbatch into the auxiliary and target tables. Three methods are added to Scd2BatchProcessor: - identifyAndTagAuxRows — tags each post-reconciliation row with shouldRouteToAuxTableColName, marking which rows belong in the auxiliary table (tombstones and other delete-boundary encodings) versus the visible target table. - mergeRowsIntoAuxiliaryTable — merges the reconciled rows tagged for the auxiliary table into it, handling tombstone insertion/advancement and deletedByBatchId-scoped garbage collection. - mergeRowsIntoTargetTable — merges the reconciled visible rows into the target table (the user-facing SCD2 history). Together with the reconciliation and cleanup steps already merged, this completes the per-microbatch transformation pipeline; the foreachBatch callback that drives it end-to-end lands in the follow-up (SPARK-57395). ### Why are the changes needed? The SCD2 batch processor is being built incrementally across the SPARK-571xx series. After a microbatch is preprocessed, decomposed, reconciled, and cleaned up, its rows must actually be written to the auxiliary and target tables using the correct MERGE semantics. This PR supplies that final merge step, which the engine's SCD2 execution path depends on. ### Does this PR introduce any user-facing change? No. These are internal (private[autocdc]) batch-processor methods; SCD2 AutoCDC is not yet reachable end-to-end (the driving foreachBatch handler and the API/planner wiring land in separate changes). ### How was this patch tested? New unit tests: - Scd2BatchProcessorMergeSuite — covers the aux/target merge behavior (tombstone insertion and advancement, GC of superseded aux rows, target-table history writes). - Additional cases in Scd2BatchProcessorSuite — identifyAndTagAuxRows routing, per-key window isolation, dotted / identifier-sensitive column names, and combined-branch / multi-column merges. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Closes #57444 from anew/SPARK-57378-merge-reconciled-rows. Lead-authored-by: Andreas Neumann <andreas.neumann@databricks.com> Co-authored-by: Anish Mahto <anish.mahto99@gmail.com> Signed-off-by: Jose Torres <jtorres@apache.org> (cherry picked from commit 0687ebf) Signed-off-by: Jose Torres <jtorres@apache.org>
Contributor
|
merged to master and 4.x |
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
4 participants
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.
Note: This continues @AnishMahto's #56457, opened from my fork while the original author is out of office. It carries his implement merge into aux and target + self-review commits and adds review fixes and additional test coverage on top. #56457 should be closed in favor of this PR to keep review in one place.
What changes were proposed in this pull request?
This is the next step in the SCD2 AutoCDC batch-processor series (following SPARK-57356, "Cleanup Delete Encoding Rows Post-Reconciliation"). It adds the final merge stage that writes a reconciled microbatch into the auxiliary and target tables. Three methods are added to Scd2BatchProcessor:
Together with the reconciliation and cleanup steps already merged, this completes the per-microbatch transformation pipeline; the foreachBatch callback that drives it end-to-end lands in the follow-up (SPARK-57395).
Why are the changes needed?
The SCD2 batch processor is being built incrementally across the SPARK-571xx series. After a microbatch is preprocessed, decomposed, reconciled, and cleaned up, its rows must actually be written to the auxiliary and target tables using the correct MERGE semantics. This PR supplies that final merge step, which the engine's SCD2 execution path depends on.
Does this PR introduce any user-facing change?
No. These are internal (private[autocdc]) batch-processor methods; SCD2 AutoCDC is not yet reachable end-to-end (the driving foreachBatch handler and the API/planner wiring land in separate changes).
How was this patch tested?
New unit tests:
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)