[
https://issues.apache.org/jira/browse/SPARK-58517?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andreas Neumann updated SPARK-58517:
------------------------------------
Description:
h3. Summary
SDP table schema evolution in {{DatasetManager.evolveTable}} is hard-wired to
compare schemas case-sensitively, ignoring the session's
{{spark.sql.caseSensitive}} setting. Under the default (case-insensitive)
session, a target column {{value}} and an incoming column {{Value}} are treated
as *distinct*, so the target is evolved to carry *both* columns -- a schema
that is internally inconsistent under the resolver the rest of the engine uses.
h3. Root cause
Two utilities on the evolution path take the case-sensitive default and never
forward the session conf:
* {{SchemaMergingUtils.mergeSchemas}} calls {{StructType.merge(tableSchema,
dataSchema)}} positionally. {{StructType.merge}} has a {{caseSensitive: Boolean
= true}} parameter that is left at its default.
* {{SchemaInferenceUtils.diffSchemas}} builds {{name -> field}} maps and
computes {{targetFields.keySet.diff(currentFields.keySet)}} -- an exact-string
set difference, also case-sensitive.
Both are invoked from {{DatasetManager.evolveTable}} (via
{{mergeWithExistingSchema}}), so the entire schema-evolution path ignores
{{spark.sql.caseSensitive}}.
h3. Impact
This is not AutoCDC-specific: {{SchemaMergingUtils}} / {{SchemaInferenceUtils}}
are general pipeline-table utilities, so any SDP table that evolves its schema
under a case-insensitive session and receives a column differing only in case
from an existing one is affected. The corrupt (two-column) target then fails
downstream where the case-insensitive 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 surfaces as two different error conditions depending
on SCD type; both are downstream symptoms of the corrupt evolved schema, not
the root cause.
h3. Current test coverage (characterization only)
These existing SCD2 E2E tests pin the *observed* (buggy) fallout and reference
this ticket:
* {{AutoCdcScd2SchemaEvolutionSuite}} -- "a source DF column whose name differs
from the target only by case fails with COLUMN_ALREADY_EXISTS under
case-insensitive resolution"
* {{AutoCdcScd1SchemaEvolutionSuite}} -- the analogous AMBIGUOUS_REFERENCE case
h3. Proposed fix
Thread {{caseSensitive}} (from {{conf.caseSensitiveAnalysis}}, or a
{{Resolver}}) into both {{mergeSchemas}} and {{diffSchemas}}, forwarding it to
{{StructType.merge}}. {{diffSchemas}}'s add/delete/type-change logic all keys
off exact names, so a case-insensitive variant must canonicalize consistently
to avoid spuriously deleting-then-adding a case-differing column. Add unit
coverage at the {{SchemaMergingUtils}} / {{diffSchemas}} level plus a
non-AutoCDC {{MaterializeTablesSuite}} case, since the blast radius is all
pipeline tables. Once fixed, the two characterization tests above should be
updated to assert the merge is a no-op (the incoming column maps onto the
existing one).
> SDP schema evolution ignores spark.sql.caseSensitive, corrupting the target
> schema on case-only column differences
> ------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-58517
> URL: https://issues.apache.org/jira/browse/SPARK-58517
> Project: Spark
> Issue Type: Bug
> Components: Declarative Pipelines
> Affects Versions: 5.0.0
> Reporter: Andreas Neumann
> Priority: Major
>
> h3. Summary
> SDP table schema evolution in {{DatasetManager.evolveTable}} is hard-wired to
> compare schemas case-sensitively, ignoring the session's
> {{spark.sql.caseSensitive}} setting. Under the default (case-insensitive)
> session, a target column {{value}} and an incoming column {{Value}} are
> treated as *distinct*, so the target is evolved to carry *both* columns -- a
> schema that is internally inconsistent under the resolver the rest of the
> engine uses.
> h3. Root cause
> Two utilities on the evolution path take the case-sensitive default and never
> forward the session conf:
> * {{SchemaMergingUtils.mergeSchemas}} calls {{StructType.merge(tableSchema,
> dataSchema)}} positionally. {{StructType.merge}} has a {{caseSensitive:
> Boolean = true}} parameter that is left at its default.
> * {{SchemaInferenceUtils.diffSchemas}} builds {{name -> field}} maps and
> computes {{targetFields.keySet.diff(currentFields.keySet)}} -- an
> exact-string set difference, also case-sensitive.
> Both are invoked from {{DatasetManager.evolveTable}} (via
> {{mergeWithExistingSchema}}), so the entire schema-evolution path ignores
> {{spark.sql.caseSensitive}}.
> h3. Impact
> This is not AutoCDC-specific: {{SchemaMergingUtils}} /
> {{SchemaInferenceUtils}} are general pipeline-table utilities, so any SDP
> table that evolves its schema under a case-insensitive session and receives a
> column differing only in case from an existing one is affected. The corrupt
> (two-column) target then fails downstream where the case-insensitive 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 surfaces as two different error conditions
> depending on SCD type; both are downstream symptoms of the corrupt evolved
> schema, not the root cause.
> h3. Current test coverage (characterization only)
> These existing SCD2 E2E tests pin the *observed* (buggy) fallout and
> reference this ticket:
> * {{AutoCdcScd2SchemaEvolutionSuite}} -- "a source DF column whose name
> differs from the target only by case fails with COLUMN_ALREADY_EXISTS under
> case-insensitive resolution"
> * {{AutoCdcScd1SchemaEvolutionSuite}} -- the analogous AMBIGUOUS_REFERENCE
> case
> h3. Proposed fix
> Thread {{caseSensitive}} (from {{conf.caseSensitiveAnalysis}}, or a
> {{Resolver}}) into both {{mergeSchemas}} and {{diffSchemas}}, forwarding it
> to {{StructType.merge}}. {{diffSchemas}}'s add/delete/type-change logic all
> keys off exact names, so a case-insensitive variant must canonicalize
> consistently to avoid spuriously deleting-then-adding a case-differing
> column. Add unit coverage at the {{SchemaMergingUtils}} / {{diffSchemas}}
> level plus a non-AutoCDC {{MaterializeTablesSuite}} case, since the blast
> radius is all pipeline tables. Once fixed, the two characterization tests
> above should be updated to assert the merge is a no-op (the incoming column
> maps onto the existing one).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
