[SPARK-57395][SDP] Implement SCD2 Batch Processor; foreachBatch Callback

[anew]

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