[SPARK-43742][SQL][FOLLOWUP] Do not use null literal as default value for non-nullable columns

core/src/main/resources/error/error-classes.json

@@ -1752,6 +1752,12 @@
     },
     "sqlState" : "46110"
   },
+  "NO_DEFAULT_COLUMN_VALUE_AVAILABLE" : {
+    "message" : [
+      "Can't determine the default value for <colName> since it is not nullable and it has no default value."
+    ],
+    "sqlState" : "42608"
+  },
   "NO_HANDLER_FOR_UDAF" : {
     "message" : [
       "No handler for UDAF '<functionName>'. Use sparkSession.udf.register(...) instead."

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

@@ -104,7 +104,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
         case assignment if assignment.key.semanticEquals(attr) => assignment
       }
       val resolvedValue = if (matchingAssignments.isEmpty) {
-        val defaultExpr = getDefaultValueExprOrNullLit(attr, conf)
+        val defaultExpr = getDefaultValueExprOrNullLit(
+          attr, conf.useNullsForMissingDefaultColumnValues)
         if (defaultExpr.isEmpty) {
           errors += s"No assignment for '${attr.name}'"
         }

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

@@ -67,7 +67,7 @@ object TableOutputResolver {
       val fillDefaultValue = supportColDefaultValue && actualExpectedCols.size > query.output.size
       val queryOutputCols = if (fillDefaultValue) {
         query.output ++ actualExpectedCols.drop(query.output.size).flatMap { expectedCol =>
-          getDefaultValueExprOrNullLit(expectedCol, conf)
+          getDefaultValueExprOrNullLit(expectedCol, conf.useNullsForMissingDefaultColumnValues)
         }
       } else {
         query.output
@@ -185,7 +185,7 @@ object TableOutputResolver {
       val newColPath = colPath :+ expectedCol.name
       if (matched.isEmpty) {
         val defaultExpr = if (fillDefaultValue) {
-          getDefaultValueExprOrNullLit(expectedCol, conf)
+          getDefaultValueExprOrNullLit(expectedCol, conf.useNullsForMissingDefaultColumnValues)
         } else {
           None
         }

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

@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
 import org.apache.spark.sql.connector.catalog.{CatalogManager, FunctionCatalog, Identifier, TableCatalog, TableCatalogCapability}
 import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
-import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.connector.V1Function
 import org.apache.spark.sql.types._
@@ -41,7 +41,7 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
 /**
  * This object contains fields to help process DEFAULT columns.
  */
-object ResolveDefaultColumns {
+object ResolveDefaultColumns extends QueryErrorsBase {
   // This column metadata indicates the default value associated with a particular table column that
   // is in effect at any given time. Its value begins at the time of the initial CREATE/REPLACE
   // TABLE statement with DEFAULT column definition(s), if any. It then changes whenever an ALTER
@@ -210,35 +210,54 @@ object ResolveDefaultColumns {
 
   /**
    * Generates the expression of the default value for the given field. If there is no
-   * user-specified default value for this field, returns null literal.
+   * user-specified default value for this field and the field is nullable, returns null
+   * literal, otherwise an exception is thrown.
    */
   def getDefaultValueExprOrNullLit(field: StructField): Expression = {
-    getDefaultValueExprOpt(field).getOrElse(Literal(null, field.dataType))
+    val defaultValue = getDefaultValueExprOrNullLit(field, useNullAsDefault = true)
+    if (defaultValue.isEmpty) {
+      throw new AnalysisException(
+        errorClass = "NO_DEFAULT_COLUMN_VALUE_AVAILABLE",
+        messageParameters = Map("colName" -> toSQLId(Seq(field.name))))
+    }
+    defaultValue.get
   }
 
   /**
-   * Generates the expression of the default value for the given column. If there is no
-   * user-specified default value for this field, returns null literal.
+   * Generates the expression of the default value for the given attribute. If there is no
+   * user-specified default value for this attribute and the attribute is nullable, returns null
+   * literal, otherwise an exception is thrown.
    */
   def getDefaultValueExprOrNullLit(attr: Attribute): Expression = {
     val field = StructField(attr.name, attr.dataType, attr.nullable, attr.metadata)
     getDefaultValueExprOrNullLit(field)
   }
 
   /**
-   * Generates the aliased expression of the default value for the given column. If there is no
-   * user-specified default value for this column, returns a null literal or None w.r.t. the config
-   * `USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES`.
+   * Generates the expression of the default value for the given field. If there is no
+   * user-specified default value for this field, returns null literal if `useNullAsDefault` is
+   * true and the field is nullable.
    */
-  def getDefaultValueExprOrNullLit(attr: Attribute, conf: SQLConf): Option[NamedExpression] = {
-    val field = StructField(attr.name, attr.dataType, attr.nullable, attr.metadata)
+  def getDefaultValueExprOrNullLit(
+      field: StructField, useNullAsDefault: Boolean): Option[NamedExpression] = {
     getDefaultValueExprOpt(field).orElse {
-      if (conf.useNullsForMissingDefaultColumnValues) {
-        Some(Literal(null, attr.dataType))
+      if (useNullAsDefault && field.nullable) {
+        Some(Literal(null, field.dataType))
       } else {
         None
       }
-    }.map(expr => Alias(expr, attr.name)())
+    }.map(expr => Alias(expr, field.name)())
+  }
+
+  /**
+   * Generates the expression of the default value for the given attribute. If there is no
+   * user-specified default value for this attribute, returns null literal if `useNullAsDefault` is
+   * true and the attribute is nullable.
+   */
+  def getDefaultValueExprOrNullLit(
+      attr: Attribute, useNullAsDefault: Boolean): Option[NamedExpression] = {
+    val field = StructField(attr.name, attr.dataType, attr.nullable, attr.metadata)
+    getDefaultValueExprOrNullLit(field, useNullAsDefault)
   }
 
   /**