[
https://issues.apache.org/jira/browse/SPARK-58452?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andreas Neumann updated SPARK-58452:
------------------------------------
Summary: Support tracking columns evolution for SCD Type 2 (was: Support
schema evolution for SCD Type 2)
> Support tracking columns evolution for SCD Type 2
> -------------------------------------------------
>
> Key: SPARK-58452
> URL: https://issues.apache.org/jira/browse/SPARK-58452
> Project: Spark
> Issue Type: Sub-task
> Components: Declarative Pipelines
> Affects Versions: 4.3.0
> Reporter: Andreas Neumann
> Priority: Major
>
> h2. Goal
> Support schema evolution for SCD Type 2 AutoCDC targets, matching the
> flexibility SCD Type 1 already has. Two independent capabilities are in scope:
> # *Column-schema evolution* -- the set of user data columns the flow emits
> changes across runs (a column is dropped from the input, a nested field is
> dropped, or the column/EXCEPT selection narrows).
> # *Tracked-column-set change* -- the set of columns that define an SCD2 "run"
> changes across runs (adding or removing a tracked column, whether explicitly
> via {{TRACK HISTORY ON}} or implicitly through the default/{{EXCEPT}}
> resolution).
> These are very different in difficulty and are analyzed separately below. All
> statements were verified empirically against master by prototyping and
> observing the resulting target rows.
> h2. Background: how the tracked set is used
> In {{Scd2BatchProcessor.reconcileStartAndEndAt}}, two
> chronologically-adjacent upsert events for the same key are coalesced into a
> single record iff they are gap-free and equal on *every tracked column*
> ({{isNoOpUpsertContinuation}}). A change in a tracked column opens a new
> record; a change in a non-tracked column is folded into the current record.
> So the tracked set is the sole determinant of where record boundaries fall.
> The reconciliation window is bounded per key:
> {{findAffectedRowsFrom{Target,Auxiliary}Table}} pull in only rows at or after
> {{minSeq}} (the minimum sequence of the *current microbatch* for that key),
> plus a single aux "anchor" row for left context. History below that point is
> normally never revisited.
> h2. Part 1 -- Column-schema evolution (JIRA claims 1 and 3): small, clean
> *Root cause of the current failure:*
> {{Scd2ForeachBatchHandler.reconcileMicrobatch}} unions the microbatch with
> the affected target/aux rows via {{unionByName}} with no
> {{allowMissingColumns}}. A microbatch narrower than the (already-evolved,
> wider) target then fails -- {{NUM_COLUMNS_MISMATCH}} for a dropped top-level
> column, {{INCOMPATIBLE_COLUMN_TYPE}} for a dropped nested struct/array field.
> This is the bug tracked as SPARK-58418. SCD1 is unaffected because it never
> unions the microbatch with target rows; it MERGEs, which tolerates a narrower
> source.
> *Fix:* pass {{allowMissingColumns = true}} to the two {{unionByName}} calls.
> Verified this:
> * fixes both the top-level and nested cases (the flag recurses into structs
> and arrays; map types remain unsupported, a documentable limitation);
> * *preserves historical values* -- the closing UPDATE of an existing record
> draws its values from the target read, not the null-padded microbatch, so
> records written before the drop keep their real values; only records opened
> after the drop carry NULL for the no-longer-emitted column (correct
> prospective semantics);
> * is safe through the aux round-trip (a no-op run pulled from the aux table
> as an anchor is not corrupted).
> No changes are needed in the two merges (they build column lists
> dynamically), the auxiliary-table schema (derived from the evolved target
> each run and evolved with {{mergeWithExistingSchema = true}}), or the
> reconciliation transforms (all name-driven). This half is essentially a
> one-line production change plus tests, and it also resolves SPARK-58418.
> h2. Part 2 -- Tracked-column-set change (JIRA claims 2 and 4): NOT safe under
> the bounded-window design
> Allowing the tracked set to change and letting the stream continue does *not*
> preserve correctness, because the per-key bounded reconciliation window
> re-segments whatever history it pulls in under the *currently configured*
> tracked set. Three findings, all verified:
> h3. Finding A -- Prospective tail rewrite (in-order events)
> With only in-order events, {{minSeq}} is always at or above everything
> already written, so the affected window only touches the tail. Changing the
> tracked set then produces a table that is a *temporal patchwork*: records
> below the change keep the old segmentation, records from the change onward
> use the new one. Boundary placement is inconsistent, though stored values are
> not corrupted. This is the "prospective" behavior one might naively accept.
> h3. Finding B -- Late events reach backward and rewrite pre-change history
> A late-arriving event drops {{minSeq}} *backwards*, dragging pre-change
> history into the affected window, where it is re-diced under the current
> tracked set. Verified both directions:
> * *Narrowing* (track color -> track a constant column), history {{[c1
> 10-20][c2 20-30][c1 30-null]}}, late event at seq 15: the three records
> collapse into a single {{[10-null]}} record. *Two historical records are
> silently destroyed*, and the survivor spans an interval no configuration ever
> produced.
> * *Widening* (track a constant column -> track color), history {{[c3
> 10-null]}} (one coalesced record), late event at seq 15: the single record is
> retroactively shredded into four color-delimited records at sequences
> 10/20/30 -- all *before* the change. *Spurious historical records are
> fabricated*, carrying sequences below the change point but the new
> segmentation.
> So the answer to "can this drop or fabricate history records with sequences
> before the change?" is *yes to both*. The rewrite window for a touched key is
> {{[minSeq(key), infinity)}}, and how far back it reaches is determined by how
> late the latest late event happens to be -- non-deterministic from the user's
> perspective. This means the "prospective, old history left as-is" model of
> Finding A is not actually achievable by relaxing validation alone.
> h3. Finding C -- Blast radius is confined to keys present in the triggering
> microbatch
> Verified that a key with no event in the microbatch that carries the changed
> tracked set is left *byte-for-byte unchanged*, even though the configuration
> change is global to the flow. This is structural:
> {{findAffectedRowsFrom{Target,Auxiliary}Table}} inner-join on the key columns
> against the per-key {{minSeq}} dataframe, which has a row only for keys
> present in the microbatch. So the re-dicing is per-key and event-triggered: a
> key's history is re-segmented only when that key next receives an event after
> the change; a quiet key keeps its old-set segmentation indefinitely.
> h3. Combined characterization
> Corruption from a tracked-set change is localized to {{(touched key,
> sequences >= that key's batch-min)}}. Findings B and C are the same "affected
> window per key" viewed along the sequence axis and the key axis respectively.
> h2. Design direction for Part 2 -- change watermark
> Because both the late-event reach (B) and the per-key re-dicing (C) are
> bounded on the *sequence* axis, a single mechanism addresses them: record the
> sequence (or batch id) at which the tracked set last changed -- a *change
> watermark* -- as auxiliary-table state, and make reconciliation refuse to
> re-segment records whose {{recordStartAt}} lies below that watermark. Sealed
> (pre-watermark) records are treated as immutable run boundaries: a late event
> below the watermark either attaches without re-dicing sealed records or is
> rejected as too-late, and the new tracked set governs only sequences at or
> above the watermark. This makes the prospective model of Finding A genuinely
> correct rather than only correct-in-the-absence-of-late-events.
> This is real algorithm work (a new persisted watermark, plus boundary logic
> in {{reconcileStartAndEndAt}} / the affected-row scan to respect it), not a
> validation relaxation. Sizing to be completed before implementation.
> h2. Proposed sequencing
> # *PR A* -- the {{allowMissingColumns}} union fix for column-schema evolution
> (claims 1 and 3). Independent, small, also closes SPARK-58418.
> # *PR B* -- tracked-column-set change (claims 2 and 4) via the
> change-watermark design. Must be reconciled with SPARK-58391 (open PR
> #57625), which adds the track-history drift validation that PR B would
> relax/replace; the two should not be landed as add-then-remove.
> h2. Verification note
> All findings above were reproduced by prototyping {{allowMissingColumns =
> true}} and observing target-table contents across two-run scenarios
> (in-order, late-event, and multi-key). The scratch suites were not committed.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
