[
https://issues.apache.org/jira/browse/SPARK-58452?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andreas Neumann updated SPARK-58452:
------------------------------------
Description:
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.
was:
Currently, SCD Type 2 supports only one type of schema evolution: adding a
non-key, non-tracking column. All other types of evolution fail:
# removing a column from the input schema
# adding a new column as a tracking column
# removing a tracking column from the column selection
# removing a column from the tracked column set
This is troublesome, especially because all of these can be triggered
implicitly by a change in the input schema, if column selection or tracking
column selection not configured (that is, defaults to all columns), or
configured as `* EXCEPT ...`.
This is not a problem in SCD Type 1, and SCD Type 2 should match this
capability.
> Support schema 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]
