Uh oh!
There was an error while loading. Please reload this page.
[SPARK-58409][SDP] Add SCD2 AutoCDC end-to-end test suites - #57689
Conversation
uros-b
commented
Jul 31, 2026
Thank you @anew! |
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.
| // 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. | ||
| // source `Value` are treated as distinct and the target is evolved to carry both columns. |
There was a problem hiding this comment.
Claude explains this carefully here. But this is a real bug, pre-existing in SDP and unrelated to AutoCDC, it happens for all tables. Filed SPARK-58517 for this.
There was a problem hiding this comment.
And the corresponding SCD1 test also merely asserts the bug and not the correct bahavior!
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
- Add an aux-sole-holder durability test (delete-only run leaves a tombstone only in the aux; a later lower-sequence upsert lands closed) and soften the transparently-recreated comment so the aux is not read as disposable. - Add an SCD_TYPE_DRIFT end-to-end test (recorded SCD1 aux, SCD2 flow), which the SCD1 suite structurally cannot cover. - Add an additive array<struct> nested-field evolution test for SCD2, and note in the header that trackHistorySelection changes and SCD1's extra-target-column preservation are covered elsewhere / have no SCD2 analog. - Correct the case-only-rename comment: COLUMN_ALREADY_EXISTS is raised by ResolveUnion over the reconcile unionByName, not at aux-table creation. Co-authored-by: Opus 4.8
The COLUMN_ALREADY_EXISTS case is fallout of a case-insensitivity bug in DatasetManager's schema evolution (SPARK-58517), not intended behavior. Reword the comment to flag it as characterization of known-buggy behavior. Co-authored-by: Opus 4.8
Adapt the suites to master changes that merged after they were written: - SPARK-58391 records the SCD2 track-history set on the auxiliary table and rejects a change to it as TRACK_HISTORY_DRIFT. That set is selection-derived, so additive column evolution under default tracking now widens it and fails. Pin trackHistorySelection to the columns present in every run in the additive and multi-pipeline tests, so each stays on the axis it names, and add a test asserting the drift rejection itself. - SPARK-58517 fixed case-insensitive schema evolution, so the case-only rename test now folds the source column onto the target instead of failing with COLUMN_ALREADY_EXISTS; rewrite it to assert the fold (its own comment had flagged this flip). - AutoCdcAuxiliaryTable.parseKeyColumnNames was renamed to parseColumnNames. Co-authored-by: Opus 4.8
6dfe798 to
e23dcccCompareszehon-ho
left a comment
There was a problem hiding this comment.
Re-reviewed at e23dccc. Everything from the previous round is addressed -- the aux-sole-holder test, the SCD_TYPE_DRIFT test, the additive array<struct> case, the trackHistorySelection deferral note, and the case-only-rename test now asserting the behavior fixed by SPARK-58517.
What's left is mostly one category: three comments make claims about SCD2 behavior that the production code contradicts, plus one test that doesn't exercise the state its name promises. Details inline.
One thing that isn't inline-able: the test table in the PR description is stale. It lists 3/4/5/7/13/8 = 40; the suites now hold 3/4/5/8/14/10 = 44, from the tests added across the review rounds.
| // t_b: aux retained, so the late seq=5 event is woven in as a closed prior record | ||
| // (endAt=10), and the pre-existing seq=10 record remains the open current record. |
There was a problem hiding this comment.
Could we make t_b's run #1 delete-only, so the retained aux is the sole holder of the state? Then the target stays empty, the aux holds the seq=10 tombstone, and the seq=5 event landing closed at 10 is only possible because the selective refresh spared t_b's aux -- which is what the test name claims.
As written, run #1's open upsert is neither a tombstone nor a no-op continuation (it has no successor), so identifyAndTagAuxRows routes it to the target and leaves t_b's aux empty. In run #2, findAffectedRowsFromTargetTable picks that row up because its __END_AT is null, so seq=5 is woven in from target state alone -- the assertion would hold identically had the aux been wiped. This comment attributes the outcome to the wrong table.
Same shape as the aux-drop concern that the new aux-sole-holder test fixed.
Lesser and related: the first test in this suite asserts nothing about the aux after the refresh either, but there the aux is empty either way, so there is nothing meaningful to assert -- the second test is what proves the wipe.
There was a problem hiding this comment.
ok, changed it to start with a delete-only
| // `name` is the only non-key data column and it is absent in run #1, so an empty tracked | ||
| // set is what keeps the set stable across the two runs (see the note in the preceding | ||
| // test). Every upsert then opens a new record on sequencing alone. | ||
| trackHistorySelection = Option(ColumnSelection.IncludeColumns(Seq.empty))) |
There was a problem hiding this comment.
Pinning to IncludeColumns(Seq("version")) instead of the empty set would satisfy the same constraint this comment is reaching for -- version is an eligible non-key column present in both runs, so the tracked set stays stable -- while keeping the semantics the comment actually describes.
"Every upsert then opens a new record on sequencing alone" is the opposite of what an empty tracked set does. In reconcileStartAndEndAt, an empty set collapses areTrackedColumnsEqual to lit(true), so every consecutive upsert pair becomes a no-op continuation and coalesces into one run; the scaladoc there calls this "the correct degenerate behavior when the user tracks nothing." The test passes because its two runs use different keys, so no continuation ever forms -- but anyone adding a same-key event would be steered wrong.
There was a problem hiding this comment.
The empty set of racking columns is indeed degenerate. I wonder whether that should even be legal. Changed this test to use the fixed singleton.
| // Schema evolution appends the framework columns after the user columns in the flow's output | ||
| // order (metadata, then the interval bounds), which differs from the declared order of a | ||
| // pre-created SCD2 target. Assert by name so the row matches regardless of physical order. |
There was a problem hiding this comment.
Could we drop this ordering claim, or correct it to __START_AT, __END_AT, then metadata -- which is the same as the declared order, not different from it?
AutoCdcMergeFlow.schema's Type2 branch emits __START_AT, __END_AT, then the metadata column, and DatasetManager merges via SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema) -> StructType.merge, which is left-biased and appends new fields in right-hand order. So the evolved order matches what scd2MetadataDdl declares, and both halves of the comment are off. The select-by-name keeps the assertion order-independent, so only the comment is affected.
(The analogous comments about email and age landing after the framework columns, in the schema-evolution and multi-pipeline suites, are correct.)
| * Two SCD1 evolution cases have no SCD2 analog and so are absent here by design: "extra columns on | ||
| * the target that the AutoCDC flow does not emit are preserved" relies on SCD1's in-place overwrite | ||
| * (an SCD2 upsert instead reads unemitted target columns as NULL onto the newly-opened record), and | ||
| * there is no SCD2-specific preservation invariant to assert. |
There was a problem hiding this comment.
Either name the second case or change the count to one. Walking AutoCdcScd1SchemaEvolutionSuite's fourteen tests against this suite's ten plus the four deferred to AutoCdcScd2ColumnEvolutionSuite leaves exactly one unaccounted for. Only "extra columns on the target that the AutoCDC flow does not emit are preserved" is named here; the clause after it is that case's justification, not a second case.
| stream.addData((1, "alice", 10L, true)) | ||
| runPipeline(buildCtx()) | ||
| checkAnswer(spark.table(s"$catalog.$namespace.target"), Seq.empty) | ||
| // The tombstone lives only in the aux (marker column set), with no matching visible target row. |
There was a problem hiding this comment.
"(marker column set)" reads as though the tombstone carries a set marker, which would mean the opposite of what the test needs -- worth rewording. mergeRowsIntoAuxiliaryTable writes live aux rows with deletedByBatchId = null; a non-null value is what marks a row logically deleted.
| * Key-drift validation itself is SCD-type-agnostic (it compares recorded vs declared key | ||
| * (name, dataType) sets), so these mirror the SCD1 cases with SCD2 flows. The tests that |
There was a problem hiding this comment.
A line noting that the pipeline-case-sensitivity case is deliberately skipped would keep this parity claim precise. AutoCdcScd1KeyDriftSuite's "AutoCDC key drift validation uses pipeline case sensitivity, not session default" -- which pins that the validator reads the pipeline's conf rather than the ambient session conf -- has no mirror here. Both suites land at fourteen tests because that one is dropped and SCD_TYPE_DRIFT is added. Skipping it is defensible for exactly the reason stated here (validation is SCD-type-agnostic); it's only the unqualified "these mirror the SCD1 cases" that overshoots.
| * Tests covering SCD Type 2 AutoCDC's behavior when the target table is pre-populated by something | ||
| * other than a prior AutoCDC run: hand-loaded open ("current") records and a target created | ||
| * without the framework columns. These verify AutoCDC interoperates gracefully with users who | ||
| * hand-populate the target. The SCD2 analog of [[AutoCdcScd1TargetTableDurabilitySuite]]. |
There was a problem hiding this comment.
Consider adding a pre-loaded closed record here, or confirming it was deliberately left out. That's the interop shape unique to SCD2, and the one that reaches decomposeOutOfOrderRows with a target row that has no aux counterpart; the suite covers only hand-loaded open records today.
The outcome is deterministic: pre-load [__START_AT=5, __END_AT=20) and feed an event at 10, and you get a head/tail split at 5, the event closing at 20, the tail dropping as redundant against it, leaving [5,10) and [10,20).
AutoCdcOutOfOrderConvergenceSuite reaches bisection randomly, but only through AutoCDC-written state -- which is precisely what this suite exists to sidestep.
… test coverage Address szehon-ho's review comments: - FullRefresh selective-refresh test: make t_b's run apache#1 delete-only so the retained aux is the sole holder of the seq=10 state, proving the selective refresh spared t_b's aux (previously the outcome held even if the aux were wiped). - TargetTableDurability: add the SCD2-unique pre-loaded closed-record bisection case (a target row with no aux counterpart split by an in-interval event). - SchemaEvolution: pin trackHistorySelection to `version` (an empty set tracks nothing, the opposite of the comment's intent) and correct the comment. - Fix comments: framework-column evolved order equals the declared order; the one (not two) absent SCD1 case; the aux tombstone's null deleted-by-batch-id marker; and note KeyDrift's deliberately-skipped pipeline-case-sensitivity case and the SCD_TYPE_DRIFT swap.
anew
commented
Aug 11, 2026
@szehon-ho I addressed your comments and updated the PR description. |
szehon-ho
left a comment
There was a problem hiding this comment.
Re-reviewed at 63adec2. Everything from the last round is fixed, and I traced the two behavioral changes -- t_b's delete-only run #1 and the new closed-record bisection test -- through Scd2BatchProcessor to confirm the asserted rows are what actually comes out. CI is green on this commit.
Two things left, both inline: the explanatory comment on the new bisection test describes a mechanism that doesn't match the code, and a suggestion about testImplicits in the new suites.
| // [__START_AT=5, __END_AT=20); feed an event at seq=10, strictly inside that interval. The | ||
| // event bisects the pre-existing record into a head [5, 10) and a tail [10, 20): the head | ||
| // closes where the incoming event opens, the incoming event closes where the tail begins, and | ||
| // the redundant closed upsert the event itself would form drops out against the tail. |
There was a problem hiding this comment.
The outcome this describes is right -- you do end up with [5, 10) and [10, 20) -- but the mechanism isn't what runs.
decomposeOutOfOrderRows splits the pre-loaded row into a head [5, null) and a synthetic tail [null, 20) carrying a null recordStartAt, not into [5, 10) and [10, 20); those are the post-reconciliation intervals, and [10, 20) is the incoming event's own row rather than the tail. The tail has no startAt at all, so "the incoming event closes where the tail begins" isn't describing anything -- the event closes at 20, which is the boundary the tail carries, reached via next.effectiveRecordStartAt falling back to endAt. And the drop goes the other way: dropLeftoverDeletesPostReconciliation drops the tail, because its immediately preceding upsert (the event, now [10, 20)) closes exactly on the tail's boundary. The event's row survives and is written to the target, as the expected rows below show.
Rather than correct it, I'd just drop the internals. The observable behavior needs no head/tail vocabulary, and Scd2BatchProcessor's scaladoc already documents decomposition in full -- restating it here is a second copy that can drift, which is how this one ended up inverted. Something like:
// The interop shape unique to SCD2: a hand-loaded *closed* record -- a target row with no aux// counterpart -- split by an event landing inside its interval. Pre-load [5, 20) and feed an// event at seq=10. The pre-existing record is closed early, at 10, and the incoming event// takes over the remainder of the span, [10, 20), so the two records partition the original// interval with no row left open.| val session = spark | ||
| import session.implicits._ |
There was a problem hiding this comment.
Could the new suites use a single class-level import testImplicits._ instead of repeating val session = spark / import session.implicits._ in every test? QueryTest provides testImplicits bound to self.spark, and AutoCdcScd2KeyDriftSuite in this PR already does exactly this -- same MemoryStream[(Int, String, Long)] construction, same $"version" -- so it's proven in this context.
| val session = spark | ||
| import session.implicits._ |
There was a problem hiding this comment.
Same as the note on AutoCdcScd2SchemaEvolutionSuite: a single class-level import testImplicits._ would replace the per-test val session = spark / import session.implicits._ here too.
| val session = spark | ||
| import session.implicits._ |
There was a problem hiding this comment.
Same as the note on AutoCdcScd2SchemaEvolutionSuite: a single class-level import testImplicits._ would replace the per-test val session = spark / import session.implicits._ here too.
| val session = spark | ||
| import session.implicits._ |
There was a problem hiding this comment.
Same as the note on AutoCdcScd2SchemaEvolutionSuite: a single class-level import testImplicits._ would replace the per-test val session = spark / import session.implicits._ here too.
| val session = spark | ||
| import session.implicits._ |
There was a problem hiding this comment.
Same as the note on AutoCdcScd2SchemaEvolutionSuite: a single class-level import testImplicits._ would replace the per-test val session = spark / import session.implicits._ here too.
…isection comment - Replace the per-test `val session = spark; import session.implicits._` in the five SCD2 E2E suites with a single class-level `import testImplicits._`, matching AutoCdcScd2KeyDriftSuite. - Rewrite the closed-record bisection test comment in the target-durability suite to describe the observable outcome instead of the decomposition internals, which the previous wording stated inaccurately.
### 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: | Suite | Tests | Covers | |---|---|---| | `AutoCdcScd2FullRefreshSuite` | 3 | Full refresh wipes the target and the (richer) SCD2 auxiliary table; selective refresh leaves non-refreshed targets intact. | | `AutoCdcScd2TargetTableDurabilitySuite` | 5 | Interop with a hand-populated target: pre-loaded open records closed/opened by later upserts, lower-sequence weaving, closed-record bisection, lazy aux-table creation, framework-column auto-add. | | `AutoCdcScd2MultiPipelineSuite` | 5 | Independent target/aux tables across pipelines, a downstream MV reading an SCD2 target, a shared target written by two pipelines, cross-pipeline schema evolution, and `KEY_SCHEMA_DRIFT`. | | `AutoCdcScd2AuxiliaryTableDurabilitySuite` | 8 | Recorded per-key history persists across runs; dry run does not provision the aux table; a dropped aux table is transparently recreated. | | `AutoCdcScd2KeyDriftSuite` | 14 | Key add/drop/swap, dataType drift, order/backtick/nullability invariance, case-sensitive vs case-insensitive resolution, and tampered-aux failure modes. | | `AutoCdcScd2SchemaEvolutionSuite` | 10 | Nullable-column merge, new top-level column, additive target-column evolution, broadening the column selection, a new field inside an `array<struct>`, incompatible type changes, track-history drift rejection under default tracking, and a case-only source-column rename folding onto the target. | These 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: - an SCD2 upsert to an existing key does not overwrite the row — it closes the prior record and opens a new one — so the assertions carry the full interval history; - the SCD2 auxiliary table mirrors the full target row schema (plus the aux-only deleted-by-batch-id marker), so non-key additive evolution also extends the aux table, unlike SCD1's keys-only aux. - the SCD2 track-history column set is recorded on the aux table and drift-checked across runs (SPARK-58391): because it is selection-derived, additive evolution under default tracking widens it and is rejected as `AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT`, so the additive-evolution tests pin `trackHistorySelection` and a dedicated test asserts the rejection. 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 `AutoCdcScd2ColumnEvolutionSuite` under 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 (45 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) Closes#57689 from anew/spark-58409-scd2-e2e-test-suites. Authored-by: Andreas Neumann <anew@apache.org> Signed-off-by: Szehon Ho <szehon.apache@gmail.com> (cherry picked from commit 297a730) Signed-off-by: Szehon Ho <szehon.apache@gmail.com>
### 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: | Suite | Tests | Covers | |---|---|---| | `AutoCdcScd2FullRefreshSuite` | 3 | Full refresh wipes the target and the (richer) SCD2 auxiliary table; selective refresh leaves non-refreshed targets intact. | | `AutoCdcScd2TargetTableDurabilitySuite` | 5 | Interop with a hand-populated target: pre-loaded open records closed/opened by later upserts, lower-sequence weaving, closed-record bisection, lazy aux-table creation, framework-column auto-add. | | `AutoCdcScd2MultiPipelineSuite` | 5 | Independent target/aux tables across pipelines, a downstream MV reading an SCD2 target, a shared target written by two pipelines, cross-pipeline schema evolution, and `KEY_SCHEMA_DRIFT`. | | `AutoCdcScd2AuxiliaryTableDurabilitySuite` | 8 | Recorded per-key history persists across runs; dry run does not provision the aux table; a dropped aux table is transparently recreated. | | `AutoCdcScd2KeyDriftSuite` | 14 | Key add/drop/swap, dataType drift, order/backtick/nullability invariance, case-sensitive vs case-insensitive resolution, and tampered-aux failure modes. | | `AutoCdcScd2SchemaEvolutionSuite` | 10 | Nullable-column merge, new top-level column, additive target-column evolution, broadening the column selection, a new field inside an `array<struct>`, incompatible type changes, track-history drift rejection under default tracking, and a case-only source-column rename folding onto the target. | These 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: - an SCD2 upsert to an existing key does not overwrite the row — it closes the prior record and opens a new one — so the assertions carry the full interval history; - the SCD2 auxiliary table mirrors the full target row schema (plus the aux-only deleted-by-batch-id marker), so non-key additive evolution also extends the aux table, unlike SCD1's keys-only aux. - the SCD2 track-history column set is recorded on the aux table and drift-checked across runs (SPARK-58391): because it is selection-derived, additive evolution under default tracking widens it and is rejected as `AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT`, so the additive-evolution tests pin `trackHistorySelection` and a dedicated test asserts the rejection. 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 `AutoCdcScd2ColumnEvolutionSuite` under 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 (45 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) Closes#57689 from anew/spark-58409-scd2-e2e-test-suites. Authored-by: Andreas Neumann <anew@apache.org> Signed-off-by: Szehon Ho <szehon.apache@gmail.com> (cherry picked from commit 297a730) Signed-off-by: Szehon Ho <szehon.apache@gmail.com>
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.AutoCdcScd2AuxiliaryTableDurabilitySuiteAutoCdcScd2KeyDriftSuiteAutoCdcScd2SchemaEvolutionSuitearray<struct>, incompatible type changes, track-history drift rejection under default tracking, and a case-only source-column rename folding onto the target.These 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:
AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT, so the additive-evolution tests pintrackHistorySelectionand a dedicated test asserts the rejection.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 (45 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)