anew opened a new pull request, #57495:
URL: https://github.com/apache/spark/pull/57495
### What changes were proposed in this pull request?
This PR adds Scd2ForeachBatchHandler, the entry point that drives one SCD
Type 2 AutoCDC microbatch reconciliation from a Structured Streaming
foreachBatch callback. It is the SCD2 analog of the existing
Scd1ForeachBatchHandler, and it composes the previously-landed
Scd2BatchProcessor transforms into the single end-to-end per-batch pipeline:
1. Validate the incoming microbatch (ScdBatchValidator — null keys / null
sequence / orderable sequence).
2. Preprocess the microbatch and compute the per-key minimum sequence.
3. Pull in the affected rows from both the auxiliary and target tables for
the keys in this batch.
4. Union the microbatch with the affected rows, then run decomposition
(decomposeOutOfOrderRows, dropRedundantRowsPostDecomposition,
assertWellFormedRowsPostDecomposition).
5. Reconcile and route (reconcileStartAndEndAt,
dropLeftoverDeletesPostReconciliation, promoteDecompositionTailsToTombstones,
identifyAndTagAuxRows).
6. Merge the reconciled rows into the auxiliary table and the target table.
The handler is idempotent under same-batchId replay, relying on the
logical-delete / garbage-collection scheme already implemented in
Scd2BatchProcessor.
Files:
- Scd2ForeachBatchHandler.scala — the new handler (~94 lines).
- Scd2ForeachBatchHandlerSuite.scala — an end-to-end behavioral suite.
### Why are the changes needed?
The SCD2 batch processor and its individual reconciliation transforms landed
in prior PRs (e.g. SPARK-57378), but there was no component wiring them into a
runnable per-microbatch callback. This handler is the missing piece that lets
an SCD2 AutoCDC flow actually execute against the auxiliary and target tables
on each streaming batch, and it provides a single place to assert the composed
reconciliation behaves correctly end-to-end.
### Does this PR introduce any user-facing change?
No. This adds internal SDP/AutoCDC machinery only. SCD2 AutoCDC flows are
still gated (AUTOCDC_SCD2_NOT_SUPPORTED) and not yet reachable by users, so
there is no change to released or user-facing behavior.
### How was this patch tested?
A new end-to-end suite Scd2ForeachBatchHandlerSuite (35 tests) exercises the
handler against real auxiliary and target tables, covering:
- input validation (null key / null sequence fail the batch without
applying changes);
- basic SCD2 semantics (insert opens a current record; update closes the
old and opens a new; delete closes the current record);
- in-batch sequences (multiple updates, insert+delete,
insert/update/delete/re-insert);
- out-of-order / late-arriving events (late insert, late update bisecting
a record, late delete shortening a record, multiple bisections in one batch);
- idempotency under same-batch replay (updates, deletes, repeated values,
redelivered events) and cross-batch garbage collection of logically-deleted
tombstones;
- tracked vs. untracked column changes (untracked change updates in place;
tracked change opens history);
- multi-key / composite-key independence and case-sensitive/insensitive
key resolution.
### Was this patch authored or co-authored using generative AI tooling?
Generated-By: Opus 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]
