Re: [PR] [SPARK-46905][SQL] Add dedicated class to keep column definition instead of StructField in Create/ReplaceTable command [spark]

["dtenedor (via GitHub)" <gi...@apache.org>]

dtenedor commented on code in PR #44935:
URL: https://github.com/apache/spark/pull/44935#discussion_r1470056757


##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ColumnDefinition.scala:
##########
@@ -0,0 +1,124 @@
+/*
+ * 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.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal, UnaryExpression, Unevaluable}
+import org.apache.spark.sql.catalyst.parser.ParserInterface
+import org.apache.spark.sql.catalyst.util.GeneratedColumn
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumnsUtils.{CURRENT_DEFAULT_COLUMN_METADATA_KEY, EXISTS_DEFAULT_COLUMN_METADATA_KEY}
+import org.apache.spark.sql.connector.catalog.{Column => V2Column, ColumnDefaultValue}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.connector.ColumnImpl
+import org.apache.spark.sql.types.{DataType, Metadata, MetadataBuilder, StructField}
+
+/**
+ * Column definition for tables. This is an expression so that analyzer can resolve the default
+ * value expression in DDL commands automatically.
+ */
+case class ColumnDefinition(
+    name: String,
+    dataType: DataType,
+    nullable: Boolean = true,
+    comment: Option[String] = None,
+    defaultValue: Option[DefaultValueExpression] = None,
+    generationExpression: Option[String] = None,
+    metadata: Metadata = Metadata.empty) extends Expression with Unevaluable {
+  override def children: Seq[Expression] = defaultValue.toSeq
+
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[Expression]): Expression = {
+    copy(defaultValue = newChildren.headOption.map(_.asInstanceOf[DefaultValueExpression]))
+  }
+
+  def toV2Column(statement: String): V2Column = {
+    ColumnImpl(
+      name,
+      dataType,
+      nullable,
+      comment.orNull,
+      defaultValue.map(_.toV2(statement, name)).orNull,
+      generationExpression.orNull,
+      if (metadata == Metadata.empty) null else metadata.json)
+  }
+
+  def toV1Column: StructField = {
+    val metadataBuilder = new MetadataBuilder().withMetadata(metadata)
+    comment.foreach { c =>
+      metadataBuilder.putString("comment", c)
+    }
+    defaultValue.foreach { default =>
+      // For v1 CREATE TABLE command, we will resolve and execute the default value expression later
+      // in the rule `DataSourceAnalysis`. We just need to put the default value SQL string here.

Review Comment:
   please briefly mention why we need to set both the "current default value" and "existence default value" column metadata here?



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala:
##########
@@ -445,6 +444,54 @@ object ResolveDefaultColumns extends QueryErrorsBase with ResolveDefaultColumnsU
   def hasExistenceDefaultValues(schema: StructType): Boolean =
     existenceDefaultValues(schema).exists(_ != null)
 
+
+  def checkColumnDefaultValues(plan: LogicalPlan): Unit = {
+    plan match {
+      // Do not check anything if the children are not resolved yet.
+      case _ if !plan.childrenResolved =>
+
+      case cmd: V2CreateTablePlan if cmd.columns.exists(_.defaultValue.isDefined) =>
+        val statement = cmd match {
+          case _: CreateTable => "CREATE TABLE"
+          case _: ReplaceTable => "REPLACE TABLE"
+          case other =>
+            val cmd = other.getClass.getSimpleName
+            throw SparkException.internalError(
+              s"Command $cmd should not have column default value expression.")
+        }
+        cmd.columns.foreach { col =>
+          col.defaultValue.foreach { default =>
+            validateDefaultValueExpr(default, statement, col.name, col.dataType)
+          }
+        }
+
+      case _ =>
+    }
+  }
+
+  private def validateDefaultValueExpr(
+      default: DefaultValueExpression,
+      statement: String,
+      colName: String,
+      targetType: DataType): Unit = {
+    if (default.containsPattern(PLAN_EXPRESSION)) {
+      throw QueryCompilationErrors.defaultValuesMayNotContainSubQueryExpressions(
+        statement, colName, default.originalSQL)
+    } else if (default.resolved) {
+      if (!Cast.canUpCast(default.child.dataType, targetType)) {
+        throw QueryCompilationErrors.defaultValuesDataTypeError(
+          statement, colName, default.originalSQL, targetType, default.child.dataType)
+      }
+      // Check passes. We do not check foldable here, as the plan is not optimized yet.
+    } else if (default.references.nonEmpty) {
+      // Ideally we should let the rest of `CheckAnalysis` to report errors about why the default

Review Comment:
   ```suggestion
         // Ideally we should let the rest of `CheckAnalysis` report errors about why the default
   ```



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ColumnDefinition.scala:
##########
@@ -0,0 +1,124 @@
+/*
+ * 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.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal, UnaryExpression, Unevaluable}
+import org.apache.spark.sql.catalyst.parser.ParserInterface
+import org.apache.spark.sql.catalyst.util.GeneratedColumn
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumnsUtils.{CURRENT_DEFAULT_COLUMN_METADATA_KEY, EXISTS_DEFAULT_COLUMN_METADATA_KEY}
+import org.apache.spark.sql.connector.catalog.{Column => V2Column, ColumnDefaultValue}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.connector.ColumnImpl
+import org.apache.spark.sql.types.{DataType, Metadata, MetadataBuilder, StructField}
+
+/**
+ * Column definition for tables. This is an expression so that analyzer can resolve the default
+ * value expression in DDL commands automatically.
+ */
+case class ColumnDefinition(
+    name: String,
+    dataType: DataType,
+    nullable: Boolean = true,
+    comment: Option[String] = None,
+    defaultValue: Option[DefaultValueExpression] = None,

Review Comment:
   Just a note that I've found when working with this code that the `Option` might confuse some of our Catalyst code that looks for expressions within operators. I think our `mapExpressions` [1] knows how to recurse into the `Option`, but our `resolved` method uses `QueryPlan.expressions`, which does not [2].
   
   It might simplify our work here to just skip the `Option` and make the field a normal `Expression` that defaults to literal NULL. We could have a separate boolean field like `hasExplicitDefault` to differentiate between when the `CREATE TABLE` command included an explicit `DEFAULT NULL` or not. Up to you.
   
   [1] https://github.com/apache/spark/blob/c468c3d5c685c5a5ecd7caf01f3004addce1f3b6/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala#L203
   
   [2] 
   https://github.com/apache/spark/blob/c468c3d5c685c5a5ecd7caf01f3004addce1f3b6/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala#L266



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala:
##########
@@ -445,6 +444,54 @@ object ResolveDefaultColumns extends QueryErrorsBase with ResolveDefaultColumnsU
   def hasExistenceDefaultValues(schema: StructType): Boolean =
     existenceDefaultValues(schema).exists(_ != null)
 
+
+  def checkColumnDefaultValues(plan: LogicalPlan): Unit = {
+    plan match {
+      // Do not check anything if the children are not resolved yet.
+      case _ if !plan.childrenResolved =>
+
+      case cmd: V2CreateTablePlan if cmd.columns.exists(_.defaultValue.isDefined) =>
+        val statement = cmd match {
+          case _: CreateTable => "CREATE TABLE"
+          case _: ReplaceTable => "REPLACE TABLE"
+          case other =>
+            val cmd = other.getClass.getSimpleName
+            throw SparkException.internalError(
+              s"Command $cmd should not have column default value expression.")
+        }
+        cmd.columns.foreach { col =>
+          col.defaultValue.foreach { default =>
+            validateDefaultValueExpr(default, statement, col.name, col.dataType)
+          }
+        }
+
+      case _ =>
+    }
+  }
+
+  private def validateDefaultValueExpr(
+      default: DefaultValueExpression,
+      statement: String,
+      colName: String,
+      targetType: DataType): Unit = {
+    if (default.containsPattern(PLAN_EXPRESSION)) {
+      throw QueryCompilationErrors.defaultValuesMayNotContainSubQueryExpressions(
+        statement, colName, default.originalSQL)
+    } else if (default.resolved) {
+      if (!Cast.canUpCast(default.child.dataType, targetType)) {
+        throw QueryCompilationErrors.defaultValuesDataTypeError(
+          statement, colName, default.originalSQL, targetType, default.child.dataType)
+      }
+      // Check passes. We do not check foldable here, as the plan is not optimized yet.

Review Comment:
   ```suggestion
         // Our analysis check passes here. We do not further inspect whether the
         // expression is `foldable` here, as the plan is not optimized yet.
   ```



##########
common/utils/src/main/resources/error/error-classes.json:
##########
@@ -1766,6 +1766,11 @@
           "which requires <expectedType> type, but the statement provided a value of incompatible <actualType> type."
         ]
       },
+      "NOT_CONSTANT" : {
+        "message" : [
+          "which is not a constant."

Review Comment:
   ```suggestion
             "which is not a constant expression whose equivalent value is known at query planning time; please update your command to change the column default value to satisfy this constraint and then retry the command again."
   ```



-- 
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