[
https://issues.apache.org/jira/browse/SPARK-58478?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andreas Neumann updated SPARK-58478:
------------------------------------
Description:
h2. Summary
SCD Type 1 AutoCDC is internally inconsistent about subtractive schema
evolution across runs:
* Dropping a *top-level* column from the flow's source (or narrowing the
{{COLUMNS}} selection) is *accepted*: the column is preserved on
already-written rows and set to {{NULL}} on newly-inserted rows. Covered by
{{AutoCdcScd1SchemaEvolutionSuite}} ("a top-level column dropped from the
source DF between runs is preserved on existing rows and left NULL on new
rows").
* Dropping a *nested* struct field, or a field inside an {{array<struct>}}
element, is *rejected* with {{INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA}}
(same suite, "dropping a nested struct field between runs fails with
INCOMPATIBLE_DATA_FOR_TABLE" and the {{array<struct>}} analog).
So the same logical operation -- the flow stops emitting a column that still
exists on the target -- succeeds at the top level but fails one level down.
h2. Why
This is not a deliberate "reject subtractive evolution" policy; it is an
artifact of where the failure surfaces in the MERGE write path:
* For a dropped top-level column, the SCD1 MERGE simply omits that column from
its INSERT/UPDATE assignment maps, so nothing references the missing data and
the write succeeds.
* For a dropped nested field, the v2 writer's {{TableOutputResolver}} walks
into the target's struct and cannot find data for the nested path (e.g.
{{value.b.c}}), so it throws {{CANNOT_FIND_DATA}}.
The nested case is stricter purely because the resolver descends into nested
paths while the top-level assignment map does not.
h2. Contrast with SCD2
SPARK-58418 makes SCD2 handle both cases consistently:
{{Scd2ForeachBatchHandler}} uses {{unionByName(..., allowMissingColumns =
true)}}, which pads a missing column -- top-level or nested (recursing into
structs/arrays) -- with {{NULL}} before the union/MERGE. As a result SCD2
preserves the field on existing records and writes {{NULL}} on new ones for a
nested drop too, matching its own top-level behavior. SCD1 does not have this
padding on its MERGE source, so it retains the nested-drop failure.
SPARK-58418's tests document this as a deliberate SCD2-vs-SCD1 divergence
({{AutoCdcScd2ColumnEvolutionSuite}}).
h2. Proposal
Make SCD1 accept a dropped nested struct / array-element field the same
additive-tolerant way it already accepts a dropped top-level column (preserve
on existing rows, {{NULL}} on new rows), so its subtractive-evolution behavior
is consistent across nesting levels and matches SCD2. Likely approach: pad the
MERGE source with the missing nested fields (analogous to SCD2's
{{allowMissingColumns}}) before the write, rather than letting the resolver
reject it.
When this is fixed, the two "fails with INCOMPATIBLE_DATA_FOR_TABLE" tests in
{{AutoCdcScd1SchemaEvolutionSuite}} should be converted to assert the
preserve/NULL behavior, and the SCD2-vs-SCD1 divergence note in
{{AutoCdcScd2ColumnEvolutionSuite}} can be removed.
Not a regression and not blocking; captured for consistency while implementing
SPARK-58418.
h2. Implementation note (investigated 2026-07-31)
The fix is more involved than "pad the MERGE source's missing nested fields",
because SCD1's target MERGE does an in-place UPDATE, and its top-level
"preserve on update" semantics must be matched at the nested level too.
Verified behavior:
* SCD1, dropped *top-level* column, on UPDATE of an existing key: the old value
is *preserved* (not nulled). This works because {{mergeMicrobatchOntoTarget}}
builds its UPDATE assignment map from {{microbatchDf.columns}} and simply omits
the dropped top-level column, leaving the target column untouched. NULL only
appears on INSERT of a brand-new key.
So parity for a nested drop is "preserve the dropped nested field on UPDATE,
NULL on INSERT" -- not uniform NULL padding. A simple
{{unionByName(allowMissingColumns = true)}} pad (as SCD2 uses) is *wrong* for
SCD1's UPDATE path: the top-level column (e.g. {{value}}) still exists and is
in the UPDATE assignment map, so SETing it to a struct padded with {{c = null}}
would overwrite the target's real {{value.b.c}} on every update -- data loss,
not parity.
Correct fix therefore needs a recursive deep struct-merge in the UPDATE
assignment: for a structurally-narrowed column, set it to a struct that takes
the fields the microbatch provides and carries the dropped nested field over
from the target row (coalesce-style, recursing through nested structs and
array-of-struct elements), while INSERT still pads with NULL. This is a
non-trivial expression builder, not a one-liner.
(SCD2 avoids all of this because an upsert never UPDATEs a data row in place:
it closes the prior record -- which keeps its real values from the target read
-- and inserts a fresh NULL-padded record. SCD1's in-place UPDATE is the entire
difficulty.)
Paused pending prioritization; the SCD2 side (SPARK-58418) is unaffected.
was:
h2. Summary
SCD Type 1 AutoCDC is internally inconsistent about subtractive schema
evolution across runs:
* Dropping a *top-level* column from the flow's source (or narrowing the
{{COLUMNS}} selection) is *accepted*: the column is preserved on
already-written rows and set to {{NULL}} on newly-inserted rows. Covered by
{{AutoCdcScd1SchemaEvolutionSuite}} ("a top-level column dropped from the
source DF between runs is preserved on existing rows and left NULL on new
rows").
* Dropping a *nested* struct field, or a field inside an {{array<struct>}}
element, is *rejected* with {{INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA}}
(same suite, "dropping a nested struct field between runs fails with
INCOMPATIBLE_DATA_FOR_TABLE" and the {{array<struct>}} analog).
So the same logical operation -- the flow stops emitting a column that still
exists on the target -- succeeds at the top level but fails one level down.
h2. Why
This is not a deliberate "reject subtractive evolution" policy; it is an
artifact of where the failure surfaces in the MERGE write path:
* For a dropped top-level column, the SCD1 MERGE simply omits that column from
its INSERT/UPDATE assignment maps, so nothing references the missing data and
the write succeeds.
* For a dropped nested field, the v2 writer's {{TableOutputResolver}} walks
into the target's struct and cannot find data for the nested path (e.g.
{{value.b.c}}), so it throws {{CANNOT_FIND_DATA}}.
The nested case is stricter purely because the resolver descends into nested
paths while the top-level assignment map does not.
h2. Contrast with SCD2
SPARK-58418 makes SCD2 handle both cases consistently:
{{Scd2ForeachBatchHandler}} uses {{unionByName(..., allowMissingColumns =
true)}}, which pads a missing column -- top-level or nested (recursing into
structs/arrays) -- with {{NULL}} before the union/MERGE. As a result SCD2
preserves the field on existing records and writes {{NULL}} on new ones for a
nested drop too, matching its own top-level behavior. SCD1 does not have this
padding on its MERGE source, so it retains the nested-drop failure.
SPARK-58418's tests document this as a deliberate SCD2-vs-SCD1 divergence
({{AutoCdcScd2ColumnEvolutionSuite}}).
h2. Proposal
Make SCD1 accept a dropped nested struct / array-element field the same
additive-tolerant way it already accepts a dropped top-level column (preserve
on existing rows, {{NULL}} on new rows), so its subtractive-evolution behavior
is consistent across nesting levels and matches SCD2. Likely approach: pad the
MERGE source with the missing nested fields (analogous to SCD2's
{{allowMissingColumns}}) before the write, rather than letting the resolver
reject it.
When this is fixed, the two "fails with INCOMPATIBLE_DATA_FOR_TABLE" tests in
{{AutoCdcScd1SchemaEvolutionSuite}} should be converted to assert the
preserve/NULL behavior, and the SCD2-vs-SCD1 divergence note in
{{AutoCdcScd2ColumnEvolutionSuite}} can be removed.
Not a regression and not blocking; captured for consistency while implementing
SPARK-58418.
> SCD1 AutoCDC rejects dropping a nested struct/array field but accepts
> dropping a top-level column
> -------------------------------------------------------------------------------------------------
>
> Key: SPARK-58478
> URL: https://issues.apache.org/jira/browse/SPARK-58478
> Project: Spark
> Issue Type: Sub-task
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Andreas Neumann
> Priority: Major
>
> h2. Summary
> SCD Type 1 AutoCDC is internally inconsistent about subtractive schema
> evolution across runs:
> * Dropping a *top-level* column from the flow's source (or narrowing the
> {{COLUMNS}} selection) is *accepted*: the column is preserved on
> already-written rows and set to {{NULL}} on newly-inserted rows. Covered by
> {{AutoCdcScd1SchemaEvolutionSuite}} ("a top-level column dropped from the
> source DF between runs is preserved on existing rows and left NULL on new
> rows").
> * Dropping a *nested* struct field, or a field inside an {{array<struct>}}
> element, is *rejected* with {{INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA}}
> (same suite, "dropping a nested struct field between runs fails with
> INCOMPATIBLE_DATA_FOR_TABLE" and the {{array<struct>}} analog).
> So the same logical operation -- the flow stops emitting a column that still
> exists on the target -- succeeds at the top level but fails one level down.
> h2. Why
> This is not a deliberate "reject subtractive evolution" policy; it is an
> artifact of where the failure surfaces in the MERGE write path:
> * For a dropped top-level column, the SCD1 MERGE simply omits that column
> from its INSERT/UPDATE assignment maps, so nothing references the missing
> data and the write succeeds.
> * For a dropped nested field, the v2 writer's {{TableOutputResolver}} walks
> into the target's struct and cannot find data for the nested path (e.g.
> {{value.b.c}}), so it throws {{CANNOT_FIND_DATA}}.
> The nested case is stricter purely because the resolver descends into nested
> paths while the top-level assignment map does not.
> h2. Contrast with SCD2
> SPARK-58418 makes SCD2 handle both cases consistently:
> {{Scd2ForeachBatchHandler}} uses {{unionByName(..., allowMissingColumns =
> true)}}, which pads a missing column -- top-level or nested (recursing into
> structs/arrays) -- with {{NULL}} before the union/MERGE. As a result SCD2
> preserves the field on existing records and writes {{NULL}} on new ones for a
> nested drop too, matching its own top-level behavior. SCD1 does not have this
> padding on its MERGE source, so it retains the nested-drop failure.
> SPARK-58418's tests document this as a deliberate SCD2-vs-SCD1 divergence
> ({{AutoCdcScd2ColumnEvolutionSuite}}).
> h2. Proposal
> Make SCD1 accept a dropped nested struct / array-element field the same
> additive-tolerant way it already accepts a dropped top-level column (preserve
> on existing rows, {{NULL}} on new rows), so its subtractive-evolution
> behavior is consistent across nesting levels and matches SCD2. Likely
> approach: pad the MERGE source with the missing nested fields (analogous to
> SCD2's {{allowMissingColumns}}) before the write, rather than letting the
> resolver reject it.
> When this is fixed, the two "fails with INCOMPATIBLE_DATA_FOR_TABLE" tests in
> {{AutoCdcScd1SchemaEvolutionSuite}} should be converted to assert the
> preserve/NULL behavior, and the SCD2-vs-SCD1 divergence note in
> {{AutoCdcScd2ColumnEvolutionSuite}} can be removed.
> Not a regression and not blocking; captured for consistency while
> implementing SPARK-58418.
> h2. Implementation note (investigated 2026-07-31)
> The fix is more involved than "pad the MERGE source's missing nested fields",
> because SCD1's target MERGE does an in-place UPDATE, and its top-level
> "preserve on update" semantics must be matched at the nested level too.
> Verified behavior:
> * SCD1, dropped *top-level* column, on UPDATE of an existing key: the old
> value is *preserved* (not nulled). This works because
> {{mergeMicrobatchOntoTarget}} builds its UPDATE assignment map from
> {{microbatchDf.columns}} and simply omits the dropped top-level column,
> leaving the target column untouched. NULL only appears on INSERT of a
> brand-new key.
> So parity for a nested drop is "preserve the dropped nested field on UPDATE,
> NULL on INSERT" -- not uniform NULL padding. A simple
> {{unionByName(allowMissingColumns = true)}} pad (as SCD2 uses) is *wrong* for
> SCD1's UPDATE path: the top-level column (e.g. {{value}}) still exists and is
> in the UPDATE assignment map, so SETing it to a struct padded with {{c =
> null}} would overwrite the target's real {{value.b.c}} on every update --
> data loss, not parity.
> Correct fix therefore needs a recursive deep struct-merge in the UPDATE
> assignment: for a structurally-narrowed column, set it to a struct that takes
> the fields the microbatch provides and carries the dropped nested field over
> from the target row (coalesce-style, recursing through nested structs and
> array-of-struct elements), while INSERT still pads with NULL. This is a
> non-trivial expression builder, not a one-liner.
> (SCD2 avoids all of this because an upsert never UPDATEs a data row in place:
> it closes the prior record -- which keeps its real values from the target
> read -- and inserts a fresh NULL-padded record. SCD1's in-place UPDATE is the
> entire difficulty.)
> Paused pending prioritization; the SCD2 side (SPARK-58418) is unaffected.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
