dongjoon-hyun commented on code in PR #59066:
URL: https://github.com/apache/spark/pull/59066#discussion_r4110940313
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala:
##########
@@ -976,7 +977,20 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends
LeafV2CommandExec {
protected def getV2Columns(schema: StructType, forceNullable: Boolean):
Array[Column] = {
Review Comment:
Related to the comment below, the non-atomic `ReplaceTableAsSelectExec`
calls `getV2Columns` after `catalog.dropTable(ident)`. With the new failure
path in this method, the existing table could be dropped without being
re-created. Could we compute the columns before dropping the table, like
`AtomicReplaceTableAsSelectExec` does?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala:
##########
@@ -976,7 +977,20 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends
LeafV2CommandExec {
protected def getV2Columns(schema: StructType, forceNullable: Boolean):
Array[Column] = {
val rawSchema =
CharVarcharUtils.getRawSchema(removeInternalMetadata(schema), conf)
val tableSchema = if (forceNullable) rawSchema.asNullable else rawSchema
- CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds = false)
+ val columns = CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds =
false)
+ columns.zip(tableSchema.fields).map {
+ case (column, field) if column.defaultValue() != null =>
+ val defaultValue = column.defaultValue()
+ val (_, expr) =
Review Comment:
This expression is not re-analyzed or validated against the target column
type. `CREATE TABLE` builds the V2 expression from the analyzed and
type-coerced `analyzedChild`, but here we pass whatever the source connector
reported after a `toCatalyst` -> `V2ExpressionBuilder` round trip.
For example, if a connector reports `LiteralValue(42, IntegerType)` for a
`BIGINT` column, the target receives `ColumnDefaultValue(sql="42", expr=<INT
literal>, value=42L)`. Should we check at least that the expression data type
matches the target column type?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala:
##########
@@ -976,7 +977,20 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends
LeafV2CommandExec {
protected def getV2Columns(schema: StructType, forceNullable: Boolean):
Array[Column] = {
val rawSchema =
CharVarcharUtils.getRawSchema(removeInternalMetadata(schema), conf)
val tableSchema = if (forceNullable) rawSchema.asNullable else rawSchema
- CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds = false)
+ val columns = CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds =
false)
+ columns.zip(tableSchema.fields).map {
+ case (column, field) if column.defaultValue() != null =>
+ val defaultValue = column.defaultValue()
+ val (_, expr) =
+
field.metadata.getExpression[Expression](CURRENT_DEFAULT_COLUMN_METADATA_KEY)
+ expr.flatMap(new V2ExpressionBuilder(_).build()).map { currentDefault
=>
Review Comment:
`V2ExpressionBuilder.build()` can throw here. With
`spark.sql.optimizer.datasourceV2ExprFolding=true` (default),
`ConstantFolding.constantFolding` evaluates the expression, and `tryFold`
swallows exceptions only inside conditional branches.
Since this expression comes from `V2ExpressionUtils.toCatalyst(<connector
expression>)` without any analysis, IIUC, a connector-reported default like
`ABS(-2147483648)` or `1 / 0` would make `CREATE TABLE ... AS SELECT * FROM
src` fail, which succeeded before this PR because only the SQL text was passed.
Shall we fall back to the SQL-only default on `NonFatal` errors?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala:
##########
@@ -976,7 +977,20 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends
LeafV2CommandExec {
protected def getV2Columns(schema: StructType, forceNullable: Boolean):
Array[Column] = {
val rawSchema =
CharVarcharUtils.getRawSchema(removeInternalMetadata(schema), conf)
val tableSchema = if (forceNullable) rawSchema.asNullable else rawSchema
- CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds = false)
+ val columns = CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds =
false)
+ columns.zip(tableSchema.fields).map {
+ case (column, field) if column.defaultValue() != null =>
+ val defaultValue = column.defaultValue()
+ val (_, expr) =
+
field.metadata.getExpression[Expression](CURRENT_DEFAULT_COLUMN_METADATA_KEY)
Review Comment:
Since `Metadata.runtimeMap` is `@transient` and lost by JSON round trips,
the result depends on how the source metadata was produced rather than on the
default value itself. For example, CTAS from a v1 source table gets
`expression=null` with `HiveExternalCatalog`, while it may get the expression
with the in-memory catalog. DataFrames with JSON schemas (PySpark/Connect) and
SQL-only connectors also get `null`.
Have you considered analyzing the SQL text (without folding) when the
runtime expression is absent, to be consistent with `CREATE TABLE`?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala:
##########
@@ -935,6 +936,95 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ gridTest("CTAS and RTAS preserve column default SQL and expressions")(
+ Seq(true, false)) { foldExpressions =>
+ withSQLConf("spark.sql.optimizer.datasourceV2ExprFolding" ->
foldExpressions.toString) {
+ val arithmeticDefault = if (foldExpressions) {
+ LiteralValue(42L, LongType)
+ } else {
+ new V2Cast(
+ new GeneralScalarExpression("+", Array(
+ LiteralValue(40, IntegerType), LiteralValue(2, IntegerType))),
+ IntegerType, LongType)
+ }
+ val functionDefault = if (foldExpressions) {
+ LiteralValue(42, IntegerType)
+ } else {
+ new GeneralScalarExpression("ABS", Array(LiteralValue(-42,
IntegerType)))
+ }
+ Seq("testcat", "testcat_atomic").foreach { catalogName =>
+ val source = s"$catalogName.default_source"
+ val target = s"$catalogName.default_target"
+ val testCatalog = catalog(catalogName).asTableCatalog
+ withTable(source, target) {
+ sql(s"""CREATE TABLE $source (
+ | id INT,
+ | literal_default INT DEFAULT 42,
+ | arithmetic_default BIGINT DEFAULT (40 + 2),
+ | function_default INT DEFAULT abs(-42),
+ | sql_only_default DATE DEFAULT current_date()
+ |) USING foo""".stripMargin)
+ val sourceDefaults = testCatalog.loadTable(
+ Identifier.of(Array.empty[String],
"default_source")).columns().map(_.defaultValue())
+ assert(sourceDefaults(1).getExpression == LiteralValue(42,
IntegerType))
+ assert(sourceDefaults(2) == new ColumnDefaultValue(
+ "(40 + 2)", arithmeticDefault, LiteralValue(42L, LongType)))
+ assert(sourceDefaults(3) == new ColumnDefaultValue(
+ "abs(-42)", functionDefault, LiteralValue(42, IntegerType)))
+ assert(sourceDefaults(4).getExpression == null)
+
+ Seq("CREATE", "REPLACE", "CREATE OR REPLACE").foreach { command =>
+ sql(s"$command TABLE $target USING foo AS SELECT * FROM $source")
+ val targetDefaults = testCatalog.loadTable(
+ Identifier.of(Array.empty[String], "default_target"))
+
.asInstanceOf[InMemoryBaseTable].initialColumns.map(_.defaultValue())
+ assert(targetDefaults.toSeq == sourceDefaults.toSeq)
+ checkAnswer(spark.table(target), spark.table(source))
Review Comment:
The source table has no rows, so this `checkAnswer` compares two empty
tables.
In addition, Spark prefers the expression over the SQL text when it analyzes
the default value of a V2 table. Could you add something like `INSERT INTO
$target(id) VALUES (1)` and verify the default values end to end?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala:
##########
@@ -935,6 +936,95 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ gridTest("CTAS and RTAS preserve column default SQL and expressions")(
Review Comment:
Could you add test coverage for the following?
- V2 session catalog (`spark_catalog` with a V2 provider or a
`DelegatingCatalogExtension`)
- A v1 source table to a v2 target table
- DataFrameWriterV2 (`writeTo(...).create()` / `createOrReplace()`)
- Projections like `SELECT c AS renamed` and `SELECT c + 1 AS c`
- A negative case where the expression type doesn't match the column type
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala:
##########
@@ -976,7 +977,20 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends
LeafV2CommandExec {
protected def getV2Columns(schema: StructType, forceNullable: Boolean):
Array[Column] = {
val rawSchema =
CharVarcharUtils.getRawSchema(removeInternalMetadata(schema), conf)
val tableSchema = if (forceNullable) rawSchema.asNullable else rawSchema
- CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds = false)
+ val columns = CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds =
false)
+ columns.zip(tableSchema.fields).map {
+ case (column, field) if column.defaultValue() != null =>
+ val defaultValue = column.defaultValue()
+ val (_, expr) =
+
field.metadata.getExpression[Expression](CURRENT_DEFAULT_COLUMN_METADATA_KEY)
+ expr.flatMap(new V2ExpressionBuilder(_).build()).map { currentDefault
=>
+ Column.builderFrom(column)
+ .defaultValue(new ColumnDefaultValue(
Review Comment:
Minor: the round trip re-folds the expression under the current session conf
and drops V2 expressions which `toCatalyst` doesn't support (e.g.,
`UserDefinedScalarFunc`). For example, a source default `ABS(-42)` becomes
`LiteralValue(42)` in the target, while `CREATE TABLE LIKE` copies the source
columns as-is via `CatalogV2Util.clearIds(sourceTable.columns)`. The new test
compares source and target only under the same folding conf.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala:
##########
@@ -976,7 +977,20 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends
LeafV2CommandExec {
protected def getV2Columns(schema: StructType, forceNullable: Boolean):
Array[Column] = {
val rawSchema =
CharVarcharUtils.getRawSchema(removeInternalMetadata(schema), conf)
val tableSchema = if (forceNullable) rawSchema.asNullable else rawSchema
- CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds = false)
+ val columns = CatalogV2Util.structTypeToV2Columns(tableSchema, keepIds =
false)
Review Comment:
This duplicates the logic of `DefaultValueExpression.toV2`. Also, the other
`structTypeToV2Columns` callers, `CreateTableLikeExec` (V1Table source) and
`DatasetManager.createTable` in pipelines, still drop the expression. What
about adding an opt-in helper in `CatalogV2Util` (like the first commit of this
PR) to handle them consistently?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
