[
https://issues.apache.org/jira/browse/SPARK-58418?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andreas Neumann updated SPARK-58418:
------------------------------------
Description:
h2. Summary
SCD Type 2 AutoCDC flows fail on every microbatch when the flow's microbatch is
_narrower_ than the target table it writes to (i.e. the target was
schema-evolved wider by an earlier, wider run). The failure is
{{NUM_COLUMNS_MISMATCH}} for a dropped top-level column and
{{INCOMPATIBLE_COLUMN_TYPE}} for a dropped nested struct / array-element field.
SCD Type 1 is unaffected.
h2. Mechanism
To reconcile an incoming microbatch against existing history,
{{Scd2ForeachBatchHandler.reconcileMicrobatch}} reads back the
potentially-affected rows from both the auxiliary table and the target table
and unions them with the microbatch into a single dataframe:
{code:scala}
val microbatchAndAffectedRows = preprocessedBatchDf
.unionByName(affectedRowsFromAuxiliaryTable)
.unionByName(affectedRowsFromTargetTable)
{code}
These are plain {{unionByName}} calls with no {{allowMissingColumns = true}}.
{{unionByName}} requires all inputs to have the same set of columns, so the
union fails as soon as the microbatch's column set diverges from what is stored
in the target/aux tables.
h2. When the column sets diverge
They diverge whenever the target has been schema-evolved to be wider than the
current microbatch:
* *Dropped top-level column* (or a narrowed {{COLUMNS}} selection): the target
still carries the column from an earlier wider run, but the new microbatch does
not produce it. The microbatch has N columns and the affected target rows have
N+1 -> {{NUM_COLUMNS_MISMATCH}}.
* *Dropped nested struct / array-element field*: same column count, but the
struct types differ (the target's struct has field {{b.c}}, the microbatch's
does not), so the union fails with {{INCOMPATIBLE_COLUMN_TYPE}}.
h2. Why SCD1 is not affected
SCD1 never unions the microbatch with existing target rows.
{{Scd1ForeachBatchHandler}} reconciles the microbatch on its own and then
issues a SQL {{MERGE}} onto the target. A {{MERGE}} tolerates a source narrower
than the target: unmentioned columns are left untouched on {{UPDATE}} and set
to {{NULL}} on {{INSERT}}. So the identical "narrow the flow after the target
has grown" scenario works under SCD1.
h2. Why it matters
This is a realistic, user-facing path -- it is exactly what cross-run schema
evolution looks like:
# Run a wide flow (target gains {{email}}).
# Later run a narrower flow, or drop a source column, or tighten {{COLUMNS}}.
# The SCD2 stream then dies on every microbatch until a full refresh.
Schema evolution is meant to be additive-tolerant (existing rows keep their old
columns; a flow can stop emitting one), and SCD1 honors that. SCD2 breaks it.
h2. Likely fix
Pass {{allowMissingColumns = true}} to those {{unionByName}} calls so a
narrower microbatch is padded with {{NULL}} for the columns it no longer
produces, matching SCD1's {{MERGE}} semantics. This needs care around the
nested-field case (union with missing nested fields) and around ensuring
downstream reconciliation treats the {{NULL}}-padded columns correctly.
h2. Test coverage
Discovered while writing the SCD2 end-to-end suites (SPARK-58409). Those suites
currently assert the actual failure and reference this ticket, so the
assertions can be flipped to the additive-preserving expectations once this is
fixed:
* {{AutoCdcScd2MultiPipelineSuite}} -- stops short of re-running the narrower
pipeline against the widened target.
* {{AutoCdcScd2SchemaEvolutionSuite}} -- has explicit tests asserting the
current {{NUM_COLUMNS_MISMATCH}} / {{INCOMPATIBLE_COLUMN_TYPE}} failures.
> SCD2 AutoCDC fails with NUM_COLUMNS_MISMATCH when a flow narrower than the
> evolved target reconciles
> ----------------------------------------------------------------------------------------------------
>
> Key: SPARK-58418
> URL: https://issues.apache.org/jira/browse/SPARK-58418
> Project: Spark
> Issue Type: Sub-task
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Andreas Neumann
> Priority: Major
>
> h2. Summary
> SCD Type 2 AutoCDC flows fail on every microbatch when the flow's microbatch
> is _narrower_ than the target table it writes to (i.e. the target was
> schema-evolved wider by an earlier, wider run). The failure is
> {{NUM_COLUMNS_MISMATCH}} for a dropped top-level column and
> {{INCOMPATIBLE_COLUMN_TYPE}} for a dropped nested struct / array-element
> field. SCD Type 1 is unaffected.
> h2. Mechanism
> To reconcile an incoming microbatch against existing history,
> {{Scd2ForeachBatchHandler.reconcileMicrobatch}} reads back the
> potentially-affected rows from both the auxiliary table and the target table
> and unions them with the microbatch into a single dataframe:
> {code:scala}
> val microbatchAndAffectedRows = preprocessedBatchDf
> .unionByName(affectedRowsFromAuxiliaryTable)
> .unionByName(affectedRowsFromTargetTable)
> {code}
> These are plain {{unionByName}} calls with no {{allowMissingColumns = true}}.
> {{unionByName}} requires all inputs to have the same set of columns, so the
> union fails as soon as the microbatch's column set diverges from what is
> stored in the target/aux tables.
> h2. When the column sets diverge
> They diverge whenever the target has been schema-evolved to be wider than the
> current microbatch:
> * *Dropped top-level column* (or a narrowed {{COLUMNS}} selection): the
> target still carries the column from an earlier wider run, but the new
> microbatch does not produce it. The microbatch has N columns and the affected
> target rows have N+1 -> {{NUM_COLUMNS_MISMATCH}}.
> * *Dropped nested struct / array-element field*: same column count, but the
> struct types differ (the target's struct has field {{b.c}}, the microbatch's
> does not), so the union fails with {{INCOMPATIBLE_COLUMN_TYPE}}.
> h2. Why SCD1 is not affected
> SCD1 never unions the microbatch with existing target rows.
> {{Scd1ForeachBatchHandler}} reconciles the microbatch on its own and then
> issues a SQL {{MERGE}} onto the target. A {{MERGE}} tolerates a source
> narrower than the target: unmentioned columns are left untouched on
> {{UPDATE}} and set to {{NULL}} on {{INSERT}}. So the identical "narrow the
> flow after the target has grown" scenario works under SCD1.
> h2. Why it matters
> This is a realistic, user-facing path -- it is exactly what cross-run schema
> evolution looks like:
> # Run a wide flow (target gains {{email}}).
> # Later run a narrower flow, or drop a source column, or tighten {{COLUMNS}}.
> # The SCD2 stream then dies on every microbatch until a full refresh.
> Schema evolution is meant to be additive-tolerant (existing rows keep their
> old columns; a flow can stop emitting one), and SCD1 honors that. SCD2 breaks
> it.
> h2. Likely fix
> Pass {{allowMissingColumns = true}} to those {{unionByName}} calls so a
> narrower microbatch is padded with {{NULL}} for the columns it no longer
> produces, matching SCD1's {{MERGE}} semantics. This needs care around the
> nested-field case (union with missing nested fields) and around ensuring
> downstream reconciliation treats the {{NULL}}-padded columns correctly.
> h2. Test coverage
> Discovered while writing the SCD2 end-to-end suites (SPARK-58409). Those
> suites currently assert the actual failure and reference this ticket, so the
> assertions can be flipped to the additive-preserving expectations once this
> is fixed:
> * {{AutoCdcScd2MultiPipelineSuite}} -- stops short of re-running the narrower
> pipeline against the widened target.
> * {{AutoCdcScd2SchemaEvolutionSuite}} -- has explicit tests asserting the
> current {{NUM_COLUMNS_MISMATCH}} / {{INCOMPATIBLE_COLUMN_TYPE}} failures.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
