Review comments

Author: szehon-ho

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableChange.java

@@ -753,7 +753,7 @@ public String newDefaultValue() {
     /**
      * Returns the column default value as {@link DefaultValue}. The default value literal
      * is not provided as updating column default values does not need to back-fill existing data.
-     * Empty string means dropping the column default value.
+     * Empty string and Null literal means dropping the column default value.
      */
     public DefaultValue newModelDefaultValue() { return newDefaultValue; }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -5247,7 +5247,7 @@ class AstBuilder extends DataTypeAstBuilder
         if (action.defaultExpression != null) {
           Option(action.defaultExpression()).map(visitDefaultExpression)
         } else if (action.dropDefault != null) {
-          Some(DefaultValueExpression(Literal(""), ""))
+          Some(DefaultValueExpression(Literal(null), ""))
         } else {
           None
         }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ColumnDefinition.scala

@@ -193,7 +193,12 @@ object ColumnDefinition {
               throw QueryCompilationErrors.defaultValuesUnresolvedExprError(
                 "ALTER TABLE ALTER COLUMN", c.column.name.quoted, d.originalSQL, null)
             }
-            validateDefaultValueExpr(d, "ALTER TABLE", c.column.name.quoted, d.dataType)
+            validateDefaultValueExpr(d, "ALTER TABLE ALTER COLUMN",
+              c.column.name.quoted, d.dataType)
+            if (!d.deterministic) {
+              throw QueryCompilationErrors.defaultValueNonDeterministicError(
+                "ALTER TABLE ALTER COLUMN", c.column.name.quoted, d.originalSQL)
+            }
           }
         }
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala

@@ -2773,7 +2773,7 @@ class DDLParserSuite extends AnalysisTest {
           None,
           None,
           None,
-          Some(DefaultValueExpression(Literal(""), ""))))))
+          Some(DefaultValueExpression(Literal(null), ""))))))
     // Make sure that the parser returns an exception when the feature is disabled.
     withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "false") {
       val sql = "CREATE TABLE my_tab(a INT, b STRING NOT NULL DEFAULT \"abc\") USING parquet"

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala

@@ -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, IntegerType, StringType}
+import org.apache.spark.sql.types.{BooleanType, CalendarIntervalType, IntegerType, NullType, StringType}
 import org.apache.spark.sql.util.QueryExecutionListener
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -557,7 +557,7 @@ class DataSourceV2DataFrameSuite
           case u: UpdateColumnDefaultValue => u
         }.head,
         new DefaultValue(
-          "", LiteralValue(UTF8String.fromString(""), StringType)))
+          "", LiteralValue(null, NullType)))
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala

@@ -3606,6 +3606,29 @@ class DataSourceV2SQLSuiteV1Filter
     }
   }
 
+  test("SPARK-52116: alter column with default value which is not deterministic") {
+    val foldableExpressions = Seq("1", "2 + 1")
+    withSQLConf(SQLConf.DEFAULT_COLUMN_ALLOWED_PROVIDERS.key -> v2Source) {
+      withTable("tab") {
+        spark.sql(s"CREATE TABLE tab (col1 DOUBLE DEFAULT 0) USING $v2Source")
+        val exception = analysisException(
+          // Rand function is not deterministic
+          s"ALTER TABLE tab ALTER COLUMN col1 SET DEFAULT rand()")
+        assert(exception.getSqlState == "42623")
+        assert(exception.errorClass.get == "INVALID_DEFAULT_VALUE.NON_DETERMINISTIC")
+        assert(exception.messageParameters("statement") == "ALTER TABLE ALTER COLUMN")
+        assert(exception.messageParameters("colName") == "`col1`")
+        assert(exception.messageParameters("defaultValue") == "rand()")
+      }
+      foldableExpressions.foreach(expr => {
+        withTable("tab") {
+          spark.sql(s"CREATE TABLE tab (col1 INT DEFAULT 100) USING $v2Source")
+          spark.sql(s"ALTER TABLE tab ADD COLUMN col2 DOUBLE DEFAULT $expr")
+        }
+      })
+    }
+  }
+
   test("SPARK-49099: Switch current schema with custom spark_catalog") {
     // Reset CatalogManager to clear the materialized `spark_catalog` instance, so that we can
     // configure a new implementation.