[
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,
and they are very different in difficulty:
# *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), WITHOUT changing the
effective tracked-history column set. Tracked as SPARK-58418.
# *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 following an input-schema change).
All statements below were verified empirically against master by prototyping
and observing the resulting target rows (scratch suites, not committed).
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. History below that point is normally never
revisited -- unless a late event drags it into the window.
h2. Part 1 -- Column-schema evolution (SPARK-58418): small, ship first
*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.
*Fix:* pass {{allowMissingColumns = true}} to the two {{unionByName}} calls.
Verified this fixes both the top-level and nested cases (the flag recurses into
structs/arrays; map types remain unsupported), preserves historical values (the
closing UPDATE draws from the target read, not the null-padded microbatch), and
is safe through the aux round-trip. No changes needed in the merges, the aux
schema (derived from the evolved target and evolved with
{{mergeWithExistingSchema = true}}), or the reconciliation transforms.
This half is a one-line production change plus tests, and resolves SPARK-58418.
It covers only column evolution that leaves the tracked set unchanged; PR A
must NOT make the tracked set evolve (that is Part 2 and requires the guard
below, or it corrupts history -- see Findings).
h2. Part 2 -- Tracked-column-set change: allow only via full refresh
Allowing the tracked set to change and letting the stream continue
incrementally does *not* preserve correctness. Verified:
* *Finding A (in-order):* changing the tracked set produces a temporal
patchwork -- records below the change keep the old segmentation, records from
the change onward use the new one.
* *Finding B (late events):* a late-arriving event drops {{minSeq}} backward,
dragging pre-change history into the affected window, where it is re-diced
under the current tracked set. Narrowing then silently *destroys* pre-change
records (they coalesce); widening *fabricates* records (retroactive splits),
both with sequences below the change point. Reproduced in both directions,
including the implicit case where a default-tracked column leaves the input
schema and the effective set evolves to a subset.
* *Finding C (blast radius):* the corruption is confined to keys present in the
triggering microbatch (the affected-row scans inner-join on the key against the
per-key {{minSeq}}), and reaches back to that key's batch-min. A key that stays
quiet keeps its old segmentation.
Because incremental tracked-set changes are unsafe under the bounded-window
design, the chosen approach is to *disallow* them incrementally and require a
full refresh, which starts from a clean target and aux table so there is no
pre-change history to corrupt. Two rules:
# *The effective tracked set is derived from the column selection.* Default =
all selected non-key/non-framework columns; {{* EXCEPT (...)}} = selected minus
the exclusions; explicit {{TRACK HISTORY ON (...)}} = the enumerated list. So
dropping a column from the selection (default / {{* EXCEPT}} modes) correctly
removes it from the tracked set -- it is not kept tracked just because it
lingers in the (sticky) target schema.
# *A guard requires a full refresh whenever the effective tracked set changes*,
however it was configured. This turns claims 2, 3, and 4 -- and claim 1 when
the removed column was tracked -- into a clear, actionable
{{AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT}} error (remedy: full refresh)
instead of a crash or silent corruption.
This needs no change watermark and no changes to {{reconcileStartAndEndAt}}; it
is a validation-plus-computation change.
h3. Reuse of SPARK-58391
SPARK-58391 (open PR apache/spark#57625) already implements exactly this guard:
it records the tracked set as the auxiliary-table property
{{pipelines.autocdc.trackHistoryColumnNames}} and rejects a changed set with
{{AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT}}. Part 2 reuses it directly, with
*one required change*:
* SPARK-58391 currently computes the recorded/compared tracked set from the
*evolved target schema* ({{buildScd2AuxiliaryTableSpecFor}} passes
{{targetTableSchema}} to {{Scd2BatchProcessor.computeTrackedHistoryColumns}}).
Because the target schema keeps columns that have been dropped from the input,
this would fail to detect the *implicit* evolution cases (default / {{*
EXCEPT}} tracking where an input column disappears): the recorded and current
sets would both still contain the dropped column, no drift is detected, and the
run proceeds into the corruption of Finding B.
* Rule (1) requires deriving the tracked set from the *column-selection-based
(user-selected source) schema* instead -- the same {{userSelectedSchema}} that
the construction-time guard
{{Flow.requireTrackHistoryColumnsResolvableInSelectedSchema}} already uses --
at both the recording site and the comparison, so implicit tracked-set changes
are detected and correctly demand a full refresh.
So the direction for apache/spark#57625 is to land (or be amended) with the
selection-derived tracked set, at which point it directly becomes the Part 2
guard -- avoiding an add-then-relax churn.
h2. Claim-by-claim outcome under this design
* *Claim 1 (remove a column from the input schema):* incremental via the Part 1
union fix if the column was not tracked; full refresh if it was tracked
(default / {{* EXCEPT}}).
* *Claim 2 (add a new column as a tracking column):* full refresh.
* *Claim 3 (remove a tracking column from the column selection):* full refresh
(default / {{* EXCEPT}}); already rejected at construction if the column is
explicitly tracked but dropped from the selection.
* *Claim 4 (remove a column from the tracked set):* full refresh.
Only column-schema evolution that leaves the tracked set unchanged stays
incremental; every tracked-set change surfaces a clear full-refresh error
rather than the current crash/corruption.
h2. Proposed sequencing
# *PR A -- SPARK-58418:* the {{allowMissingColumns}} union fix. Independent,
small, ship first. Must not evolve the tracked set.
# *PR B -- this ticket:* rules (1)+(2), built on apache/spark#57625
(SPARK-58391) with the selection-derived-tracked-set change and tests for the
implicit evolution cases. No watermark.
was:
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.
> 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, and they are very different in difficulty:
> # *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), WITHOUT changing the
> effective tracked-history column set. Tracked as SPARK-58418.
> # *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 following an input-schema change).
> All statements below were verified empirically against master by prototyping
> and observing the resulting target rows (scratch suites, not committed).
> 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. History below that point is normally never
> revisited -- unless a late event drags it into the window.
> h2. Part 1 -- Column-schema evolution (SPARK-58418): small, ship first
> *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.
> *Fix:* pass {{allowMissingColumns = true}} to the two {{unionByName}} calls.
> Verified this fixes both the top-level and nested cases (the flag recurses
> into structs/arrays; map types remain unsupported), preserves historical
> values (the closing UPDATE draws from the target read, not the null-padded
> microbatch), and is safe through the aux round-trip. No changes needed in the
> merges, the aux schema (derived from the evolved target and evolved with
> {{mergeWithExistingSchema = true}}), or the reconciliation transforms.
> This half is a one-line production change plus tests, and resolves
> SPARK-58418. It covers only column evolution that leaves the tracked set
> unchanged; PR A must NOT make the tracked set evolve (that is Part 2 and
> requires the guard below, or it corrupts history -- see Findings).
> h2. Part 2 -- Tracked-column-set change: allow only via full refresh
> Allowing the tracked set to change and letting the stream continue
> incrementally does *not* preserve correctness. Verified:
> * *Finding A (in-order):* changing the tracked set produces a temporal
> patchwork -- records below the change keep the old segmentation, records from
> the change onward use the new one.
> * *Finding B (late events):* a late-arriving event drops {{minSeq}} backward,
> dragging pre-change history into the affected window, where it is re-diced
> under the current tracked set. Narrowing then silently *destroys* pre-change
> records (they coalesce); widening *fabricates* records (retroactive splits),
> both with sequences below the change point. Reproduced in both directions,
> including the implicit case where a default-tracked column leaves the input
> schema and the effective set evolves to a subset.
> * *Finding C (blast radius):* the corruption is confined to keys present in
> the triggering microbatch (the affected-row scans inner-join on the key
> against the per-key {{minSeq}}), and reaches back to that key's batch-min. A
> key that stays quiet keeps its old segmentation.
> Because incremental tracked-set changes are unsafe under the bounded-window
> design, the chosen approach is to *disallow* them incrementally and require a
> full refresh, which starts from a clean target and aux table so there is no
> pre-change history to corrupt. Two rules:
> # *The effective tracked set is derived from the column selection.* Default =
> all selected non-key/non-framework columns; {{* EXCEPT (...)}} = selected
> minus the exclusions; explicit {{TRACK HISTORY ON (...)}} = the enumerated
> list. So dropping a column from the selection (default / {{* EXCEPT}} modes)
> correctly removes it from the tracked set -- it is not kept tracked just
> because it lingers in the (sticky) target schema.
> # *A guard requires a full refresh whenever the effective tracked set
> changes*, however it was configured. This turns claims 2, 3, and 4 -- and
> claim 1 when the removed column was tracked -- into a clear, actionable
> {{AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT}} error (remedy: full refresh)
> instead of a crash or silent corruption.
> This needs no change watermark and no changes to {{reconcileStartAndEndAt}};
> it is a validation-plus-computation change.
> h3. Reuse of SPARK-58391
> SPARK-58391 (open PR apache/spark#57625) already implements exactly this
> guard: it records the tracked set as the auxiliary-table property
> {{pipelines.autocdc.trackHistoryColumnNames}} and rejects a changed set with
> {{AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT}}. Part 2 reuses it directly,
> with *one required change*:
> * SPARK-58391 currently computes the recorded/compared tracked set from the
> *evolved target schema* ({{buildScd2AuxiliaryTableSpecFor}} passes
> {{targetTableSchema}} to
> {{Scd2BatchProcessor.computeTrackedHistoryColumns}}). Because the target
> schema keeps columns that have been dropped from the input, this would fail
> to detect the *implicit* evolution cases (default / {{* EXCEPT}} tracking
> where an input column disappears): the recorded and current sets would both
> still contain the dropped column, no drift is detected, and the run proceeds
> into the corruption of Finding B.
> * Rule (1) requires deriving the tracked set from the *column-selection-based
> (user-selected source) schema* instead -- the same {{userSelectedSchema}}
> that the construction-time guard
> {{Flow.requireTrackHistoryColumnsResolvableInSelectedSchema}} already uses --
> at both the recording site and the comparison, so implicit tracked-set
> changes are detected and correctly demand a full refresh.
> So the direction for apache/spark#57625 is to land (or be amended) with the
> selection-derived tracked set, at which point it directly becomes the Part 2
> guard -- avoiding an add-then-relax churn.
> h2. Claim-by-claim outcome under this design
> * *Claim 1 (remove a column from the input schema):* incremental via the Part
> 1 union fix if the column was not tracked; full refresh if it was tracked
> (default / {{* EXCEPT}}).
> * *Claim 2 (add a new column as a tracking column):* full refresh.
> * *Claim 3 (remove a tracking column from the column selection):* full
> refresh (default / {{* EXCEPT}}); already rejected at construction if the
> column is explicitly tracked but dropped from the selection.
> * *Claim 4 (remove a column from the tracked set):* full refresh.
> Only column-schema evolution that leaves the tracked set unchanged stays
> incremental; every tracked-set change surfaces a clear full-refresh error
> rather than the current crash/corruption.
> h2. Proposed sequencing
> # *PR A -- SPARK-58418:* the {{allowMissingColumns}} union fix. Independent,
> small, ship first. Must not evolve the tracked set.
> # *PR B -- this ticket:* rules (1)+(2), built on apache/spark#57625
> (SPARK-58391) with the selection-derived-tracked-set change and tests for the
> implicit evolution cases. No watermark.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
