anew opened a new pull request, #57669:
URL: https://github.com/apache/spark/pull/57669
### What changes were proposed in this pull request?
`Scd2ForeachBatchHandler.reconcileMicrobatch` unions the incoming microbatch
with the affected auxiliary- and target-table rows in order to reconcile SCD
Type 2 history. Those two unions used plain `unionByName`, which requires all
inputs to have an identical column set:
```scala
val microbatchAndAffectedRows = preprocessedBatchDf
.unionByName(affectedRowsFromAuxiliaryTable)
.unionByName(affectedRowsFromTargetTable)
```
After cross-run schema evolution, the target table (and the auxiliary table,
whose schema mirrors it) can carry user columns that the current microbatch no
longer emits — a dropped source column, a narrowed `COLUMNS` selection, or a
dropped nested struct/array field. The microbatch is then *narrower* than the
affected rows, and the union fails.
This passes `allowMissingColumns = true` to both unions, so a narrower
microbatch is padded with `null` for the columns it no longer emits (recursing
into structs and arrays; map types are not supported by `unionByName`, a
documented limitation).
### Why are the changes needed?
Before this change, an SCD2 AutoCDC stream fails on **every** microbatch
once a flow becomes narrower than its evolved target:
- dropped top-level column / narrowed `COLUMNS` selection →
`NUM_COLUMNS_MISMATCH`
- dropped nested struct or array-element field → `INCOMPATIBLE_COLUMN_TYPE`
SCD Type 1 does not have this problem because it `MERGE`s the microbatch
onto the target rather than unioning it with existing rows, and a `MERGE`
tolerates a source narrower than the target. SCD2 should match that
additive-tolerant behavior so column-schema evolution works without a full
refresh.
Records already written to the target keep their values: the target-table
`MERGE`'s `UPDATE` assignments draw from the affected target rows, not from the
null-padded microbatch rows, so only records *opened* by the narrower
microbatch carry `null` for the no-longer-emitted column — the same semantics
SCD1 gives.
Note: this covers column-schema evolution that leaves the effective
tracked-history column set unchanged. Changing the tracked-history set is a
distinct concern with different correctness implications and is scoped
separately under SPARK-58452 - and initially will require a full-refresh under
SPARK-58391.
### Does this PR introduce _any_ user-facing change?
Yes. An SCD2 AutoCDC flow whose microbatch is narrower than its
already-evolved target now reconciles correctly instead of failing the stream
with `NUM_COLUMNS_MISMATCH` / `INCOMPATIBLE_COLUMN_TYPE`. Columns dropped from
the flow are preserved on records already written and are `null` on records
opened after the drop. This is within the unreleased AutoCDC feature on master.
### How was this patch tested?
New `AutoCdcScd2ColumnEvolutionSuite` (4 end-to-end tests):
- a source column dropped between runs — preserved on existing records,
`null` on new ones;
- the `COLUMNS` selection narrowed between runs — same contract;
- a late narrower event that bisects existing history — pre-existing
records keep their values, only the new record carries `null`;
- a nested struct field dropped between runs — preserved on existing
records, `null` on new ones.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
