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 2025/05/09 04:51:14 UTC
(spark) branch master updated: [SPARK-51906][SQL] Dsv2 expressions in alter table add columns
This is an automated email from the ASF dual-hosted git repository.
wenchen 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 1cec8567df1f [SPARK-51906][SQL] Dsv2 expressions in alter table add columns
1cec8567df1f is described below
commit 1cec8567df1f59e77865c7d26b03aad435273b15
Author: Szehon Ho <sz...@gmail.com>
AuthorDate: Fri May 9 12:51:01 2025 +0800
[SPARK-51906][SQL] Dsv2 expressions in alter table add columns
### What changes were proposed in this pull request?
Add DSV2 expression to alter table add columm.
This harmonizes with the Create/Replace table default value code path (#50593) and now changes the plan to:
1. Change from modeling default value from string to DefaultValueExpression
2. Change its parent QualifiedColType to Expression so that it gets handled by the main loop
This will then analyze/resolve the user-provided string to an Expression which can neatly be converted to V2Expression.
It also has side effect that the existsDefault will use this as well (instead of being analyzed/optimized by manually calling ResolveDefaultColumns::analyze), hence the original exception flow in that method is now integrated into various other rules in the analyzer phases.
### Why are the changes needed?
Data sources implementing DSV2 need to get a V2 expression to interpret the default value.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Add new test in DataSourceV2DataFrameSuite
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #50701 from szehon-ho/add_column_default_val.
Authored-by: Szehon Ho <sz...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/parser/AstBuilder.scala | 2 +-
.../catalyst/plans/logical/ColumnDefinition.scala | 12 +++
.../sql/catalyst/plans/logical/statements.scala | 25 +++---
.../plans/logical/v2AlterTableCommands.scala | 4 +-
.../spark/sql/catalyst/parser/DDLParserSuite.scala | 5 +-
.../catalyst/analysis/ResolveSessionCatalog.scala | 3 +-
.../sql/connector/DataSourceV2DataFrameSuite.scala | 89 +++++++++++++++++++++-
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 11 ++-
8 files changed, 125 insertions(+), 26 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index b15c5b17332c..389956bb931d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -5145,7 +5145,7 @@ class AstBuilder extends DataTypeAstBuilder
// Add the 'DEFAULT expression' clause in the column definition, if any, to the column metadata.
val defaultExpr = defaultExpression.map(visitDefaultExpression).map { field =>
if (conf.getConf(SQLConf.ENABLE_DEFAULT_COLUMNS)) {
- field.originalSQL
+ field
} else {
throw QueryParsingErrors.defaultColumnNotEnabledError(ctx)
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ColumnDefinition.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ColumnDefinition.scala
index 83219d188db8..67b4ca331774 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ColumnDefinition.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ColumnDefinition.scala
@@ -164,6 +164,18 @@ object ColumnDefinition {
}
}
+ case cmd: AddColumns if cmd.columnsToAdd.exists(_.default.isDefined) =>
+ // Wrap analysis errors for default values in a more user-friendly message.
+ cmd.columnsToAdd.foreach { c =>
+ c.default.foreach { d =>
+ if (!d.resolved) {
+ throw QueryCompilationErrors.defaultValuesUnresolvedExprError(
+ "ALTER TABLE", c.colName, d.originalSQL, null)
+ }
+ validateDefaultValueExpr(d, "ALTER TABLE", c.colName, c.dataType)
+ }
+ }
+
case _ =>
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala
index 9efc3b13bc26..0afc2227cb67 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala
@@ -18,11 +18,9 @@
package org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.sql.catalyst.analysis.{FieldName, FieldPosition}
-import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, Unevaluable}
import org.apache.spark.sql.catalyst.trees.{LeafLike, UnaryLike}
-import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns
import org.apache.spark.sql.connector.catalog.ColumnDefaultValue
-import org.apache.spark.sql.connector.expressions.LiteralValue
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.types.DataType
@@ -134,19 +132,20 @@ case class QualifiedColType(
nullable: Boolean,
comment: Option[String],
position: Option[FieldPosition],
- default: Option[String]) {
+ default: Option[DefaultValueExpression]) extends Expression with Unevaluable {
+ override lazy val resolved: Boolean = path.forall(_.resolved) && position.forall(_.resolved) &&
+ default.forall(_.resolved)
+
def name: Seq[String] = path.map(_.name).getOrElse(Nil) :+ colName
- def resolved: Boolean = path.forall(_.resolved) && position.forall(_.resolved)
+ def getV2Default(statement: String): ColumnDefaultValue =
+ default.map(_.toV2(statement, colName)).orNull
+
+ override def children: Seq[Expression] = default.toSeq
- def getV2Default: ColumnDefaultValue = {
- default.map { sql =>
- val e = ResolveDefaultColumns.analyze(colName, dataType, sql, "ALTER TABLE")
- assert(e.resolved && e.foldable,
- "The existence default value must be a simple SQL string that is resolved and foldable, " +
- "but got: " + sql)
- new ColumnDefaultValue(sql, LiteralValue(e.eval(), dataType))
- }.orNull
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[Expression]): Expression = {
+ copy(default = newChildren.headOption.map(_.asInstanceOf[DefaultValueExpression]))
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2AlterTableCommands.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2AlterTableCommands.scala
index bc3e86351b11..4bd6588c0104 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2AlterTableCommands.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2AlterTableCommands.scala
@@ -123,7 +123,7 @@ case class AddColumns(
col.nullable,
col.comment.orNull,
col.position.map(_.position).orNull,
- col.getV2Default)
+ col.getV2Default("ALTER TABLE"))
}
}
@@ -159,7 +159,7 @@ case class ReplaceColumns(
col.nullable,
col.comment.orNull,
null,
- col.getV2Default)
+ col.getV2Default("ALTER TABLE"))
}
(deleteChanges ++ addChanges).toImmutableArraySeq
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
index 1589bcb8a3d7..9e8e63dae9e1 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
@@ -2723,7 +2723,8 @@ class DDLParserSuite extends AnalysisTest {
comparePlans(
parsePlan("ALTER TABLE t1 ADD COLUMN x int NOT NULL DEFAULT 42"),
AddColumns(UnresolvedTable(Seq("t1"), "ALTER TABLE ... ADD COLUMN"),
- Seq(QualifiedColType(None, "x", IntegerType, false, None, None, Some("42")))))
+ Seq(QualifiedColType(None, "x", IntegerType, false, None, None,
+ Some(DefaultValueExpression(Literal(42), "42"))))))
comparePlans(
parsePlan("ALTER TABLE t1 ALTER COLUMN a.b.c SET DEFAULT 42"),
AlterColumns(
@@ -3138,7 +3139,7 @@ class DDLParserSuite extends AnalysisTest {
nullable = false,
comment = Some("a"),
position = Some(UnresolvedFieldPosition(first())),
- default = Some("'abc'")
+ default = Some(DefaultValueExpression(Literal("abc"), "'abc'"))
)
)
)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
index 03ce8882f2fa..64cc9da1f189 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
@@ -699,7 +699,8 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
val builder = new MetadataBuilder
col.comment.foreach(builder.putString("comment", _))
col.default.map {
- value: String => builder.putString(DefaultCols.CURRENT_DEFAULT_COLUMN_METADATA_KEY, value)
+ value: DefaultValueExpression => builder.putString(
+ DefaultCols.CURRENT_DEFAULT_COLUMN_METADATA_KEY, value.originalSQL)
}
StructField(col.name.head, col.dataType, nullable = true, builder.build())
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
index b6e27aea16c7..ba5d3cd940ab 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
@@ -24,10 +24,11 @@ import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
import org.apache.spark.sql.catalyst.plans.logical.{AppendData, CreateTableAsSelect, LogicalPlan, ReplaceTableAsSelect}
import org.apache.spark.sql.connector.catalog.{Column, ColumnDefaultValue, Identifier, InMemoryTableCatalog}
+import org.apache.spark.sql.connector.catalog.TableChange.AddColumn
import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, GeneralScalarExpression, LiteralValue, Transform}
import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
import org.apache.spark.sql.execution.ExplainUtils.stripAQEPlan
-import org.apache.spark.sql.execution.datasources.v2.{CreateTableExec, DataSourceV2Relation, ReplaceTableExec}
+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, IntegerType, StringType}
import org.apache.spark.sql.util.QueryExecutionListener
@@ -438,6 +439,47 @@ class DataSourceV2DataFrameSuite
}
}
+ test("alter table with complex foldable default values") {
+ val tableName = "testcat.ns1.ns2.tbl"
+ withTable(tableName) {
+ sql(
+ s"""
+ |CREATE TABLE $tableName (
+ | dummy INT
+ |) USING foo
+ |""".stripMargin)
+
+ val alterExec = executeAndKeepPhysicalPlan[AlterTableExec] {
+ sql(s"ALTER TABLE $tableName ADD COLUMNS (" +
+ s"salary INT DEFAULT (100 + 23), " +
+ s"dep STRING DEFAULT ('h' || 'r'), " +
+ s"active BOOLEAN DEFAULT CAST(1 AS BOOLEAN))")
+ }
+
+ checkDefaultValues(
+ alterExec.changes.map(_.asInstanceOf[AddColumn]).toArray,
+ Array(
+ new ColumnDefaultValue(
+ "(100 + 23)",
+ new GeneralScalarExpression(
+ "+",
+ Array(LiteralValue(100, IntegerType), LiteralValue(23, IntegerType))),
+ LiteralValue(123, IntegerType)),
+ new ColumnDefaultValue(
+ "('h' || 'r')",
+ new GeneralScalarExpression(
+ "CONCAT",
+ Array(
+ LiteralValue(UTF8String.fromString("h"), StringType),
+ LiteralValue(UTF8String.fromString("r"), StringType))),
+ LiteralValue(UTF8String.fromString("hr"), StringType)),
+ new ColumnDefaultValue(
+ "CAST(1 AS BOOLEAN)",
+ new V2Cast(LiteralValue(1, IntegerType), IntegerType, BooleanType),
+ LiteralValue(true, BooleanType))))
+ }
+ }
+
test("create/replace table with current like default values") {
val tableName = "testcat.ns1.ns2.tbl"
withTable(tableName) {
@@ -483,6 +525,37 @@ class DataSourceV2DataFrameSuite
}
}
+ test("alter table with current like default values") {
+ val tableName = "testcat.ns1.ns2.tbl"
+ withTable(tableName) {
+ sql(
+ s"""
+ |CREATE TABLE $tableName (
+ | dummy INT
+ |) USING foo
+ |""".stripMargin)
+
+ val alterExec = executeAndKeepPhysicalPlan[AlterTableExec] {
+ sql(s"ALTER TABLE $tableName ADD COLUMNS (cat STRING DEFAULT current_catalog())")
+ }
+
+ checkDefaultValues(
+ alterExec.changes.map(_.asInstanceOf[AddColumn]).toArray,
+ Array(
+ new ColumnDefaultValue(
+ "current_catalog()",
+ null, /* no V2 expression */
+ LiteralValue(UTF8String.fromString("spark_catalog"), StringType))))
+
+ val df1 = Seq(1).toDF("dummy")
+ df1.writeTo(tableName).append()
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableName"),
+ Seq(Row(1, "spark_catalog")))
+ }
+ }
+
private def executeAndKeepPhysicalPlan[T <: SparkPlan](func: => Unit): T = {
val qe = withQueryExecutionsCaptured(spark) {
func
@@ -503,4 +576,18 @@ class DataSourceV2DataFrameSuite
s"expected $expectedDefault but found ${column.defaultValue}")
}
}
+
+ private def checkDefaultValues(
+ columns: Array[AddColumn],
+ expectedDefaultValues: Array[ColumnDefaultValue]): Unit = {
+ assert(columns.length == expectedDefaultValues.length)
+
+ columns.zip(expectedDefaultValues).foreach {
+ case (column, expectedDefault) =>
+ assert(
+ column.defaultValue == expectedDefault,
+ s"Default value mismatch for column '${column.toString}': " +
+ s"expected $expectedDefault but found ${column.defaultValue}")
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 747ecd7d9833..07740410f24a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -3552,7 +3552,7 @@ class DataSourceV2SQLSuiteV1Filter
}
}
- test("SPARK-48286: Add new column with default value which is not foldable") {
+ test("SPARK-48286: Add new column with default value which is not deterministic") {
val foldableExpressions = Seq("1", "2 + 1")
withSQLConf(SQLConf.DEFAULT_COLUMN_ALLOWED_PROVIDERS.key -> v2Source) {
withTable("tab") {
@@ -3560,11 +3560,10 @@ class DataSourceV2SQLSuiteV1Filter
val exception = analysisException(
// Rand function is not foldable
s"ALTER TABLE tab ADD COLUMN col2 DOUBLE DEFAULT rand()")
- assert(exception.getSqlState == "42623")
- assert(exception.errorClass.get == "INVALID_DEFAULT_VALUE.NOT_CONSTANT")
- assert(exception.messageParameters("colName") == "`col2`")
- assert(exception.messageParameters("defaultValue") == "rand()")
- assert(exception.messageParameters("statement") == "ALTER TABLE")
+ assert(exception.getSqlState == "42K0E")
+ assert(exception.errorClass.get == "INVALID_NON_DETERMINISTIC_EXPRESSIONS")
+ assert(exception.messageParameters("sqlExprs") ==
+ "\"qualifiedcoltype(defaultvalueexpression(rand()))\"")
}
foldableExpressions.foreach(expr => {
withTable("tab") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org