[SPARK-52235][SQL] Add implicit cast to DefaultValue V2 Expressions passed to DSV2

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AnsiTypeCoercion.scala

@@ -78,6 +78,7 @@ object AnsiTypeCoercion extends TypeCoercionBase {
     UnpivotCoercion ::
     WidenSetOperationTypes ::
     ProcedureArgumentCoercion ::
+    DefaultValueExpressionCoercion ::
     new AnsiCombinedTypeCoercionRule(
       CollationTypeCasts ::
       InConversion ::

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala

@@ -50,6 +50,7 @@ object TypeCoercion extends TypeCoercionBase {
     UnpivotCoercion ::
     WidenSetOperationTypes ::
     ProcedureArgumentCoercion ::
+    DefaultValueExpressionCoercion ::
     new CombinedTypeCoercionRule(
       CollationTypeCasts ::
       InConversion ::

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionBase.scala

@@ -32,16 +32,22 @@ import org.apache.spark.sql.catalyst.expressions.{
   WindowSpecDefinition
 }
 import org.apache.spark.sql.catalyst.plans.logical.{
+  AddColumns,
+  AlterColumns,
   Call,
+  CreateTable,
   Except,
   Intersect,
   LogicalPlan,
   Project,
+  ReplaceTable,
   Union,
   Unpivot
 }
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
 import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure
 import org.apache.spark.sql.types.DataType
 
@@ -81,6 +87,71 @@ abstract class TypeCoercionBase extends TypeCoercionHelper {
     }
   }
 
+  /**
+   * A type coercion rule that implicitly casts default value expression in DDL statements
+   * to expected types.
+   */
+  object DefaultValueExpressionCoercion extends Rule[LogicalPlan] {
+    override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+      case createTable @ CreateTable(_, cols, _, _, _) if createTable.resolved &&
+        cols.exists(_.defaultValue.isDefined) =>
+        val newCols = cols.map { c =>
+          c.copy(defaultValue = c.defaultValue.map(d =>
+            d.copy(child = ResolveDefaultColumns.coerceDefaultValue(
+              d.child,
+              c.dataType,
+              "CREATE TABLE",
+              c.name,
+              d.originalSQL,
+              castWiderOnlyLiterals = false))))
+        }
+        createTable.copy(columns = newCols)
+
+      case replaceTable @ ReplaceTable(_, cols, _, _, _) if replaceTable.resolved &&
+        cols.exists(_.defaultValue.isDefined) =>
+        val newCols = cols.map { c =>
+          c.copy(defaultValue = c.defaultValue.map(d =>
+            d.copy(child = ResolveDefaultColumns.coerceDefaultValue(
+              d.child,
+              c.dataType,
+              "REPLACE TABLE",
+              c.name,
+              d.originalSQL,
+              castWiderOnlyLiterals = false))))
+        }
+        replaceTable.copy(columns = newCols)
+
+      case addColumns @ AddColumns(_, cols) if addColumns.resolved &&
+        cols.exists(_.default.isDefined) =>
+        val newCols = cols.map { c =>
+          c.copy(default = c.default.map(d =>
+            d.copy(child = ResolveDefaultColumns.coerceDefaultValue(
+              d.child,
+              c.dataType,
+              "ALTER TABLE ADD COLUMNS",
+              c.colName,
+              d.originalSQL,
+              castWiderOnlyLiterals = false))))
+        }
+        addColumns.copy(columnsToAdd = newCols)
+
+      case alterColumns @ AlterColumns(_, specs) if alterColumns.resolved &&
+        specs.exists(_.newDefaultExpression.isDefined) =>
+        val newSpecs = specs.map { c =>
+          val dataType = c.column.asInstanceOf[ResolvedFieldName].field.dataType
+          c.copy(newDefaultExpression = c.newDefaultExpression.map(d =>
+            d.copy(child = ResolveDefaultColumns.coerceDefaultValue(
+              d.child,
+              dataType,
+              "ALTER TABLE ALTER COLUMN",
+              c.column.name.quoted,
+              d.originalSQL,
+              castWiderOnlyLiterals = false))))
+        }
+        alterColumns.copy(specs = newSpecs)
+    }
+  }
+
   /**
    * Widens the data types of the [[Unpivot]] values.
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala

@@ -432,18 +432,34 @@ object ResolveDefaultColumns extends QueryErrorsBase
       targetType: DataType,
       colName: String): Option[Expression] = {
     expr match {
-      case l: Literal if !Seq(targetType, l.dataType).exists(_ match {
+      case l: Literal => defaultValueFromWiderType(l, targetType, colName)
+      case _ => None
+    }
+  }
+
+  /**
+   * If the provided default value is a literal of a wider type than the target column,
+   * but the literal value fits within the narrower type, just coerce it for convenience.
+   * Exclude boolean/array/struct/map types from consideration for this type coercion to
+   * avoid surprising behavior like interpreting "false" as integer zero.
+   */
+  def defaultValueFromWiderType(
+      expr: Expression,
+      targetType: DataType,
+      colName: String): Option[Expression] = {
+    expr match {
+      case e if !Seq(targetType, e.dataType).exists(_ match {
         case _: BooleanType | _: ArrayType | _: StructType | _: MapType => true
         case _ => false
       }) =>
-        val casted = Cast(l, targetType, Some(conf.sessionLocalTimeZone), evalMode = EvalMode.TRY)
+        val casted = Cast(e, targetType, Some(conf.sessionLocalTimeZone), evalMode = EvalMode.TRY)
         try {
           Option(casted.eval(EmptyRow)).map(Literal(_, targetType))
         } catch {
           case e @ ( _: SparkThrowable | _: RuntimeException) =>
-            logWarning(log"Failed to cast default value '${MDC(COLUMN_DEFAULT_VALUE, l)}' " +
+            logWarning(log"Failed to cast default value '${MDC(COLUMN_DEFAULT_VALUE, e)}' " +
               log"for column ${MDC(COLUMN_NAME, colName)} " +
-              log"from ${MDC(COLUMN_DATA_TYPE_SOURCE, l.dataType)} " +
+              log"from ${MDC(COLUMN_DATA_TYPE_SOURCE, expr.dataType)} " +
               log"to ${MDC(COLUMN_DATA_TYPE_TARGET, targetType)} " +
               log"due to ${MDC(ERROR, e.getMessage)}", e)
             None
@@ -461,7 +477,8 @@ object ResolveDefaultColumns extends QueryErrorsBase
       dataType: DataType,
       statementType: String,
       colName: String,
-      defaultSQL: String): Expression = {
+      defaultSQL: String,
+      castWiderOnlyLiterals: Boolean = true): Expression = {
     val supplanted = CharVarcharUtils.replaceCharVarcharWithString(dataType)
     // Perform implicit coercion from the provided expression type to the required column type.
     val ret = analyzed match {
@@ -470,7 +487,12 @@ object ResolveDefaultColumns extends QueryErrorsBase
       case canUpCast if Cast.canUpCast(canUpCast.dataType, supplanted) =>
         Cast(analyzed, supplanted, Some(conf.sessionLocalTimeZone))
       case other =>
-        defaultValueFromWiderTypeLiteral(other, supplanted, colName).getOrElse(
+        val casted = if (castWiderOnlyLiterals) {
+          defaultValueFromWiderTypeLiteral(other, supplanted, colName)
+        } else {
+          defaultValueFromWiderType(other, supplanted, colName)
+        }
+        casted.getOrElse(
           throw QueryCompilationErrors.defaultValuesDataTypeError(
             statementType, colName, defaultSQL, dataType, other.dataType))
     }