[SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end by anew · Pull Request #57584 · apache/spark

anew · 2026-07-27T21:59:02Z

What changes were proposed in this pull request?

Adds Scd2MergeStreamingWrite (mirroring Scd1MergeStreamingWrite): it resolves the auxiliary-table identifier, constructs an Scd2ForeachBatchHandler over an Scd2BatchProcessor, and drives it via Structured Streaming foreachBatch. FlowPlanner now routes an SCD2 AutoCdcMergeFlow to it, replacing the AUTOCDC_SCD2_NOT_SUPPORTED throw.

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 2 pipelines are now supported and executable, where previously they failed with AUTOCDC_SCD2_NOT_SUPPORTED.

How was this patch tested?

New AutoCdcScd2SinglePipelineSuite runs 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 an AutoCdcFlowSuite test 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

anew · 2026-07-28T01:18:56Z

+    )
+  }
+
+  test("SCD2: an upsert lands an open current record in an empty target table") {


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.

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.

jose-torres

Looks good conditional on the two things I raised checking out

jose-torres · 2026-07-28T06:27:21Z

+   * 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 {


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.

I made this symmetric by

renaming cdcMetadataDdl to scd1MetadataDdl

adding a new scd2MetadataDdl to the Mixin

using these constants everywhere.

But digging into this, I realized that there are not many paces where the new scd2MetadataDdl are used. The reason is a gap in test coverage: while we have quite a few end to end tests for SCD1, we have very few for SCD2. I will close that test gap in a follow-up PR with Jira: https://issues.apache.org/jira/browse/SPARK-58409

jose-torres · 2026-07-28T06:35:08Z

        runPipeline(outOfOrderCtx)
      }

+      // Only the user-visible target must converge. The auxiliary tables legitimately differ by


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.)

Confirmed. findAffectedRowsFromAuxiliaryTable bounds aux-row participation per key to recordStartAt >= minSequenceInMicrobatch plus a single "anchor" — the aux row with the largest recordStartAt strictly below the batch's min sequence. There's no unbounded look-back (no "N+1000" case): an aux row from version N only affects microbatches adjacent to N, so once ingestion advances past it (with a newer anchor present) it stops mattering. That's what makes the target-only convergence comparison valid.

jose-torres · 2026-07-28T06:36:55Z

+    )
+  }
+
+  test("SCD2: an upsert lands an open current record in an empty target table") {


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.

uros-b · 2026-07-28T16:46:17Z

            }
          case ScdType.Type2 =>
