anew opened a new pull request, #57722:
URL: https://github.com/apache/spark/pull/57722
### What changes were proposed in this pull request?
Declarative Pipelines' table schema evolution in
`DatasetManager.evolveTable` compared schemas **case-sensitively regardless of
the session's `spark.sql.caseSensitive`**. Under the default (case-insensitive)
session, a target column `value` and an incoming column `Value` were treated as
distinct, so the target was evolved to carry *both* columns -- a schema that is
internally inconsistent under the resolver the rest of the engine uses.
The root cause is on two utilities on the evolution path, each taking the
case-sensitive default and never forwarding the session conf:
- `SchemaMergingUtils.mergeSchemas` called `StructType.merge(tableSchema,
dataSchema)` positionally, leaving `StructType.merge`'s `caseSensitive: Boolean
= true` parameter at its default.
- `SchemaInferenceUtils.diffSchemas` built `name -> field` maps and
computed `targetFields.keySet.diff(currentFields.keySet)` -- an exact-string
set difference, also case-sensitive.
This PR threads case-sensitivity through both:
- `mergeSchemas` and `diffSchemas` gain a `caseSensitive: Boolean = true`
parameter. The default preserves existing behavior for the schema-*inference*
callers (`inferSchemaFromFlows`, `DataflowGraph.inferredSchema`), which
intentionally merge case-sensitively.
- `DatasetManager.evolveTable` forwards
`context.spark.sessionState.conf.caseSensitiveAnalysis` at both call sites (the
target table and the AutoCDC auxiliary table).
- With case-insensitivity, `StructType.merge` folds the incoming `Value`
onto the existing `value` (the left field's name/position wins), so no
duplicate column is added; `diffSchemas` matches on the normalized name while
emitting the column's already-persisted name, so a case-only difference is an
in-place no-op rather than a spurious drop-then-add, and no rename is produced.
This is not AutoCDC-specific -- `SchemaMergingUtils` /
`SchemaInferenceUtils` are general pipeline-table utilities, so the fix covers
any SDP table that evolves its schema under a case-insensitive session.
### Why are the changes needed?
It is a correctness bug. Evolving a table under the default
case-insensitive session with a column that differs only in case from an
existing one produced a target schema with two columns (`value` and `Value`)
that the engine's own case-insensitive resolver cannot tell apart. The corrupt
schema then failed downstream where that resolver runs:
- AutoCDC SCD2: `COLUMN_ALREADY_EXISTS`, raised by
`ResolveUnion.checkColumnNames` over the `unionByName` in
`Scd2ForeachBatchHandler.reconcileMicrobatch` (the affected-rows union reads
the now-two-column target back).
- AutoCDC SCD1: `AMBIGUOUS_REFERENCE`, deeper in the MERGE plan (no
reconcile union).
The same user mistake thus surfaced as two different, confusing error
conditions -- both downstream symptoms of the corrupt evolved schema rather
than the root cause.
### Does this PR introduce _any_ user-facing change?
Yes, a bug fix (only within the unreleased `master` / `branch-4.x`; the
buggy behavior was never in a release). Under case-insensitive resolution
(Spark's default), a pipeline that emits a column differing only in case from
an existing target column now evolves the table as a no-op (the incoming value
is written to the existing column) instead of corrupting the schema and failing
with `COLUMN_ALREADY_EXISTS` (SCD2) or `AMBIGUOUS_REFERENCE` (SCD1). Under
`spark.sql.caseSensitive=true`, the two names remain distinct and the new
column is added, as before.
### How was this patch tested?
Added and updated tests:
- `SchemaInferenceUtilsSuite` -- new unit tests for `diffSchemas`:
case-only difference is distinct under case-sensitive, a no-op under
case-insensitive, and a case-insensitive type change addresses the current
(persisted) column name.
- `MaterializeTablesSuite` -- an end-to-end pair (non-AutoCDC path)
asserting no `alterTable` is issued and the persisted column keeps its
original case under case-insensitive resolution, versus a column being added
under case-sensitive.
- `AutoCdcScd1SchemaEvolutionSuite` -- rewrote the case-only-difference
test from asserting the old `AMBIGUOUS_REFERENCE` failure to asserting the
fixed no-op-merge success; refreshed the suite header and a now-stale comment
in `AutoCdcScd1KeyDriftSuite`.
Ran the full `pipelines` module test suite. All pass except two failures
(`TriggeredGraphExecutionSuite`: "stream failure on deletes and updates",
"stopping a pipeline mid-execution") that I confirmed are pre-existing on clean
`master` (they fail identically with this change stashed), so this PR
introduces no regressions.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude 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]
