[
https://issues.apache.org/jira/browse/SPARK-58418?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Szehon Ho updated SPARK-58418:
------------------------------
Component/s: Declarative Pipelines
(was: SQL)
> 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: Declarative Pipelines
> Affects Versions: 5.0.0
> Reporter: Andreas Neumann
> Assignee: Andreas Neumann
> Priority: Major
> Labels: pull-request-available
>
> 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]