-            throw new AnalysisException(
-              errorClass = "AUTOCDC_SCD2_NOT_SUPPORTED",


If this error class is no longer used, please remove it completely (for example, from common/utils/src/main/resources/error/error-conditions.json) to prevent confusion and to keep the error-class registry up to date

good catch, thanks, fixed!

szehon-ho

LGTM. The wiring mirrors the SCD1 path faithfully, and generalizing the convergence test to SCD2 is convincing evidence that the reconciliation is order-invariant end to end.

I checked the "last gate" claim: no AUTOCDC_SCD2_NOT_SUPPORTED references remain anywhere, and no release tag contains the commit that introduced it, so dropping it from the registry is safe. Flow.schema / load, AutoCdcAuxiliaryTable.buildAuxiliaryTableSpecFor, SqlGraphRegistrationContext, and the Python stored_as_scd_type path all already handle Type2, and AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE lives in DatasetManager and is SCD-type-agnostic. Two non-blocking comments inline.

One question on coverage rather than on the code. This PR is what makes several SCD2-only paths reachable for the first time, and I would like to understand which of them you consider covered:

TRACK HISTORY ON / ON * EXCEPT is unit-tested in Scd2BatchProcessorSuite, and its ChangeArgs mapping is tested in SqlPipelineSuite, but nothing runs it through a pipeline. Is an E2E case worth adding here, or is the processor-level coverage the intended line?
SCD1 has dedicated E2E suites for full refresh, schema evolution, key drift, multi-pipeline, and aux/target durability. The DatasetManager aux validation looks SCD-type-agnostic, but a couple seem SCD2-specific in ways the SCD1 suites would not catch: full refresh has to reset accumulated history alongside the aux table, and schema evolution has to keep __START_AT / __END_AT in the positions Flow.schema's Type2 branch specifies. Are those planned as follow-ups, or is there a reason they collapse onto the SCD1 coverage?

szehon-ho · 2026-07-28T18:18:55Z

+    // The SCD2 auxiliary table exists and carries the aux-only deleted-by-batch-id marker column
+    // in addition to the full target row schema.
+    val auxColumns = spark.table(auxTableNameFor("target")).schema.fieldNames.toSet
+    assert(auxColumns.contains(AutoCdcReservedNames.cdcMetadataColName))
+    assert(auxColumns.contains("__START_AT"))
+    assert(auxColumns.contains("__END_AT"))


The comment says the aux table "carries the aux-only deleted-by-batch-id marker column", but no assertion checks for it. Relatedly, the class scaladoc says "both the target table and the auxiliary table contents are asserted at the end", whereas what is asserted here is the schema rather than the contents.

Scd2BatchProcessor.deletedByBatchIdColName is private[pipelines] and AutoCdcScd2AuxiliaryTableSpecSuite already uses it from this package, so this could assert the full materialized schema (the target fields plus the marker appended) and then match its own comment. The distinct value of this test over the spec suite is proving the table really is created by a live run, and a full-schema assertion keeps that while making it stronger.

Same thought for the two literals: Scd2BatchProcessor.startAtColName / endAtColName are reachable here, and those two are precisely the reserved columns that do not carry AutoCdcReservedNames.prefix, so a rename would slip past a string literal.

good suggestion, done.

szehon-ho · 2026-07-28T18:18:55Z

+      s"\nscdType=${scdType.label} seed=$seed " +
+      s"(rerun with -D$seedSystemProperty=$seed to reproduce)\n" +


The property this message tells the developer to set is still spark.sql.test.autocdc.scd1OutOfOrderConvergenceSeed (line 80), so a failing SCD2 run prints an SCD1-named knob. Worth renaming along with the suite.

Separately, the comment on line 193 justifies the scd-type table-name suffix as keeping the SCD1 and SCD2 tests from colliding within a run, but the mixin's afterEach calls SharedTablesInMemoryRowLevelOperationTableCatalog.reset(), so the tables cannot collide. The suffix is harmless; the stated reason just is not the real one.

fixed both of them!

anew · 2026-07-28T19:23:09Z

@szehon-ho thanks for pointng out the test gap. I had just discovered that myself while addressing Jose's comments and I already created a follow-up Jira to close that gap: https://issues.apache.org/jira/browse/SPARK-58409

…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

…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

…rror class Wiring the SCD2 streaming write removed the last thrower of AUTOCDC_SCD2_NOT_SUPPORTED (the FlowPlanner gate; the Flow.schema/load gates were already removed by SPARK-58319). The error class is now unused, so remove it from the error-conditions registry to keep it up to date and avoid confusion. Co-authored-by: Opus 4.8

…dataDdl to scd1MetadataDdl The SCD1 and SCD2 target-table DDLs were expressed asymmetrically in the E2E test suites: SCD1 shared a single cdcMetadataDdl mixin helper, while every SCD2 suite inlined its framework columns. Make the two symmetric: - Rename cdcMetadataDdl to scd1MetadataDdl and add a parallel scd2MetadataDdl in AutoCdcGraphExecutionTestMixin. scd2MetadataDdl encapsulates the full SCD2 reserved column set (the __START_AT / __END_AT interval bounds plus the CDC metadata column), built from the Scd2BatchProcessor column-name constants. - Use scd2MetadataDdl in AutoCdcOutOfOrderConvergenceSuite (SCD2 branch) and AutoCdcScd2SinglePipelineSuite, removing the inlined DDL. - Update the ~50 SCD1 call sites to the new name. Widen Scd2BatchProcessor.recordStartAtFieldName to private[pipelines] so the metadata struct field name can be referenced from the test helper (consistent with the other already-widened SCD2 column-name members). Pure test refactor; no behavior change. Co-authored-by: Opus 4.8

…on and fix convergence-suite naming - AutoCdcScd2SinglePipelineSuite: assert the auxiliary table's exact field list (user columns + framework columns + the deleted-by-batch-id marker) via the reserved-name constants instead of substring `.contains` checks on string literals, so a rename of any framework column -- including the non-prefixed __START_AT / __END_AT -- is caught. This also makes the test match its comment, which claims the marker column is present. - AutoCdcOutOfOrderConvergenceSuite: rename the seed system property from the SCD1-specific `scd1OutOfOrderConvergenceSeed` to `outOfOrderConvergenceSeed`, since the suite now covers both SCD types. Correct the table-name-suffix comment: afterEach resets the catalog between the SCD1 and SCD2 test cases, so the suffix is for readability, not collision avoidance. Co-authored-by: Opus 4.8

Co-authored-by: Opus 4.8

anew commented Jul 28, 2026

View reviewed changes

jose-torres approved these changes Jul 28, 2026

View reviewed changes

uros-b approved these changes Jul 28, 2026

View reviewed changes

szehon-ho approved these changes Jul 28, 2026

View reviewed changes

anew added 6 commits July 28, 2026 23:49

[SPARK-58321][SDP] Empty commit to retrigger CI

bc381be

Co-authored-by: Opus 4.8

anew force-pushed the spark-58321-scd2-streaming-write-v2 branch from 3a264f7 to bc381be Compare July 28, 2026 23:49

		s"\nscdType=${scdType.label} seed=$seed " +
		s"(rerun with -D$seedSystemProperty=$seed to reproduce)\n" +

Uh oh!

Conversation

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

Uh oh!

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!

anew Jul 28, 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!

jose-torres left a comment

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!

szehon-ho left a comment

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 28, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

anew commented Jul 27, 2026 •

edited

Loading

anew Jul 28, 2026 •

edited

Loading