[
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. Status and sequencing
# *PR A -- SPARK-58418 (DONE):* the {{allowMissingColumns}} union fix for
column-schema evolution that leaves the tracked set unchanged. Implemented and
pushed; does not evolve the tracked set.
# *PR B -- this ticket:* rules (1)+(2). *These are now delivered by
apache/spark#57625 (SPARK-58391).* That PR's latest revision derives the
tracked set from the user-selected source schema (rule 1, via a new
{{AutoCdcMergeFlow.trackHistoryColumnNames}} val) and rejects any change to it
with {{TRACK_HISTORY_DRIFT}} (rule 2), including the implicit default / {{*
EXCEPT}} cases where a source column is added or dropped. So no separate
watermark work and no add-then-relax churn: once #57625 merges, the
tracked-set-change semantics for this ticket are in place, and this ticket's
remaining scope is documentation / any additional end-to-end coverage on top of
it.
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,
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.
> 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. Status and sequencing
> # *PR A -- SPARK-58418 (DONE):* the {{allowMissingColumns}} union fix for
> column-schema evolution that leaves the tracked set unchanged. Implemented
> and pushed; does not evolve the tracked set.
> # *PR B -- this ticket:* rules (1)+(2). *These are now delivered by
> apache/spark#57625 (SPARK-58391).* That PR's latest revision derives the
> tracked set from the user-selected source schema (rule 1, via a new
> {{AutoCdcMergeFlow.trackHistoryColumnNames}} val) and rejects any change to
> it with {{TRACK_HISTORY_DRIFT}} (rule 2), including the implicit default /
> {{* EXCEPT}} cases where a source column is added or dropped. So no separate
> watermark work and no add-then-relax churn: once #57625 merges, the
> tracked-set-change semantics for this ticket are in place, and this ticket's
> remaining scope is documentation / any additional end-to-end coverage on top
> of it.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
