[SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites - #57689
Conversation
SCD2 analog of AutoCdcScd1FullRefreshSuite: full refresh wipes both the target rows and the richer SCD2 auxiliary table for the refreshed flow, resets the watermark so a previously-suppressed lower-sequence event lands, and under selective refresh leaves a non-refreshed target's aux state intact. The selective test also documents the SCD2-specific behavior that a non-refreshed target weaves a late lower-sequence event into history as a closed prior record rather than suppressing it as SCD1 does. Co-authored-by: Opus 4.8
SCD2 analog of AutoCdcScd1TargetTableDurabilitySuite, covering interop with a hand-populated target: a pre-loaded open current record closed/opened by a higher-sequence upsert, a lower-sequence upsert woven in as a closed prior record (SCD2-specific, no SCD1 equivalent), lazy auxiliary-table creation on the first run over a pre-loaded target, and framework-column auto-add when the target is created without __START_AT / __END_AT / _cdc_metadata. Adds a suite-local helper to seed a pre-existing open SCD2 record. Co-authored-by: Opus 4.8
SCD2 analog of AutoCdcScd1MultiPipelineSuite: independent target/auxiliary tables per target across pipelines, a downstream materialized view reading an SCD2 target without the framework columns, two pipelines merging into a shared SCD2 target, cross-pipeline schema evolution, and KEY_SCHEMA_DRIFT rejection when a second pipeline uses different keys. The schema-evolution test stops short of re-running the narrower pipeline against the widened target (which the SCD1 analog does): that path currently fails with NUM_COLUMNS_MISMATCH because Scd2ForeachBatchHandler unions the microbatch with affected rows without allowMissingColumns. Tracked as a separate bug (SPARK-58418). Co-authored-by: Opus 4.8
SCD2 analog of AutoCdcScd1AuxiliaryTableDurabilitySuite: recorded per-key history persists across incremental runs, a dry run does not provision the aux table, and a dropped aux table is transparently recreated. Schema-layout assertions reflect the SCD2 aux table (full target row schema plus the aux-only deleted-by-batch-id marker) rather than SCD1's keys+metadata, and the lower-sequence and dropped-aux cases document SCD2's history-preserving behavior (closed prior record) where SCD1 suppresses/overwrites. Co-authored-by: Opus 4.8
SCD2 analog of AutoCdcScd1KeyDriftSuite (13 tests): key-arity add/drop, key swap, dataType drift, order-invariance, nullability/metadata invariance, backtick invariance (both directions), case-sensitive vs case-insensitive resolver behavior, and the tampered-aux failure modes (missing/malformed keyColumnNames property, recorded key absent from the aux schema). Key-drift validation is SCD-type-agnostic, so the tampered-aux tests hand-build an SCD2-shaped aux table (full target row schema plus the deleted-by-batch-id marker) carrying the SCD2 scd-type property. Co-authored-by: Opus 4.8
SCD2 analog of AutoCdcScd1SchemaEvolutionSuite: nullable-column merge, new top-level column add, additive target-column evolution (which for SCD2 also extends the aux table since it mirrors the full target row), broadening the column selection, incompatible type changes (widen/narrow/timestamp->string), and a case-only source-column rename. The narrowing / dropped-column cases (a microbatch narrower than the evolved target, including dropped nested struct/array fields) are owned by AutoCdcScd2ColumnEvolutionSuite under SPARK-58418, which makes them reconcile correctly, so they are not duplicated here. Co-authored-by: Opus 4.8
|
Thank you @anew! |
szehon-ho
left a comment
szehon-ho
left a comment
There was a problem hiding this comment.
Test-only parity work, and the SCD2 semantics it encodes match what the reconciliation code actually does — the places where SCD2 diverges from SCD1 are called out deliberately rather than assumed away, which made this easy to follow. My comments are all about coverage the suites imply but don't exercise; nothing here looks wrong.
| // The dropped auxiliary table must be transparently recreated. The seq=1 record still lives | ||
| // in the target, and SCD2 reconciliation reads affected rows from the target as well as the | ||
| // aux table, so the seq=2 event closes the seq=1 record and opens a new one -- the recorded | ||
| // history survives the aux-table drop. |
There was a problem hiding this comment.
The last clause overstates what this test shows. Both records here are visible rows in the target, so dropping the aux table changes nothing observable — the aux contributes no state to this outcome. But the SCD2 aux exists precisely to hold rows that are not in the target (tombstones and coalesced no-op upserts, per buildScd2AuxiliaryTableSpecFor), and those don't survive a drop.
Could we cover the case where the aux is the only holder of the state? Roughly the shape of AutoCdcScd2FullRefreshSuite's second test: run #1 delete-only at seq=10 (target stays empty, aux holds the tombstone), drop the aux, then run #2 upsert at seq=5. With the aux the event lands closed at 10; without it, open. At minimum I'd soften the comment so a future reader doesn't take the aux as disposable.
| private def scd2AuxProps(keyColumnNamesJson: String): String = | ||
| s"TBLPROPERTIES (" + | ||
| s"'${AutoCdcAuxiliaryTable.scdTypePropertyKey}' = '${ScdType.Type2.label}', " + | ||
| s"'${AutoCdcAuxiliaryTable.keyColumnNamesProperty}' = '$keyColumnNamesJson')" |
There was a problem hiding this comment.
validateNoScdTypeDrift is reachable end-to-end from DatasetManager, right after validateNoKeyColumnDrift, but SCD_TYPE_DRIFT only has unit coverage in AutoCdcAuxiliaryTableSuite. The SCD1 suite structurally can't cover it, so this suite seems like the right home — and scd2AuxProps makes it cheap: pre-create the aux with scdTypePropertyKey set to ScdType.Type1.label, run an SCD2 flow, expect AUTOCDC_INVALID_STATE.SCD_TYPE_DRIFT.
| * Note: the *narrowing* / dropped-column cases (a microbatch narrower than the already-evolved | ||
| * target, incl. dropped nested struct/array fields) are covered by | ||
| * [[AutoCdcScd2ColumnEvolutionSuite]] under SPARK-58418, which makes them reconcile correctly, so | ||
| * they are intentionally not duplicated here. |
There was a problem hiding this comment.
This exclusion is accurate for the drop cases, but it leaves an additive one uncovered: AutoCdcScd1SchemaEvolutionSuite has "a new field added inside an array<struct> element between runs is added to the target", and AutoCdcScd2ColumnEvolutionSuite covers only dropped nested struct/array fields. So no SCD2 suite exercises a nested type additively — everything here is top-level scalars. Same for SCD1's "extra columns on the target that the AutoCDC flow does not emit are preserved". Either add the additive nested case or narrow this note so it doesn't read as full parity.
| test("broadening the column selection between runs adds the newly-included column to " + | ||
| "the target") { |
There was a problem hiding this comment.
This covers broadening columnSelection. Is the SCD2-only analog — changing trackHistorySelection between runs — deliberately out of scope? It directly determines whether an upsert opens a new record, and its only end-to-end exercise today is in AutoCdcScd2ColumnEvolutionSuite. A line in the suite header noting the deferral would keep the parity claim precise.
| // existing `value`. Under case-insensitive resolution that collides, and (unlike SCD1, which | ||
| // surfaces AMBIGUOUS_REFERENCE deeper in the MERGE plan) the SCD2 write path reports | ||
| // COLUMN_ALREADY_EXISTS when adding the duplicate column. |
There was a problem hiding this comment.
Where exactly is COLUMN_ALREADY_EXISTS raised here? My reading is that it comes from creating the SCD2 aux table, whose schema mirrors the target and therefore contains both value and Value, tripping duplicate-column validation before the MERGE plan is analyzed — which would also explain why SCD1, with its keys-only aux, instead reaches AMBIGUOUS_REFERENCE deeper in. If that's right, "the SCD2 write path" is vague and naming the aux-table creation would make the comment durable.
Separately: is it intended that the same user mistake surfaces two different error conditions depending on SCD type? Between this test and the SCD1 one we're now locking both in.
3 participants
What changes were proposed in this pull request?
This adds end-to-end (graph-execution) test coverage for AutoCDC SCD Type 2 flows, bringing it up to parity with the existing SCD Type 1 suites. SCD2 was wired end-to-end in SPARK-58321 but had only two E2E suites (single-pipeline and auxiliary-table-spec), whereas SCD1 has a broad set. This PR adds six SCD2 suites mirroring their SCD1 analogs:
AutoCdcScd2FullRefreshSuiteAutoCdcScd2TargetTableDurabilitySuiteAutoCdcScd2MultiPipelineSuiteKEY_SCHEMA_DRIFT.AutoCdcScd2AuxiliaryTableDurabilitySuiteAutoCdcScd2KeyDriftSuiteAutoCdcScd2SchemaEvolutionSuiteThese are test-only additions; there is no production code change.
Where SCD2 behaves differently from SCD1, the tests assert (and document) the SCD2 semantics rather than assuming parity:
The narrowing / dropped-column cases (a microbatch narrower than the evolved target, including dropped nested struct/array fields) are intentionally not duplicated here — they are owned by
AutoCdcScd2ColumnEvolutionSuiteunder SPARK-58418, which makes them reconcile correctly.Why are the changes needed?
SCD2 AutoCDC shipped end-to-end (SPARK-58321) with much thinner test coverage than SCD1: only single-pipeline and aux-spec suites, versus SCD1's full-refresh, durability, multi-pipeline, key-drift, and schema-evolution suites. That left full-refresh semantics, cross-pipeline behavior, key-drift validation, auxiliary-table durability, and schema evolution unverified for SCD2. These suites close that gap and lock in the SCD2 semantics against regression.
Does this PR introduce any user-facing change?
No. Test-only.
How was this patch tested?
This is tests. All six suites pass locally (40 tests total), alongside the existing SCD1 and SCD2 suites, plus scalastyle on the module.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)