Skip to content

[SPARK-55690] Schema evolution in DSv2 AppendData, OverwriteByExpression, OverwritePartitionsDynamic #54488

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
johanl-db wants to merge 16 commits into from
+1,041 −322
Closed

[SPARK-55690] Schema evolution in DSv2 AppendData, OverwriteByExpression, OverwritePartitionsDynamic #54488

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
b3963d0
Move ResolveMergeIntoSchemaEvolution.scala -> ResolveMergeIntoSchemaE…
johanl-db Feb 25, 2026
05b32cb
Move schemaChanges from MergeIntoTable to ResolveSchemaEvolution
johanl-db Feb 25, 2026
7be9d2a
Schema evolution for DSv2 INSERT
johanl-db Feb 25, 2026
6bc2956
Merge branch 'master' into dsv2-schema-evolution-insert
johanl-db Feb 25, 2026
a3042e3
Add tests, address comments
johanl-db Feb 26, 2026
aeae2b4
Add tests
johanl-db Feb 27, 2026
862fdb6
Fix checking catalog name in test
johanl-db Mar 3, 2026
42ca213
Refactor to use single schema evolution rule + address comments
johanl-db Mar 10, 2026
18f10a1
Minor improvements
johanl-db Mar 10, 2026
b7e303f
Merge remote-tracking branch 'spark/master' into dsv2-schema-evolutio…
johanl-db Mar 11, 2026
0cc771f
Resolve conflicts from https://github.com/apache/spark/pull/54704
johanl-db Mar 11, 2026
65ba49a
Address comments
johanl-db Mar 13, 2026
ac8fca6
Merge remote-tracking branch 'spark/master' into dsv2-schema-evolutio…
johanl-db Mar 13, 2026
56ea329
Fix val->def writePrivileges override
johanl-db Mar 13, 2026
d777acc
Update tests, remove mergeSchema in REPLACE WHERE
johanl-db Mar 14, 2026
f8374b6
Address nits
johanl-db Mar 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions common/utils/src/main/resources/error/error-conditions.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7126,6 +7126,11 @@
"message" : [
"The catalog did not support or only partially applied the following changes: <changes>."
]
},
"TABLE_NOT_SUPPORTED" : {
"message" : [
"The write target does not support automatic schema evolution."
]
}
},
"sqlState" : "42000"
Expand Down
20 changes: 9 additions & 11 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -450,7 +450,7 @@ class Analyzer(
AddMetadataColumns ::
DeduplicateRelations ::
ResolveCollationName ::
ResolveMergeIntoSchemaEvolution ::
ResolveSchemaEvolution ::
ValidateEventTimeWatermarkColumn ::
new ResolveReferences(catalogManager) ::
// Please do not insert any other rules in between. See the TODO comments in rule
Expand Down Expand Up @@ -1167,9 +1167,6 @@ class Analyzer(
val partCols = partitionColumnNames(r.table)
validatePartitionSpec(partCols, i.partitionSpec)

val schemaEvolutionWriteOption: Map[String, String] =
if (i.withSchemaEvolution) Map("mergeSchema" -> "true") else Map.empty

val staticPartitions = i.partitionSpec.filter(_._2.isDefined).transform((_, v) => v.get)
val query = addStaticPartitionColumns(r, projectByName.getOrElse(i.query), staticPartitions,
isByName)
Expand All @@ -1179,38 +1176,38 @@ class Analyzer(
AppendData.byName(
r,
query,
writeOptions = schemaEvolutionWriteOption)
withSchemaEvolution = i.withSchemaEvolution)
} else {
AppendData.byPosition(
r,
query,
writeOptions = schemaEvolutionWriteOption)
withSchemaEvolution = i.withSchemaEvolution)
}
} else if (conf.partitionOverwriteMode == PartitionOverwriteMode.DYNAMIC) {
if (isByName) {
OverwritePartitionsDynamic.byName(
r,
query,
writeOptions = schemaEvolutionWriteOption)
withSchemaEvolution = i.withSchemaEvolution)
} else {
OverwritePartitionsDynamic.byPosition(
r,
query,
writeOptions = schemaEvolutionWriteOption)
withSchemaEvolution = i.withSchemaEvolution)
}
} else {
if (isByName) {
OverwriteByExpression.byName(
table = r,
df = query,
deleteExpr = staticDeleteExpression(r, staticPartitions),
writeOptions = schemaEvolutionWriteOption)
withSchemaEvolution = i.withSchemaEvolution)
} else {
OverwriteByExpression.byPosition(
table = r,
query = query,
deleteExpr = staticDeleteExpression(r, staticPartitions),
writeOptions = schemaEvolutionWriteOption)
withSchemaEvolution = i.withSchemaEvolution)
}
}
}
Expand Down Expand Up @@ -3637,7 +3634,8 @@ class Analyzer(
override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsWithPruning(
_.containsPattern(COMMAND), ruleId) {
case v2Write: V2WriteCommand
Copy link
Copy Markdown
Contributor

@aokolnychyi aokolnychyi Mar 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@johanl-db @szehon-ho, can you folks explain the relation between skipSchemaEvolution via ACCEPT_ANY_SCHEMA and automatic schema evolution via AUTOMATIC_SCHEMA_EVOLUTION? Are these two mutually exclusive? Or can they co-exist? MERGE vs INSERT?

Copy link
Copy Markdown
Member

@szehon-ho szehon-ho Mar 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Merge with ACCEPT_ANY_SCHEMA on normal DSV2 data source breaks today as it relies on external rule to resolve the merge.

I think insert already works with ACCEPT_ANY_SCHEMA, and this would be another mode. Probably should be mutually exclusive?

Copy link
Copy Markdown
Contributor Author

@johanl-db johanl-db Mar 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Discussed with @aokolnychyi this morning: AUTOMATIC_SCHEMA_EVOLUTION and ACCEPT_ANY_SCHEMA are not exclusive:

  • AUTOMATIC_SCHEMA_EVOLUTION allows the rule ResolveSchemaEvolution to trigger
  • ACCEPT_ANY_SCHEMA skips some resolution steps in Spark, under the assumption that the connector will handle them. In particular:
    • For INSERT, skips schema alignment in ResolveOutputRelation
    • For MERGE: skips clause resolution in Analyzer, and skips schema alignment in ResolveRowLevelCommandAssignments

At least, that's how Spark applies these capabilities today, even though the name ACCEPT_ANY_SCHEMA suggests more.

The connector can choose to set either depending on the resolution flow that suits.
For example, Delta today always handles schema evolution itself (doesn't set AUTOMATIC_SCHEMA_EVOLUTION) and does resolution / schema alignment (sets ACCEPT_ANY_SCHEMA)

As Delta moves to DSv2, my plan is to have two phases:

  1. Delta sets both AUTOMATIC_SCHEMA_EVOLUTION and ACCEPT_ANY_SCHEMA: Spark handles schema evolution, but Delta takes over to do the resolution of MERGE clauses initially, and then do schema alignment for both INSERT and MERGE
  2. Delta sets only AUTOMATIC_SCHEMA_EVOLUTION: once we've reconciled all behavior differences between how Delta and Spark do schema alignment today, we hand over schema alignment to Spark. This will require substantial efforts, and careful breaking changes (if at all possible) in Delta

if v2Write.table.resolved && v2Write.query.resolved && !v2Write.outputResolved =>
if v2Write.table.resolved && v2Write.query.resolved && !v2Write.outputResolved &&
v2Write.pendingSchemaChanges.isEmpty =>
validateStoreAssignmentPolicy()
TableOutputResolver.suitableForByNameCheck(v2Write.isByName,
expected = v2Write.table.output, queryOutput = v2Write.query.output)
Expand Down
98 changes: 0 additions & 98 deletions ...c/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveMergeIntoSchemaEvolution.scala
View file
Open in desktop

This file was deleted.

Loading
⚡