Skip to content

[SPARK-58517][SDP] Honor spark.sql.caseSensitive in table schema evolution - #57722

Open
anew wants to merge 1 commit into
apache:masterfrom
anew:spark-58517-schema-evolution-case-sensitivity
Open

[SPARK-58517][SDP] Honor spark.sql.caseSensitive in table schema evolution#57722
anew wants to merge 1 commit into
apache:masterfrom
anew:spark-58517-schema-evolution-case-sensitivity

Conversation

@anew

@anew anew commented Aug 3, 2026

Copy link
Copy Markdown
Contributor

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

…ution

DatasetManager's schema evolution merged and diffed schemas case-sensitively
regardless of the session's spark.sql.caseSensitive. Under the default
(case-insensitive) session, a target column `value` and an incoming `Value`
were treated as distinct, so the target was evolved to carry both -- a schema
self-inconsistent under the resolver the rest of the engine uses. The corrupt
two-column target then failed downstream (AutoCDC SCD2: COLUMN_ALREADY_EXISTS
from ResolveUnion over the reconcile unionByName; SCD1: AMBIGUOUS_REFERENCE in
the MERGE plan). This is not AutoCDC-specific -- it affects any SDP table that
evolves its schema under a case-insensitive session.

Thread the session's case-sensitivity into SchemaMergingUtils.mergeSchemas and
SchemaInferenceUtils.diffSchemas (both default to case-sensitive to preserve the
schema-inference callers) and forward it from DatasetManager.evolveTable. With
case-insensitivity, StructType.merge folds `Value` onto the existing `value`
(the left field's name wins) so no duplicate column is added, and diffSchemas
matches on the normalized name while addressing the column by its persisted
name (no spurious drop-then-add, no rename).

Tests: unit coverage in SchemaInferenceUtilsSuite (case-sensitive vs
-insensitive diff, and that a case-insensitive type change addresses the current
column name); an end-to-end MaterializeTablesSuite pair (no-op under
case-insensitive, add under case-sensitive). Updated
AutoCdcScd1SchemaEvolutionSuite's case-only test to assert the fixed
no-op-merge behavior, and refreshed two now-stale comments.

Co-authored-by: Opus 4.8
@anew
anew force-pushed the spark-58517-schema-evolution-case-sensitivity branch from 67eb4ec to 76144f5 Compare August 3, 2026 22:34
@uros-b

uros-b commented Aug 4, 2026

Copy link
Copy Markdown
Member

Thank you @anew! Adding @szehon-ho

@uros-b
uros-b requested a review from szehon-ho August 4, 2026 07:32
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants