You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "szehon-ho (via GitHub)" <gi...@apache.org> on 2025/09/18 01:01:39 UTC
[PR] [SPARK-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
szehon-ho opened a new pull request, #52377:
URL: https://github.com/apache/spark/pull/52377
### What changes were proposed in this pull request?
MERGE INTO WITH SCHEMA EVOLUTION already support adding new column, and also some type widening (if structs are missing some fields)
It should support type widening for primitive data types. Spark will call the V2DataSource TableCatalog to alter the schema, so the V2DataSource can decide whether it is acceptable or not.
This change also fixes InMemoryDataSource to support this case
### Why are the changes needed?
Support more use case for MERGE INTO WITH SCHEMA EVOLUTION.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Add unit test.
### Was this patch authored or co-authored using generative AI tooling?
No
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52377:
URL: https://github.com/apache/spark/pull/52377#discussion_r2361553293
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -962,6 +972,14 @@ private class BufferedRowsReader(
wKeyType, wValueType)
}
}
+ case (readType: AtomicType, writeType: AtomicType) if readType != writeType =>
Review Comment:
yes
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52377:
URL: https://github.com/apache/spark/pull/52377#discussion_r2367867338
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala:
##########
@@ -2450,6 +2450,409 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
}
}
+ test("merge into schema evolution type widening from short to int") {
+ Seq((true, true), (false, true), (true, false)).foreach {
+ case (withSchemaEvolution, schemaEvolutionEnabled) =>
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary SMALLINT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "finance" }
+ |{ "pk": 3, "salary": 300, "dep": "engineering" }
+ |""".stripMargin)
+
+ if (!schemaEvolutionEnabled) {
+ sql(s"""ALTER TABLE $tableNameAsString SET TBLPROPERTIES
+ | ('auto-schema-evolution' = 'false')""".stripMargin)
+ }
+
+ // Source data with int salary values that would exceed short range
+ val sourceRows = Seq(
+ (1, 50000, "hr"),
+ (4, 40000, "sales"),
+ (5, 500, "marketing"))
+ sourceRows.toDF("pk", "salary", "dep").createOrReplaceTempView("source")
+
+ val schemaEvolutionClause = if (withSchemaEvolution) "WITH SCHEMA EVOLUTION" else ""
+ val mergeStmt = s"""MERGE $schemaEvolutionClause
+ |INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN
+ | UPDATE SET salary = s.salary
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, salary, dep) VALUES (s.pk, s.salary, s.dep)
+ |""".stripMargin
+
+ if (withSchemaEvolution && schemaEvolutionEnabled) {
+ // Schema evolution should allow type widening from SMALLINT to INT for salary column
+ sql(mergeStmt)
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Seq(
+ Row(1, 50000, "hr"),
+ Row(2, 200, "finance"),
+ Row(3, 300, "engineering"),
+ Row(4, 40000, "sales"),
+ Row(5, 500, "marketing")))
+ val tableSchema = sql(s"SELECT * FROM $tableNameAsString").schema
+ val salaryField = tableSchema.find(_.name == "salary").get
+ assert(salaryField.dataType == IntegerType)
+ } else {
+ val exception = intercept[Exception] {
+ sql(mergeStmt)
+ }
+ assert(exception.getMessage.contains(
+ "Fail to assign a value of \"INT\" type to the \"SMALLINT\" " +
+ "type column or variable `salary` due to an overflow"))
+ }
+ sql(s"DROP TABLE IF EXISTS $tableNameAsString")
+ }
+ }
+ }
+
+ test("merge into schema evolution type widening nested struct from int to long") {
+ Seq(true, false).foreach { withSchemaEvolution =>
+ withTempView("source") {
+ // Create table with nested struct containing int field
+ createAndInitTable(
+ s"""pk INT NOT NULL,
+ |employee STRUCT<salary: INT, details: STRUCT<bonus: INT, years: INT>>,
+ |dep STRING""".stripMargin,
+ """{ "pk": 1, "employee": { "salary": 50000, "details":
+ |{ "bonus": 5000, "years": 2 } }, "dep": "hr" }""".stripMargin.replace("\n", "")
+ + "\n" +
+ """{ "pk": 2, "employee": { "salary": 60000, "details":
+ |{ "bonus": 6000, "years": 3 } }, "dep": "finance" }""".stripMargin.replace("\n", "")
+ )
+
+ // Source data with long values that exceed int range for nested fields
+ val sourceTableSchema = StructType(Seq(
+ StructField("pk", IntegerType),
+ StructField("employee", StructType(Seq(
+ StructField("salary", IntegerType),
+ StructField("details", StructType(Seq(
+ StructField("bonus", LongType), // Changed from INT to LONG
+ StructField("years", IntegerType)
+ )))
+ ))),
+ StructField("dep", StringType)
+ ))
+
+ val data = Seq(
+ Row(1, Row(75000, Row(3000000000L, 5)), "hr"),
+ Row(3, Row(80000, Row(4000000000L, 1)), "engineering")
+ )
+
+ spark.createDataFrame(spark.sparkContext.parallelize(data), sourceTableSchema)
+ .createOrReplaceTempView("source")
+
+ val schemaEvolutionClause = if (withSchemaEvolution) "WITH SCHEMA EVOLUTION" else ""
+ val mergeStmt = s"""MERGE $schemaEvolutionClause
+ |INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN
+ | UPDATE SET *
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, employee, dep) VALUES (s.pk, s.employee, s.dep)
+ |""".stripMargin
+
+ if (withSchemaEvolution) {
+ // Schema evolution should allow type widening from INT to LONG for nested bonus field
+ sql(mergeStmt)
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Seq(
+ Row(1, Row(75000, Row(3000000000L, 5)), "hr"),
+ Row(2, Row(60000, Row(6000, 3)), "finance"),
+ Row(3, Row(80000, Row(4000000000L, 1)), "engineering")
+ ))
+
+ val tableSchema = sql(s"SELECT * FROM $tableNameAsString").schema
+ val employeeField = tableSchema.find(_.name == "employee").get.dataType
+ .asInstanceOf[StructType]
+ val detailsField = employeeField.find(_.name == "details").get.dataType
+ .asInstanceOf[StructType]
+ val bonusField = detailsField.find(_.name == "bonus").get
+ assert(bonusField.dataType == LongType)
+ } else {
+ val exception = intercept[Exception] {
+ sql(mergeStmt)
+ }
+ assert(exception.getMessage.contains("Fail to assign a value of \"BIGINT\" type " +
+ "to the \"INT\" type column or variable `employee`.`details`.`bonus`" +
+ " due to an overflow"))
+ }
+ sql(s"DROP TABLE IF EXISTS $tableNameAsString")
Review Comment:
let's use `withTable` for cleanup.
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52377:
URL: https://github.com/apache/spark/pull/52377#discussion_r2369714092
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala:
##########
@@ -2450,6 +2450,409 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
}
}
+ test("merge into schema evolution type widening from short to int") {
+ Seq((true, true), (false, true), (true, false)).foreach {
+ case (withSchemaEvolution, schemaEvolutionEnabled) =>
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary SMALLINT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "finance" }
+ |{ "pk": 3, "salary": 300, "dep": "engineering" }
+ |""".stripMargin)
+
+ if (!schemaEvolutionEnabled) {
+ sql(s"""ALTER TABLE $tableNameAsString SET TBLPROPERTIES
+ | ('auto-schema-evolution' = 'false')""".stripMargin)
+ }
+
+ // Source data with int salary values that would exceed short range
+ val sourceRows = Seq(
+ (1, 50000, "hr"),
+ (4, 40000, "sales"),
+ (5, 500, "marketing"))
+ sourceRows.toDF("pk", "salary", "dep").createOrReplaceTempView("source")
+
+ val schemaEvolutionClause = if (withSchemaEvolution) "WITH SCHEMA EVOLUTION" else ""
+ val mergeStmt = s"""MERGE $schemaEvolutionClause
+ |INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN
+ | UPDATE SET salary = s.salary
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, salary, dep) VALUES (s.pk, s.salary, s.dep)
+ |""".stripMargin
+
+ if (withSchemaEvolution && schemaEvolutionEnabled) {
+ // Schema evolution should allow type widening from SMALLINT to INT for salary column
+ sql(mergeStmt)
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Seq(
+ Row(1, 50000, "hr"),
+ Row(2, 200, "finance"),
+ Row(3, 300, "engineering"),
+ Row(4, 40000, "sales"),
+ Row(5, 500, "marketing")))
+ val tableSchema = sql(s"SELECT * FROM $tableNameAsString").schema
+ val salaryField = tableSchema.find(_.name == "salary").get
+ assert(salaryField.dataType == IntegerType)
+ } else {
+ val exception = intercept[Exception] {
+ sql(mergeStmt)
+ }
+ assert(exception.getMessage.contains(
+ "Fail to assign a value of \"INT\" type to the \"SMALLINT\" " +
+ "type column or variable `salary` due to an overflow"))
+ }
+ sql(s"DROP TABLE IF EXISTS $tableNameAsString")
+ }
+ }
+ }
+
+ test("merge into schema evolution type widening nested struct from int to long") {
+ Seq(true, false).foreach { withSchemaEvolution =>
+ withTempView("source") {
+ // Create table with nested struct containing int field
+ createAndInitTable(
+ s"""pk INT NOT NULL,
+ |employee STRUCT<salary: INT, details: STRUCT<bonus: INT, years: INT>>,
+ |dep STRING""".stripMargin,
+ """{ "pk": 1, "employee": { "salary": 50000, "details":
+ |{ "bonus": 5000, "years": 2 } }, "dep": "hr" }""".stripMargin.replace("\n", "")
+ + "\n" +
+ """{ "pk": 2, "employee": { "salary": 60000, "details":
+ |{ "bonus": 6000, "years": 3 } }, "dep": "finance" }""".stripMargin.replace("\n", "")
+ )
+
+ // Source data with long values that exceed int range for nested fields
+ val sourceTableSchema = StructType(Seq(
+ StructField("pk", IntegerType),
+ StructField("employee", StructType(Seq(
+ StructField("salary", IntegerType),
+ StructField("details", StructType(Seq(
+ StructField("bonus", LongType), // Changed from INT to LONG
+ StructField("years", IntegerType)
+ )))
+ ))),
+ StructField("dep", StringType)
+ ))
+
+ val data = Seq(
+ Row(1, Row(75000, Row(3000000000L, 5)), "hr"),
+ Row(3, Row(80000, Row(4000000000L, 1)), "engineering")
+ )
+
+ spark.createDataFrame(spark.sparkContext.parallelize(data), sourceTableSchema)
+ .createOrReplaceTempView("source")
+
+ val schemaEvolutionClause = if (withSchemaEvolution) "WITH SCHEMA EVOLUTION" else ""
+ val mergeStmt = s"""MERGE $schemaEvolutionClause
+ |INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN
+ | UPDATE SET *
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, employee, dep) VALUES (s.pk, s.employee, s.dep)
+ |""".stripMargin
+
+ if (withSchemaEvolution) {
+ // Schema evolution should allow type widening from INT to LONG for nested bonus field
+ sql(mergeStmt)
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Seq(
+ Row(1, Row(75000, Row(3000000000L, 5)), "hr"),
+ Row(2, Row(60000, Row(6000, 3)), "finance"),
+ Row(3, Row(80000, Row(4000000000L, 1)), "engineering")
+ ))
+
+ val tableSchema = sql(s"SELECT * FROM $tableNameAsString").schema
+ val employeeField = tableSchema.find(_.name == "employee").get.dataType
+ .asInstanceOf[StructType]
+ val detailsField = employeeField.find(_.name == "details").get.dataType
+ .asInstanceOf[StructType]
+ val bonusField = detailsField.find(_.name == "bonus").get
+ assert(bonusField.dataType == LongType)
+ } else {
+ val exception = intercept[Exception] {
+ sql(mergeStmt)
+ }
+ assert(exception.getMessage.contains("Fail to assign a value of \"BIGINT\" type " +
+ "to the \"INT\" type column or variable `employee`.`details`.`bonus`" +
+ " due to an overflow"))
+ }
+ sql(s"DROP TABLE IF EXISTS $tableNameAsString")
Review Comment:
done, there's other tests in the file (not added by this pr) that also don't use withTable, let me fix those in another 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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52377:
URL: https://github.com/apache/spark/pull/52377#discussion_r2361551862
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -921,14 +922,23 @@ private class BufferedRowsReader(
readValueType: DataType,
writeKeyType: DataType,
writeValueType: DataType): MapData = {
- val keys = mapData.keyArray().toArray[Any](readKeyType)
- val values = mapData.valueArray().toArray[Any](readValueType)
+ val keys = mapData.keyArray().toArray[Any](writeKeyType)
+ val values = mapData.valueArray().toArray[Any](writeValueType)
val convertedKeys = extractCollection(keys, readKeyType, writeKeyType)
val convertedValues = extractCollection(values, readValueType, writeValueType)
ArrayBasedMapData(convertedKeys, convertedValues)
}
+ private def castElement(elem: Any, toType: DataType, fromType: DataType): Any = {
+ try {
+ val castExpr = Cast(Literal(elem, fromType), toType)
+ castExpr.eval(null)
+ } catch {
+ case _: Exception => elem
Review Comment:
shall we return null for invalid cast?
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #52377:
URL: https://github.com/apache/spark/pull/52377#issuecomment-3310050148
i believe the test error is not related (AdaptiveQueryExecSuite), I couldnt reproduct it, re-running
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52377:
URL: https://github.com/apache/spark/pull/52377#discussion_r2367868161
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala:
##########
@@ -2450,6 +2450,409 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
}
}
+ test("merge into schema evolution type widening from short to int") {
+ Seq((true, true), (false, true), (true, false)).foreach {
+ case (withSchemaEvolution, schemaEvolutionEnabled) =>
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary SMALLINT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "finance" }
+ |{ "pk": 3, "salary": 300, "dep": "engineering" }
+ |""".stripMargin)
+
+ if (!schemaEvolutionEnabled) {
+ sql(s"""ALTER TABLE $tableNameAsString SET TBLPROPERTIES
+ | ('auto-schema-evolution' = 'false')""".stripMargin)
+ }
+
+ // Source data with int salary values that would exceed short range
+ val sourceRows = Seq(
+ (1, 50000, "hr"),
+ (4, 40000, "sales"),
+ (5, 500, "marketing"))
+ sourceRows.toDF("pk", "salary", "dep").createOrReplaceTempView("source")
+
+ val schemaEvolutionClause = if (withSchemaEvolution) "WITH SCHEMA EVOLUTION" else ""
+ val mergeStmt = s"""MERGE $schemaEvolutionClause
+ |INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN
+ | UPDATE SET salary = s.salary
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, salary, dep) VALUES (s.pk, s.salary, s.dep)
+ |""".stripMargin
+
+ if (withSchemaEvolution && schemaEvolutionEnabled) {
+ // Schema evolution should allow type widening from SMALLINT to INT for salary column
+ sql(mergeStmt)
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Seq(
+ Row(1, 50000, "hr"),
+ Row(2, 200, "finance"),
+ Row(3, 300, "engineering"),
+ Row(4, 40000, "sales"),
+ Row(5, 500, "marketing")))
+ val tableSchema = sql(s"SELECT * FROM $tableNameAsString").schema
+ val salaryField = tableSchema.find(_.name == "salary").get
+ assert(salaryField.dataType == IntegerType)
+ } else {
+ val exception = intercept[Exception] {
+ sql(mergeStmt)
+ }
+ assert(exception.getMessage.contains(
+ "Fail to assign a value of \"INT\" type to the \"SMALLINT\" " +
+ "type column or variable `salary` due to an overflow"))
+ }
+ sql(s"DROP TABLE IF EXISTS $tableNameAsString")
+ }
+ }
+ }
+
+ test("merge into schema evolution type widening nested struct from int to long") {
+ Seq(true, false).foreach { withSchemaEvolution =>
+ withTempView("source") {
+ // Create table with nested struct containing int field
+ createAndInitTable(
+ s"""pk INT NOT NULL,
+ |employee STRUCT<salary: INT, details: STRUCT<bonus: INT, years: INT>>,
+ |dep STRING""".stripMargin,
+ """{ "pk": 1, "employee": { "salary": 50000, "details":
+ |{ "bonus": 5000, "years": 2 } }, "dep": "hr" }""".stripMargin.replace("\n", "")
+ + "\n" +
+ """{ "pk": 2, "employee": { "salary": 60000, "details":
+ |{ "bonus": 6000, "years": 3 } }, "dep": "finance" }""".stripMargin.replace("\n", "")
+ )
+
+ // Source data with long values that exceed int range for nested fields
+ val sourceTableSchema = StructType(Seq(
+ StructField("pk", IntegerType),
+ StructField("employee", StructType(Seq(
+ StructField("salary", IntegerType),
+ StructField("details", StructType(Seq(
+ StructField("bonus", LongType), // Changed from INT to LONG
+ StructField("years", IntegerType)
+ )))
+ ))),
+ StructField("dep", StringType)
+ ))
+
+ val data = Seq(
+ Row(1, Row(75000, Row(3000000000L, 5)), "hr"),
+ Row(3, Row(80000, Row(4000000000L, 1)), "engineering")
+ )
+
+ spark.createDataFrame(spark.sparkContext.parallelize(data), sourceTableSchema)
+ .createOrReplaceTempView("source")
+
+ val schemaEvolutionClause = if (withSchemaEvolution) "WITH SCHEMA EVOLUTION" else ""
+ val mergeStmt = s"""MERGE $schemaEvolutionClause
+ |INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN
+ | UPDATE SET *
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, employee, dep) VALUES (s.pk, s.employee, s.dep)
+ |""".stripMargin
+
+ if (withSchemaEvolution) {
+ // Schema evolution should allow type widening from INT to LONG for nested bonus field
+ sql(mergeStmt)
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Seq(
+ Row(1, Row(75000, Row(3000000000L, 5)), "hr"),
+ Row(2, Row(60000, Row(6000, 3)), "finance"),
+ Row(3, Row(80000, Row(4000000000L, 1)), "engineering")
+ ))
+
+ val tableSchema = sql(s"SELECT * FROM $tableNameAsString").schema
+ val employeeField = tableSchema.find(_.name == "employee").get.dataType
+ .asInstanceOf[StructType]
+ val detailsField = employeeField.find(_.name == "details").get.dataType
+ .asInstanceOf[StructType]
+ val bonusField = detailsField.find(_.name == "bonus").get
+ assert(bonusField.dataType == LongType)
+ } else {
+ val exception = intercept[Exception] {
+ sql(mergeStmt)
+ }
+ assert(exception.getMessage.contains("Fail to assign a value of \"BIGINT\" type " +
+ "to the \"INT\" type column or variable `employee`.`details`.`bonus`" +
+ " due to an overflow"))
+ }
+ sql(s"DROP TABLE IF EXISTS $tableNameAsString")
Review Comment:
ditto for other new test cases in ths file
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #52377: [SPARK-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION
URL: https://github.com/apache/spark/pull/52377
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #52377:
URL: https://github.com/apache/spark/pull/52377#issuecomment-3322780427
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52377:
URL: https://github.com/apache/spark/pull/52377#discussion_r2361552781
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -921,14 +922,23 @@ private class BufferedRowsReader(
readValueType: DataType,
writeKeyType: DataType,
writeValueType: DataType): MapData = {
- val keys = mapData.keyArray().toArray[Any](readKeyType)
- val values = mapData.valueArray().toArray[Any](readValueType)
+ val keys = mapData.keyArray().toArray[Any](writeKeyType)
+ val values = mapData.valueArray().toArray[Any](writeValueType)
val convertedKeys = extractCollection(keys, readKeyType, writeKeyType)
val convertedValues = extractCollection(values, readValueType, writeValueType)
ArrayBasedMapData(convertedKeys, convertedValues)
}
+ private def castElement(elem: Any, toType: DataType, fromType: DataType): Any = {
+ try {
+ val castExpr = Cast(Literal(elem, fromType), toType)
+ castExpr.eval(null)
+ } catch {
+ case _: Exception => elem
Review Comment:
or we set the `evalMode` to `TRY` when constructing `Cast`
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #52377:
URL: https://github.com/apache/spark/pull/52377#issuecomment-3322018729
rebase and re-running the 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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52377:
URL: https://github.com/apache/spark/pull/52377#discussion_r2357213064
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -962,6 +972,14 @@ private class BufferedRowsReader(
wKeyType, wValueType)
}
}
+ case (readType: AtomicType, writeType: AtomicType) if readType != writeType =>
Review Comment:
it seems the InternalRow format requires cast of each element , if the nested type changes for collections like Map/Array.
--
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-53629][SQL] Implement type widening for MERGE INTO WITH SCHEMA EVOLUTION [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #52377:
URL: https://github.com/apache/spark/pull/52377#discussion_r2364758599
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -921,14 +922,23 @@ private class BufferedRowsReader(
readValueType: DataType,
writeKeyType: DataType,
writeValueType: DataType): MapData = {
- val keys = mapData.keyArray().toArray[Any](readKeyType)
- val values = mapData.valueArray().toArray[Any](readValueType)
+ val keys = mapData.keyArray().toArray[Any](writeKeyType)
+ val values = mapData.valueArray().toArray[Any](writeValueType)
val convertedKeys = extractCollection(keys, readKeyType, writeKeyType)
val convertedValues = extractCollection(values, readValueType, writeValueType)
ArrayBasedMapData(convertedKeys, convertedValues)
}
+ private def castElement(elem: Any, toType: DataType, fromType: DataType): Any = {
+ try {
+ val castExpr = Cast(Literal(elem, fromType), toType)
+ castExpr.eval(null)
+ } catch {
+ case _: Exception => elem
Review Comment:
done, let me know if its what you mean
--
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