You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/04/20 15:08:15 UTC
(spark) branch master updated: [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 0cf7c2a68f12 [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests
0cf7c2a68f12 is described below
commit 0cf7c2a68f12bc2f74f75b02a7eb2273bceb7fa7
Author: Johan Lasperas <jo...@databricks.com>
AuthorDate: Mon Apr 20 23:07:54 2026 +0800
[SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests
### What changes were proposed in this pull request?
Follow up from https://github.com/apache/spark/pull/54891 in particular [this comment](https://github.com/apache/spark/pull/54891/changes#r3022286893).
Adds more tests covering schema evolution in MERGE INTO when the assignment contains an alias.
This change also undoes the fix from https://github.com/apache/spark/pull/54891. On closer inspection, this isn't needed in Spark as Spark already removes trivial aliases on nested field accessors in MERGE in [resolveExprInAssignment](https://github.com/apache/spark/blob/b490770d049acfcffa409e16917e92aa3db41f3d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala#L489), which is what the change aimed to allow. See [this test added here](https [...]
Delta doesn't strip aliases during resolution in that way and will require special handling, but that's up to Delta to implement in its custom resolution logic to replicate what Spark already does
### How was this patch tested?
Adds tests covering MERGE INTO schema evolution with aliases in assignments
Closes #55239 from johanl-db/SPARK-56054-merge-into-schema-evolution-alias.
Authored-by: Johan Lasperas <jo...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/catalyst/plans/logical/v2Commands.scala | 1 -
.../DeltaBasedMergeIntoSchemaEvolutionSuite.scala | 3 +-
.../GroupBasedMergeIntoSchemaEvolutionSuite.scala | 3 +-
.../MergeIntoSchemaEvolutionExtraSQLTests.scala | 54 +++++++++
.../MergeIntoSchemaEvolutionExtraScalaTests.scala | 125 +++++++++++++++++++++
...maEvolutionTypeWideningAndExtraFieldTests.scala | 62 ----------
6 files changed, 183 insertions(+), 65 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
index 500d648d23ac..3a4fcbf0d5db 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
@@ -1150,7 +1150,6 @@ object MergeIntoTable {
expr match {
case UnresolvedAttribute(nameParts) if allowUnresolved => nameParts
case a: AttributeReference => Seq(a.name)
- case Alias(child, _) => extractFieldPath(child, allowUnresolved)
case GetStructField(child, ordinal, nameOpt) =>
extractFieldPath(child, allowUnresolved) :+ nameOpt.getOrElse(s"col$ordinal")
case _ => Seq.empty
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoSchemaEvolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoSchemaEvolutionSuite.scala
index dcb102395c77..27d8eeff6942 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoSchemaEvolutionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoSchemaEvolutionSuite.scala
@@ -34,7 +34,8 @@ class DeltaBasedMergeIntoSchemaEvolutionSQLSuite
// Scala/DataFrame API-based tests for delta-based row-level operations
class DeltaBasedMergeIntoSchemaEvolutionScalaSuite
extends MergeIntoSchemaEvolutionScalaSuiteBase
- with MergeIntoSchemaEvolutionTests {
+ with MergeIntoSchemaEvolutionTests
+ with MergeIntoSchemaEvolutionExtraScalaTests {
override protected lazy val extraTableProps: java.util.Map[String, String] = {
val props = new java.util.HashMap[String, String]()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoSchemaEvolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoSchemaEvolutionSuite.scala
index badb7146b0ae..5468ecbb530c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoSchemaEvolutionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoSchemaEvolutionSuite.scala
@@ -27,5 +27,6 @@ class GroupBasedMergeIntoSchemaEvolutionSQLSuite
// Scala/DataFrame API-based tests for group-based row-level operations
class GroupBasedMergeIntoSchemaEvolutionScalaSuite
extends MergeIntoSchemaEvolutionScalaSuiteBase
- with MergeIntoSchemaEvolutionTests {
+ with MergeIntoSchemaEvolutionTests
+ with MergeIntoSchemaEvolutionExtraScalaTests {
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraSQLTests.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraSQLTests.scala
index 8565c0b31c0c..caa7f878708a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraSQLTests.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraSQLTests.scala
@@ -95,6 +95,60 @@ trait MergeIntoSchemaEvolutionExtraSQLTests extends RowLevelOperationSuiteBase {
}
}
+ Seq(true, false).foreach { withSchemaEvolution =>
+ test(s"source is subquery with projection and alias - INSERT * -" +
+ s" schema evolution=$withSchemaEvolution") {
+ withTempView("source") {
+ createAndInitTable(
+ "pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }""".stripMargin)
+
+ Seq((3, 300), (4, 400)).toDF("pk", "salary")
+ .createOrReplaceTempView("source")
+
+ val schemaEvolutionClause = if (withSchemaEvolution) "WITH SCHEMA EVOLUTION" else ""
+ val mergeStmt =
+ s"""MERGE $schemaEvolutionClause
+ |INTO $tableNameAsString t
+ |USING (SELECT pk, salary, 'new' AS dep, true AS active FROM source) AS s
+ |ON t.pk = s.pk
+ |WHEN NOT MATCHED THEN INSERT *
+ |""".stripMargin
+
+ if (withSchemaEvolution) {
+ sql(mergeStmt)
+ val result = sql(s"SELECT * FROM $tableNameAsString")
+ checkAnswer(result,
+ Seq(
+ Row(1, 100, "hr", null),
+ Row(2, 200, "software", null),
+ Row(3, 300, "new", true),
+ Row(4, 400, "new", true)))
+ assert(result.schema === StructType(Seq(
+ StructField("pk", IntegerType, nullable = false),
+ StructField("salary", IntegerType),
+ StructField("dep", StringType),
+ StructField("active", BooleanType))))
+ } else {
+ sql(mergeStmt)
+ val result = sql(s"SELECT * FROM $tableNameAsString")
+ checkAnswer(result,
+ Seq(
+ Row(1, 100, "hr"),
+ Row(2, 200, "software"),
+ Row(3, 300, "new"),
+ Row(4, 400, "new")))
+ assert(result.schema === StructType(Seq(
+ StructField("pk", IntegerType, nullable = false),
+ StructField("salary", IntegerType),
+ StructField("dep", StringType))))
+ }
+ }
+ sql(s"DROP TABLE IF EXISTS $tableNameAsString")
+ }
+ }
+
test("error when catalog ignores schema changes in MERGE WITH SCHEMA EVOLUTION") {
spark.conf.set("spark.sql.catalog.cat", classOf[PartialSchemaEvolutionCatalog].getName)
spark.sessionState.catalogManager.reset()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraScalaTests.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraScalaTests.scala
new file mode 100644
index 000000000000..e9a4742653fb
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraScalaTests.scala
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.connector.catalog.CatalogV2Util
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.types._
+
+/**
+ * DataFrame API-only tests for MERGE INTO schema evolution with aliased assignments.
+ * These tests use the `.as()` DataFrame method which has no SQL equivalent, so they only run
+ * in the Scala/DataFrame API test suites.
+ */
+trait MergeIntoSchemaEvolutionExtraScalaTests extends MergeIntoSchemaEvolutionSuiteBase {
+
+ private def withNestedTestData(body: => Unit): Unit = {
+ withTable(tableNameAsString) {
+ withTempView("source") {
+ val targetSchema = StructType(Seq(
+ StructField("pk", IntegerType, nullable = false),
+ StructField("info", StructType(Seq(
+ StructField("a", IntegerType)
+ )))
+ ))
+ createTable(CatalogV2Util.structTypeToV2Columns(targetSchema), Seq.empty)
+ val targetDf = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row(1, Row(10)),
+ Row(2, Row(20))
+ )), targetSchema)
+ targetDf.writeTo(tableNameAsString).append()
+
+ val sourceSchema = StructType(Seq(
+ StructField("pk", IntegerType, nullable = false),
+ StructField("info", StructType(Seq(
+ StructField("a", IntegerType),
+ StructField("b", IntegerType)
+ )))
+ ))
+ val sourceDf = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row(2, Row(30, 50)),
+ Row(3, Row(40, 75))
+ )), sourceSchema)
+ sourceDf.createOrReplaceTempView("source")
+
+ body
+ }
+ }
+ }
+
+ test("schema evolution - top-level aliased struct column is not evolved") {
+ withNestedTestData {
+ val ex = intercept[AnalysisException] {
+ spark.table("source")
+ .mergeInto(tableNameAsString,
+ col(s"$tableNameAsString.pk") === col("source.pk"))
+ .whenMatched().update(Map("info" -> col("source.info").as("info")))
+ .withSchemaEvolution()
+ .merge()
+ }
+ assert(ex.getCondition === "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_STRUCT_FIELDS")
+ }
+ }
+
+ // Same as above with a mismatched alias name.
+ test("schema evolution - top-level aliased struct column with mismatched name is not evolved") {
+ withNestedTestData {
+ val ex = intercept[AnalysisException] {
+ spark.table("source")
+ .mergeInto(tableNameAsString,
+ col(s"$tableNameAsString.pk") === col("source.pk"))
+ .whenMatched().update(Map(
+ "info" -> col("source.info").as("something_else")))
+ .withSchemaEvolution()
+ .merge()
+ }
+ assert(ex.getCondition === "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_STRUCT_FIELDS")
+ }
+ }
+
+ test("schema evolution - nested field through aliased struct column is not evolved") {
+ withNestedTestData {
+ val ex = intercept[AnalysisException] {
+ spark.table("source")
+ .mergeInto(tableNameAsString,
+ col(s"$tableNameAsString.pk") === col("source.pk"))
+ .whenMatched().update(Map(
+ "info.b" -> col("source.info").as("x").getField("b")))
+ .withSchemaEvolution()
+ .merge()
+ }
+ assert(ex.getCondition === "UNRESOLVED_COLUMN.WITH_SUGGESTION")
+ }
+ }
+
+ test("schema evolution - complex expression value is not considered for evolution") {
+ withNestedTestData {
+ val ex = intercept[AnalysisException] {
+ spark.table("source")
+ .mergeInto(tableNameAsString,
+ col(s"$tableNameAsString.pk") === col("source.pk"))
+ .whenMatched().update(Map(
+ "info.b" -> (col("source.info.b") + 1).as("b")))
+ .withSchemaEvolution()
+ .merge()
+ }
+ assert(ex.getCondition === "UNRESOLVED_COLUMN.WITH_SUGGESTION")
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests.scala
index ccbce30ad26e..205ffe2545d7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests.scala
@@ -18,8 +18,6 @@
package org.apache.spark.sql.connector
import org.apache.spark.sql.Row
-import org.apache.spark.sql.connector.catalog.CatalogV2Util
-import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types._
/**
@@ -1690,64 +1688,4 @@ trait MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests
StructField("dep", StringType))),
expectErrorWithoutEvolutionContains = "CAST_OVERFLOW_IN_TABLE_INSERT"
)
-
- test("schema evolution - aliased assignment value should evolve nested struct fields") {
- val targetSchema = StructType(Seq(
- StructField("pk", IntegerType, nullable = false),
- StructField("info", StructType(Seq(
- StructField("a", IntegerType)
- )))
- ))
- val sourceSchema = StructType(Seq(
- StructField("pk", IntegerType, nullable = false),
- StructField("info", StructType(Seq(
- StructField("a", IntegerType),
- StructField("b", IntegerType) // new field
- )))
- ))
-
- def readJson(json: Seq[String], schema: StructType): DataFrame =
- spark.createDataFrame(spark.read.schema(schema).json(json.toDS()).rdd, schema)
-
- withTable(tableNameAsString) {
- withTempView("source") {
- createTable(CatalogV2Util.structTypeToV2Columns(targetSchema), Seq.empty)
- readJson(Seq(
- """{ "pk": 1, "info": { "a": 10 } }""",
- """{ "pk": 2, "info": { "a": 20 } }"""
- ), targetSchema).writeTo(tableNameAsString).append()
-
- readJson(Seq(
- """{ "pk": 2, "info": { "a": 30, "b": 50 } }""",
- """{ "pk": 3, "info": { "a": 40, "b": 75 } }"""
- ), sourceSchema).createOrReplaceTempView("source")
-
- // Use DataFrame merge API and alias the source. The alias shouldn't prevent schema
- // evolution.
- spark.table("source")
- .mergeInto(tableNameAsString,
- col(s"$tableNameAsString.pk") === col("source.pk"))
- .whenMatched().update(Map("info" -> col("source.info").as("info")))
- .whenNotMatched().insert(Map(
- "pk" -> col("source.pk").as("pk"),
- "info" -> col("source.info").as("info")))
- .withSchemaEvolution()
- .merge()
-
- val result = sql(s"SELECT * FROM $tableNameAsString")
- assert(result.schema === StructType(Seq(
- StructField("pk", IntegerType, nullable = false),
- StructField("info", StructType(Seq(
- StructField("a", IntegerType),
- // Field `b` is correctly added during schema evolution.
- StructField("b", IntegerType)
- )))
- )))
- checkAnswer(result, Seq(
- Row(1, Row(10, null)),
- Row(2, Row(30, 50)),
- Row(3, Row(40, 75))))
- }
- }
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org