[SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end - #57584
[SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end#57584anew wants to merge 2 commits into
Conversation
…end to end Add Scd2MergeStreamingWrite (mirroring Scd1MergeStreamingWrite): it resolves the auxiliary-table identifier, constructs an Scd2ForeachBatchHandler over an Scd2BatchProcessor, and drives it via foreachBatch. Route an SCD2 AutoCdcMergeFlow to it in FlowPlanner, replacing the AUTOCDC_SCD2_NOT_SUPPORTED throw (the last remaining SCD2 gate; the flow-schema and auxiliary-table gates were removed by SPARK-58319 and SPARK-58320). This makes SCD2 AutoCDC flows runnable end to end. Adds AutoCdcScd2SinglePipelineSuite, an end-to-end suite that registers an SCD2 flow, runs it through the pipeline, and asserts SCD2 target semantics (an upsert opens a current record; an update closes the prior record and opens a new one; a delete closes the current record) plus auxiliary-table materialization. Also refreshes a now-stale AutoCdcFlowSuite test whose name/comment referenced the removed SCD2-not-supported gate; its behavior is already covered by the reserved framework-column rejection test, so the redundant copy is removed. Co-authored-by: Opus 4.8
| ) | ||
| } | ||
|
|
||
| test("SCD2: an upsert lands an open current record in an empty target table") { |
There was a problem hiding this comment.
note: the test coverage here s not exhaustive, but it would be hard to test all possible cases in an end-to-end suite. This suite merely asserts that it works end-to-end. The convergence test, however, should provide decent coverage.
There was a problem hiding this comment.
I agree with this analysis. I trust the principle that the results shouldn't be ingestion order dependent more than I trust my ability to read through a big pile of specific rows and intuit the correct result.
…SCD2 Generalize AutoCdcScd1OutOfOrderConvergenceSuite to run its randomized order-invariance differential test for both SCD Type 1 and SCD Type 2, and rename it to AutoCdcOutOfOrderConvergenceSuite. The target-table DDL and the AutoCDC flow are now parameterized on ScdType (SCD2 adds the __START_AT / __END_AT interval columns), and a second test runs the same random-event-stream convergence check under SCD2. Only the user-visible target is compared across the sorted-single-batch vs shuffled-multi-batch runs; the auxiliary tables legitimately differ by arrival order (deletedByBatchId stamps and cross-batch GC depend on batching) even when the target converges, so they are not compared. Co-authored-by: Opus 4.8
jose-torres
left a comment
jose-torres
left a comment
There was a problem hiding this comment.
Looks good conditional on the two things I raised checking out
| * user-selected data columns: just the CDC metadata column for SCD1, plus the __START_AT / | ||
| * __END_AT interval bounds for SCD2. The sequencing type is BIGINT here. | ||
| */ | ||
| private def reservedColumnsDdl(scdType: ScdType): String = scdType match { |
There was a problem hiding this comment.
I ultimately decided I don't care that SCD1 and SCD2 disagree on the encapsulation boundary for the DDL, but I'm leaving this comment as a record of the train of thought in case it jumps out at anyone.
| runPipeline(outOfOrderCtx) | ||
| } | ||
|
|
||
| // Only the user-visible target must converge. The auxiliary tables legitimately differ by |
There was a problem hiding this comment.
This is fine under the assumption, which I think is correct but want to confirm, that any auxiliary row generated at version N is guaranteed to stop affecting results if the ingestion durably advances to some version M not too much higher than N. (It would be a problem, to pick an exaggerated example, if some category of auxiliary row caused different results starting at N + 1000; then just checking consistency betwen the targets would not be enough to confirm that the behavior is meaningfully the same.)
| ) | ||
| } | ||
|
|
||
| test("SCD2: an upsert lands an open current record in an empty target table") { |
There was a problem hiding this comment.
I agree with this analysis. I trust the principle that the results shouldn't be ingestion order dependent more than I trust my ability to read through a big pile of specific rows and intuit the correct result.
2 participants
What changes were proposed in this pull request?
Adds
Scd2MergeStreamingWrite(mirroringScd1MergeStreamingWrite): it resolves the auxiliary-table identifier, constructs anScd2ForeachBatchHandlerover anScd2BatchProcessor, and drives it via Structured StreamingforeachBatch.FlowPlannernow routes an SCD2AutoCdcMergeFlowto it, replacing theAUTOCDC_SCD2_NOT_SUPPORTEDthrow.Why are the changes needed?
This removes the last remaining SCD2 gate. The flow-schema derivation (SPARK-58319), the auxiliary-table spec (SPARK-58320), the reserved-column / track-history validation (SPARK-57251, SPARK-58313), and the per-microbatch reconciliation handler (SPARK-57395) have all merged; this change makes SCD2 AutoCDC flows runnable end to end.
Does this PR introduce any user-facing change?
Yes:
AUTO CDC ... STORED AS SCD TYPE 2pipelines are now supported and executable, where previously they failed withAUTOCDC_SCD2_NOT_SUPPORTED.How was this patch tested?
New
AutoCdcScd2SinglePipelineSuiteruns SCD2 flows end to end through the pipeline and asserts SCD2 target semantics (an upsert opens a current record; an update closes the prior record and opens a new one; a delete closes the current record) plus auxiliary-table materialization. Also refreshed anAutoCdcFlowSuitetest whose name/comment referenced the removed gate.Generalized the randomized out-of-order convergence test (AutoCdcScd1OutOfOrderConvergenceSuite, renamed to AutoCdcOutOfOrderConvergenceSuite) to run under both SCD Type 1 and SCD Type 2. It generates a random CDC event stream (with deletes, duplicate events, and nulls), feeds it once as a single sorted micro-batch and once as several shuffled micro-batches, and asserts the two target tables converge — verifying SCD2 reconciliation is order-invariant end to end. Only the user-visible target is compared; the auxiliary tables legitimately differ by arrival order (deletedByBatchId stamps and cross-batch GC depend on batching) even when the target converges. Verified non-seed-fragile across several seeds.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Opus 4.8