[SPARK-52235][SQL] Add implicit cast to DefaultValue V2 Expressions passed to DSV2 #50959
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -30,7 +30,7 @@ import org.apache.spark.sql.execution.{QueryExecution, SparkPlan} | |
| import org.apache.spark.sql.execution.ExplainUtils.stripAQEPlan | ||
| import org.apache.spark.sql.execution.datasources.v2.{AlterTableExec, CreateTableExec, DataSourceV2Relation, ReplaceTableExec} | ||
| import org.apache.spark.sql.internal.SQLConf | ||
| import org.apache.spark.sql.types.{BooleanType, CalendarIntervalType, DoubleType, IntegerType, StringType, TimestampType} | ||
| import org.apache.spark.sql.util.QueryExecutionListener | ||
| import org.apache.spark.unsafe.types.UTF8String | ||
|
|
||
|
|
@@ -498,43 +498,32 @@ class DataSourceV2DataFrameSuite | |
| |""".stripMargin) | ||
|
|
||
| val alterExecCol1 = executeAndKeepPhysicalPlan[AlterTableExec] { | ||
| sql( | ||
|
There was a problem hiding this comment. Some test cleanup, I didnt realize multiple ALTER COLUMN can be the same statement, so it was unnecessarily long |
||
| s""" | ||
| |ALTER TABLE $tableName ALTER COLUMN | ||
| | salary SET DEFAULT (123 + 56), | ||
| | dep SET DEFAULT ('r' || 'l'), | ||
| | active SET DEFAULT CAST(0 AS BOOLEAN) | ||
| |""".stripMargin) | ||
| } | ||
| checkDefaultValues( | ||
| alterExecCol1.changes.map(_.asInstanceOf[UpdateColumnDefaultValue]).toArray, | ||
| Array( | ||
| new DefaultValue( | ||
| "(123 + 56)", | ||
| new GeneralScalarExpression( | ||
| "+", | ||
| Array(LiteralValue(123, IntegerType), LiteralValue(56, IntegerType)))), | ||
| new DefaultValue( | ||
| "('r' || 'l')", | ||
| new GeneralScalarExpression( | ||
| "CONCAT", | ||
| Array( | ||
| LiteralValue(UTF8String.fromString("r"), StringType), | ||
| LiteralValue(UTF8String.fromString("l"), StringType)))), | ||
| new DefaultValue( | ||
| "CAST(0 AS BOOLEAN)", | ||
| new V2Cast(LiteralValue(0, IntegerType), IntegerType, BooleanType)))) | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -666,13 +655,9 @@ class DataSourceV2DataFrameSuite | |
| sql(s"ALTER TABLE $tableName ALTER COLUMN cat SET DEFAULT current_catalog()") | ||
| } | ||
|
|
||
| checkDefaultValues( | ||
| alterExec.changes.map(_.asInstanceOf[UpdateColumnDefaultValue]).toArray, | ||
| Array(new DefaultValue("current_catalog()", null /* No V2 Expression */))) | ||
|
|
||
| val df1 = Seq(1).toDF("dummy") | ||
| df1.writeTo(tableName).append() | ||
|
|
@@ -683,6 +668,109 @@ class DataSourceV2DataFrameSuite | |
| } | ||
| } | ||
|
|
||
| test("create/replace table default value expression should have a cast") { | ||
|
There was a problem hiding this comment. Tests for
We can execute the following command to reproduce the issue locally: Do you have time to take a look? @szehon-ho |
||
| val tableName = "testcat.ns1.ns2.tbl" | ||
| withTable(tableName) { | ||
|
|
||
| val createExec = executeAndKeepPhysicalPlan[CreateTableExec] { | ||
| sql( | ||
| s""" | ||
| |CREATE TABLE $tableName ( | ||
| | col1 int, | ||
| | col2 timestamp DEFAULT '2018-11-17 13:33:33', | ||
| | col3 double DEFAULT 1) | ||
| |""".stripMargin) | ||
| } | ||
| checkDefaultValues( | ||
| createExec.columns, | ||
| Array( | ||
| null, | ||
| new ColumnDefaultValue( | ||
| "'2018-11-17 13:33:33'", | ||
| new LiteralValue(1542490413000000L, TimestampType), | ||
| new LiteralValue(1542490413000000L, TimestampType)), | ||
| new ColumnDefaultValue( | ||
| "1", | ||
| new V2Cast(LiteralValue(1, IntegerType), IntegerType, DoubleType), | ||
| LiteralValue(1.0, DoubleType)))) | ||
|
|
||
| val replaceExec = executeAndKeepPhysicalPlan[ReplaceTableExec] { | ||
| sql( | ||
| s""" | ||
| |REPLACE TABLE $tableName ( | ||
| | col1 int, | ||
| | col2 timestamp DEFAULT '2022-02-23 05:55:55', | ||
| | col3 double DEFAULT (1 + 1)) | ||
| |""".stripMargin) | ||
| } | ||
| checkDefaultValues( | ||
| replaceExec.columns, | ||
| Array( | ||
| null, | ||
| new ColumnDefaultValue( | ||
| "'2022-02-23 05:55:55'", | ||
| LiteralValue(1645624555000000L, TimestampType), | ||
| LiteralValue(1645624555000000L, TimestampType)), | ||
| new ColumnDefaultValue( | ||
| "(1 + 1)", | ||
| new V2Cast( | ||
| new GeneralScalarExpression("+", Array(LiteralValue(1, IntegerType), | ||
| LiteralValue(1, IntegerType))), | ||
| IntegerType, | ||
| DoubleType), | ||
| LiteralValue(2.0, DoubleType)))) | ||
| } | ||
| } | ||
|
|
||
| test("alter table default value expression should have a cast") { | ||
| val tableName = "testcat.ns1.ns2.tbl" | ||
| withTable(tableName) { | ||
|
|
||
| sql(s"CREATE TABLE $tableName (col1 int) using foo") | ||
| val alterExec = executeAndKeepPhysicalPlan[AlterTableExec] { | ||
| sql( | ||
| s""" | ||
| |ALTER TABLE $tableName ADD COLUMNS ( | ||
| | col2 timestamp DEFAULT '2018-11-17 13:33:33', | ||
| | col3 double DEFAULT 1) | ||
| |""".stripMargin) | ||
| } | ||
|
|
||
| checkDefaultValues( | ||
| alterExec.changes.map(_.asInstanceOf[AddColumn]).toArray, | ||
| Array( | ||
| new ColumnDefaultValue( | ||
| "'2018-11-17 13:33:33'", | ||
| LiteralValue(1542490413000000L, TimestampType), | ||
| LiteralValue(1542490413000000L, TimestampType)), | ||
| new ColumnDefaultValue( | ||
| "1", | ||
| new V2Cast(LiteralValue(1, IntegerType), IntegerType, DoubleType), | ||
| LiteralValue(1.0, DoubleType)))) | ||
|
|
||
| val alterCol1 = executeAndKeepPhysicalPlan[AlterTableExec] { | ||
| sql( | ||
| s""" | ||
| |ALTER TABLE $tableName ALTER COLUMN | ||
| | col2 SET DEFAULT '2022-02-23 05:55:55', | ||
| | col3 SET DEFAULT (1 + 1) | ||
| |""".stripMargin) | ||
| } | ||
| checkDefaultValues( | ||
| alterCol1.changes.map(_.asInstanceOf[UpdateColumnDefaultValue]).toArray, | ||
| Array( | ||
| new DefaultValue("'2022-02-23 05:55:55'", | ||
| LiteralValue(1645624555000000L, TimestampType)), | ||
| new DefaultValue( | ||
| "(1 + 1)", | ||
| new V2Cast( | ||
| new GeneralScalarExpression("+", Array(LiteralValue(1, IntegerType), | ||
| LiteralValue(1, IntegerType))), | ||
| IntegerType, | ||
| DoubleType)))) | ||
| } | ||
| } | ||
|
|
||
| private def executeAndKeepPhysicalPlan[T <: SparkPlan](func: => Unit): T = { | ||
| val qe = withQueryExecutionsCaptured(spark) { | ||
| func | ||
|
|
@@ -718,13 +806,18 @@ class DataSourceV2DataFrameSuite | |
| } | ||
| } | ||
|
|
||
| private def checkDefaultValues( | ||
|
There was a problem hiding this comment. I guess you could define new There was a problem hiding this comment. i think i prefer to use the new method in all the place, as it makes the test significantly shorter (no need to check each change individually) |
||
| columns: Array[UpdateColumnDefaultValue], | ||
| expectedDefaultValues: Array[DefaultValue]): Unit = { | ||
| assert(columns.length == expectedDefaultValues.length) | ||
|
|
||
| columns.zip(expectedDefaultValues).foreach { | ||
| case (column, expectedDefault) => | ||
| assert( | ||
| column.newCurrentDefault() == expectedDefault, | ||
| s"Default value mismatch for column '${column.toString}': " + | ||
| s"expected $expectedDefault but found ${column.newCurrentDefault}") | ||
| } | ||
| } | ||
|
|
||
| private def checkDropDefaultValue( | ||
|
|
||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.