[jira] [Assigned] (SPARK-58418) SCD2 AutoCDC fails with NUM_COLUMNS_MISMATCH when a flow narrower than the evolved target reconciles

["Szehon Ho (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-58418?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Szehon Ho reassigned SPARK-58418:
---------------------------------

    Assignee: Andreas Neumann

> 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
>            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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org