You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "johanl-db (via GitHub)" <gi...@apache.org> on 2026/04/07 16:58:53 UTC
[PR] [SPARK-56054][SQL] Add test for MERGE INTO schema evolution with aliased assignments [spark]
johanl-db opened a new pull request, #55239:
URL: https://github.com/apache/spark/pull/55239
### 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.
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
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #55239: [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests
URL: https://github.com/apache/spark/pull/55239
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Add test for MERGE INTO schema evolution with aliased assignments [spark]
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on code in PR #55239:
URL: https://github.com/apache/spark/pull/55239#discussion_r3058745285
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests.scala:
##########
@@ -1435,63 +1435,120 @@ trait MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests
requiresNestedTypeCoercion = true
)
- 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)
-
+ 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)
- readJson(Seq(
- """{ "pk": 1, "info": { "a": 10 } }""",
- """{ "pk": 2, "info": { "a": 20 } }"""
- ), targetSchema).writeTo(tableNameAsString).append()
+ 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")
- readJson(Seq(
- """{ "pk": 2, "info": { "a": 30, "b": 50 } }""",
- """{ "pk": 3, "info": { "a": 40, "b": 75 } }"""
- ), sourceSchema).createOrReplaceTempView("source")
+ body
+ }
+ }
+ }
- // Use DataFrame merge API and alias the source. The alias shouldn't prevent schema
- // evolution.
+ 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")))
- .whenNotMatched().insert(Map(
- "pk" -> col("source.pk").as("pk"),
- "info" -> col("source.info").as("info")))
.withSchemaEvolution()
.merge()
+ }
+ assert(ex.getCondition === "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_STRUCT_FIELDS")
+ }
+ }
- 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))))
+ // 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")
+ }
+ }
+
+ test("schema evolution - direct nested field assignment triggers evolution") {
Review Comment:
I removed that test
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Add test for MERGE INTO schema evolution with aliased assignments [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #55239:
URL: https://github.com/apache/spark/pull/55239#discussion_r3059503121
##########
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")))
Review Comment:
Okay, this seems like a behavior change compared to the initial PR then? The code in master would evolve the schema in this case because we had that alias support?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Add test for MERGE INTO schema evolution with aliased assignments [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #55239:
URL: https://github.com/apache/spark/pull/55239#discussion_r3047681156
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests.scala:
##########
@@ -1435,63 +1435,120 @@ trait MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests
requiresNestedTypeCoercion = true
)
- 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)
-
+ 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)
- readJson(Seq(
- """{ "pk": 1, "info": { "a": 10 } }""",
- """{ "pk": 2, "info": { "a": 20 } }"""
- ), targetSchema).writeTo(tableNameAsString).append()
+ 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")
- readJson(Seq(
- """{ "pk": 2, "info": { "a": 30, "b": 50 } }""",
- """{ "pk": 3, "info": { "a": 40, "b": 75 } }"""
- ), sourceSchema).createOrReplaceTempView("source")
+ body
+ }
+ }
+ }
- // Use DataFrame merge API and alias the source. The alias shouldn't prevent schema
- // evolution.
+ 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")))
- .whenNotMatched().insert(Map(
- "pk" -> col("source.pk").as("pk"),
- "info" -> col("source.info").as("info")))
.withSchemaEvolution()
.merge()
+ }
+ assert(ex.getCondition === "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_STRUCT_FIELDS")
+ }
+ }
- 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))))
+ // 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")
+ }
+ }
+
+ test("schema evolution - direct nested field assignment triggers evolution") {
Review Comment:
i think this test is already covered
MergeIntoSchemaEvolutionTypeWideningAndExtraFieldTests.scala
Lines 72-119
testNestedStructsEvolution("source has extra struct fields -" +
"only evolve referenced struct field
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Add test for MERGE INTO schema evolution with aliased assignments [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #55239:
URL: https://github.com/apache/spark/pull/55239#discussion_r3059503121
##########
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")))
Review Comment:
Okay, this seems like a behavior change compare to the initial PR then? The code in master would evolve the schema in this case because we had that alias support?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Add test for MERGE INTO schema evolution with aliased assignments [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #55239:
URL: https://github.com/apache/spark/pull/55239#discussion_r3059507659
##########
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")))
Review Comment:
How does Delta behave in this case?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests [spark]
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on PR #55239:
URL: https://github.com/apache/spark/pull/55239#issuecomment-4243181987
CI is failing due to formatting on files that are not touched in this PR. Likely from https://github.com/apache/spark/pull/55281
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Add test for MERGE INTO schema evolution with aliased assignments [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on PR #55239:
URL: https://github.com/apache/spark/pull/55239#issuecomment-4216121932
@johanl-db, what about cases when the source query has an alias?
```
MERGE INTO t
USING (SELECT 'blah' AS col1, 1 AS col2 FROM source) AS s
ON false
WHEN NOT MATCHED THEN INSERT *
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #55239:
URL: https://github.com/apache/spark/pull/55239#issuecomment-4255514439
also , apparently this ci error is fixed by [#55334](https://github.com/apache/spark/pull/55334)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests [spark]
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on PR #55239:
URL: https://github.com/apache/spark/pull/55239#issuecomment-4258344401
> I think this one is ready to go @cloud-fan
>
> i also added the test requested by @aokolnychyi in #55304 (i didnt see the comment), we can add it separately , or optionally as part this pr, i guess it doesnt matter. As we looked into it together, its a bit orthogonal to this pr (the aliases within source doens't matter, to MERGE it just comes as a source table)
@szehon-ho I picked up your test from https://github.com/apache/spark/pull/55304 in this PR
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests [spark]
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on code in PR #55239:
URL: https://github.com/apache/spark/pull/55239#discussion_r3077498499
##########
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")))
Review Comment:
Delta allows schema evolution here, but only because it allows arbitrary expressions on the right hand side of the assignment. `col("source.info").as("something_else")` can qualifies for schema evolution the same way `col("source.other_column").plus(lit(1))` does.
Spark rejects both (with this change)
##########
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")))
Review Comment:
Correct. The main motivation for the initial PR adding support for alias was to support nested assignments `SET col.x = s.col.x`, which translates to `Assignment("col.x", Alias(GetStructField("col", "x"), "x")`
I've noticed in the meantime that Spark actually strips the alias on struct field access after resolution (Delta doesn't).
We don't have a strong argument to support the remaining cases where the user provides an explicit alias using the dataframe API, so I'd rather not allow these
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #55239:
URL: https://github.com/apache/spark/pull/55239#issuecomment-4281967533
thanks, merging to master!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Undo handling aliased assignments in MERGE INTO schema evolution and add tests [spark]
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on PR #55239:
URL: https://github.com/apache/spark/pull/55239#issuecomment-4241723533
> @johanl-db, what about cases when the source query has an alias?
>
> ```
> MERGE INTO t
> USING (SELECT 'blah' AS col1, 1 AS col2 FROM source) AS s
> ON false
> WHEN NOT MATCHED THEN INSERT *
> ```
Discussed directly with @aokolnychyi and @szehon-ho :
The shape of source doesn't matter at all, the only parts that matter when performing schema evolution are:
- The schema of the source
- The assignment expressions in the match clauses
MERGE doesn't look inside the source query to see what it contains, so the aliases there are irrelevant.
This case will trigger schema evolution for all new columns in the source schema, since there's an INSERT * assignment.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56054][SQL] Add test for MERGE INTO schema evolution with aliased assignments [spark]
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on PR #55239:
URL: https://github.com/apache/spark/pull/55239#issuecomment-4215246141
I moved the tests to a dedicated trait for dataframe tests.
There's no SQL equivalent, it's not possible to specify an alias in an assignment expression using SQL afaict: `SET col = source.col AS other_col`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org