[SPARK-51290][SQL] Enable filling default values in DSv2 writes

sql/api/src/main/scala/org/apache/spark/sql/types/StructField.scala

@@ -214,6 +214,10 @@ case class StructField(
     }
   }
 
+  private[sql] def hasExistenceDefaultValue: Boolean = {
+    metadata.contains(EXISTS_DEFAULT_COLUMN_METADATA_KEY)
+  }
+
   private def getDDLDefault = getCurrentDefaultValue()
     .map(" DEFAULT " + _)
     .getOrElse("")

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

@@ -3534,7 +3534,8 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
         TableOutputResolver.suitableForByNameCheck(v2Write.isByName,
           expected = v2Write.table.output, queryOutput = v2Write.query.output)
         val projection = TableOutputResolver.resolveOutputColumns(
-          v2Write.table.name, v2Write.table.output, v2Write.query, v2Write.isByName, conf)
+          v2Write.table.name, v2Write.table.output, v2Write.query, v2Write.isByName, conf,
+          supportColDefaultValue = true)
         if (projection != v2Write.query) {
           val cleanedTable = v2Write.table match {
             case r: DataSourceV2Relation =>

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

@@ -80,7 +80,6 @@ object TableOutputResolver extends SQLConfHelper with Logging {
       query: LogicalPlan,
       byName: Boolean,
       conf: SQLConf,
-      // TODO: Only DS v1 writing will set it to true. We should enable in for DS v2 as well.
       supportColDefaultValue: Boolean = false): LogicalPlan = {
 
     if (expected.size < query.output.size) {

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

@@ -459,15 +459,17 @@ object ResolveDefaultColumns extends QueryErrorsBase
    *         Any type suitable for assigning into a row using the InternalRow.update method.
    */
   def getExistenceDefaultValues(schema: StructType): Array[Any] = {
-    schema.fields.map { field: StructField =>
-      val defaultValue: Option[String] = field.getExistenceDefaultValue()
-      defaultValue.map { _: String =>
-        val expr = analyzeExistenceDefaultValue(field)
-
-        // The expression should be a literal value by this point, possibly wrapped in a cast
-        // function. This is enforced by the execution of commands that assign default values.
-        expr.eval()
-      }.orNull
+    schema.fields.map(getExistenceDefaultValue)
+  }
+
+  def getExistenceDefaultValue(field: StructField): Any = {
+    if (field.hasExistenceDefaultValue) {
+      val expr = analyzeExistenceDefaultValue(field)
+      // The expression should be a literal value by this point, possibly wrapped in a cast
+      // function. This is enforced by the execution of commands that assign default values.
+      expr.eval()
+    } else {
+      null
     }
   }
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/V2WriteAnalysisSuite.scala

@@ -423,8 +423,8 @@ abstract class V2WriteAnalysisSuiteBase extends AnalysisTest {
     assertNotResolved(parsedPlan)
     assertAnalysisErrorCondition(
       parsedPlan,
-      expectedErrorCondition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA",
-      expectedMessageParameters = Map("tableName" -> "`table-name`", "colName" -> "`x`")
+      expectedErrorCondition = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS",
+      expectedMessageParameters = Map("tableName" -> "`table-name`", "extraColumns" -> "`a`, `b`")
     )
   }
 
@@ -438,8 +438,8 @@ abstract class V2WriteAnalysisSuiteBase extends AnalysisTest {
     assertNotResolved(parsedPlan)
     assertAnalysisErrorCondition(
       parsedPlan,
-      expectedErrorCondition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA",
-      expectedMessageParameters = Map("tableName" -> "`table-name`", "colName" -> "`x`")
+      expectedErrorCondition = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS",
+      expectedMessageParameters = Map("tableName" -> "`table-name`", "extraColumns" -> "`X`")
     )
   }
 
@@ -513,12 +513,14 @@ abstract class V2WriteAnalysisSuiteBase extends AnalysisTest {
 
     val parsedPlan = byName(table, query)
 
-    assertNotResolved(parsedPlan)
-    assertAnalysisErrorCondition(
-      parsedPlan,
-      expectedErrorCondition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA",
-      expectedMessageParameters = Map("tableName" -> "`table-name`", "colName" -> "`x`")
-    )
+    withSQLConf(SQLConf.USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES.key -> "false") {
+      assertNotResolved(parsedPlan)
+      assertAnalysisErrorCondition(
+        parsedPlan,
+        expectedErrorCondition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA",
+        expectedMessageParameters = Map("tableName" -> "`table-name`", "colName" -> "`x`")
+      )
+    }
   }
 
   test("byName: insert safe cast") {

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala

@@ -28,7 +28,7 @@ import com.google.common.base.Objects
 
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, JoinedRow, MetadataStructFieldWithLogicalName}
-import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, DateTimeUtils}
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, DateTimeUtils, ResolveDefaultColumns}
 import org.apache.spark.sql.connector.distributions.{Distribution, Distributions}
 import org.apache.spark.sql.connector.expressions._
 import org.apache.spark.sql.connector.metric.{CustomMetric, CustomSumMetric, CustomTaskMetric}
@@ -141,7 +141,8 @@ abstract class InMemoryBaseTable(
         schema: StructType,
         row: InternalRow): (Any, DataType) = {
       val index = schema.fieldIndex(fieldNames(0))
-      val value = row.toSeq(schema).apply(index)
+      val field = schema(index)
+      val value = row.get(index, field.dataType)
       if (fieldNames.length > 1) {
         (value, schema(index).dataType) match {
           case (row: InternalRow, nestedSchema: StructType) =>
@@ -400,18 +401,23 @@ abstract class InMemoryBaseTable(
       val sizeInBytes = numRows * rowSizeInBytes
 
       val numOfCols = tableSchema.fields.length
-      val dataTypes = tableSchema.fields.map(_.dataType)
-      val colValueSets = new Array[util.HashSet[Object]](numOfCols)
+      val colValueSets = new Array[util.HashSet[Any]](numOfCols)
       val numOfNulls = new Array[Long](numOfCols)
       for (i <- 0 until numOfCols) {
-        colValueSets(i) = new util.HashSet[Object]
+        colValueSets(i) = new util.HashSet[Any]
       }
 
       inputPartitions.foreach(inputPartition =>
         inputPartition.rows.foreach(row =>
           for (i <- 0 until numOfCols) {
-            colValueSets(i).add(row.get(i, dataTypes(i)))
-            if (row.isNullAt(i)) {
+            val field = tableSchema(i)
+            val colValue = if (i < row.numFields) {
+              row.get(i, field.dataType)
+            } else {
+              ResolveDefaultColumns.getExistenceDefaultValue(field)
+            }
+            colValueSets(i).add(colValue)
+            if (colValue == null) {
               numOfNulls(i) += 1
             }
           }
@@ -718,6 +724,11 @@ private class BufferedRowsReader(
       schema: StructType,
       row: InternalRow): Any = {
     val index = schema.fieldIndex(field.name)
+
+    if (index >= row.numFields) {
+      return ResolveDefaultColumns.getExistenceDefaultValue(field)
+    }
+
     field.dataType match {
       case StructType(fields) =>
         if (row.isNullAt(index)) {

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala

@@ -122,7 +122,7 @@ class BasicInMemoryTableCatalog extends TableCatalog {
   override def alterTable(ident: Identifier, changes: TableChange*): Table = {
     val table = loadTable(ident).asInstanceOf[InMemoryTable]
     val properties = CatalogV2Util.applyPropertiesChanges(table.properties, changes)
-    val schema = CatalogV2Util.applySchemaChanges(table.schema, changes, None, "ALTER TABLE")
+    val schema = CatalogV2Util.applySchemaChanges(table.schema, changes, Some(name), "ALTER TABLE")
     val finalPartitioning = CatalogV2Util.applyClusterByChanges(table.partitioning, schema, changes)
 
     // fail if the last column in the schema was dropped

sql/core/src/test/scala/org/apache/spark/sql/DataFrameWriterV2Suite.scala

@@ -146,8 +146,8 @@ class DataFrameWriterV2Suite extends QueryTest with SharedSparkSession with Befo
       exception = intercept[AnalysisException] {
         spark.table("source").withColumnRenamed("data", "d").writeTo("testcat.table_name").append()
       },
-      condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA",
-      parameters = Map("tableName" -> "`testcat`.`table_name`", "colName" -> "`data`")
+      condition = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS",
+      parameters = Map("tableName" -> "`testcat`.`table_name`", "extraColumns" -> "`d`")
     )
 
     checkAnswer(
@@ -251,8 +251,8 @@ class DataFrameWriterV2Suite extends QueryTest with SharedSparkSession with Befo
         spark.table("source").withColumnRenamed("data", "d")
           .writeTo("testcat.table_name").overwrite(lit(true))
       },
-      condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA",
-      parameters = Map("tableName" -> "`testcat`.`table_name`", "colName" -> "`data`")
+      condition = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS",
+      parameters = Map("tableName" -> "`testcat`.`table_name`", "extraColumns" -> "`d`")
     )
 
     checkAnswer(
@@ -356,8 +356,8 @@ class DataFrameWriterV2Suite extends QueryTest with SharedSparkSession with Befo
         spark.table("source").withColumnRenamed("data", "d")
           .writeTo("testcat.table_name").overwritePartitions()
       },
-      condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA",
-      parameters = Map("tableName" -> "`testcat`.`table_name`", "colName" -> "`data`")
+      condition = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS",
+      parameters = Map("tableName" -> "`testcat`.`table_name`", "extraColumns" -> "`d`")
     )
 
     checkAnswer(

sql/core/src/test/scala/org/apache/spark/sql/SQLInsertTestSuite.scala

@@ -218,10 +218,10 @@ trait SQLInsertTestSuite extends QueryTest with SQLTestUtils with AdaptiveSparkP
           processInsert("t1", df, overwrite = false, byName = true)
         },
         v1ErrorClass = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS",
-        v2ErrorClass = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA",
+        v2ErrorClass = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS",
         v1Parameters = Map("tableName" -> "`spark_catalog`.`default`.`t1`",
           "extraColumns" -> "`x1`"),
-        v2Parameters = Map("tableName" -> "`testcat`.`t1`", "colName" -> "`c1`")
+        v2Parameters = Map("tableName" -> "`testcat`.`t1`", "extraColumns" -> "`x1`")
       )
       val df2 = Seq((3, 2, 1, 0)).toDF(Seq("c3", "c2", "c1", "c0"): _*)
       checkV1AndV2Error(

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

@@ -37,6 +37,14 @@ trait AlterTableTests extends SharedSparkSession with QueryErrorsBase {
 
   protected val catalogAndNamespace: String
 
+  protected def catalog: String = {
+    if (catalogAndNamespace.nonEmpty) {
+      catalogAndNamespace.split('.').headOption.getOrElse("spark_catalog")
+    } else {
+      "spark_catalog"
+    }
+  }
+
   protected val v2Format: String
 
   private def fullTableName(tableName: String): String = {
@@ -328,7 +336,7 @@ trait AlterTableTests extends SharedSparkSession with QueryErrorsBase {
   }
 
   test("SPARK-39383 DEFAULT columns on V2 data sources with ALTER TABLE ADD/ALTER COLUMN") {
-    withSQLConf(SQLConf.DEFAULT_COLUMN_ALLOWED_PROVIDERS.key -> s"$v2Format, ") {
+    withSQLConf(SQLConf.DEFAULT_COLUMN_ALLOWED_PROVIDERS.key -> s"$v2Format,$catalog") {
       val t = fullTableName("table_name")
       withTable("t") {
         sql(s"create table $t (a string) using $v2Format")

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

@@ -38,6 +38,7 @@ class DataSourceV2DataFrameSuite
   before {
     spark.conf.set("spark.sql.catalog.testcat", classOf[InMemoryTableCatalog].getName)
     spark.conf.set("spark.sql.catalog.testcat2", classOf[InMemoryTableCatalog].getName)
+    spark.conf.set(SQLConf.DEFAULT_COLUMN_ALLOWED_PROVIDERS.key, "testcat")
   }
 
   after {
@@ -263,4 +264,38 @@ class DataSourceV2DataFrameSuite
       spark.listenerManager.unregister(listener)
     }
   }
+
+  test("columns with default values") {
+    val tableName = "testcat.ns1.ns2.tbl"
+    withTable(tableName) {
+      sql(s"CREATE TABLE $tableName (id INT, dep STRING) USING foo")
+
+      val df1 = Seq((1, "hr")).toDF("id", "dep")
+      df1.writeTo(tableName).append()
+
+      sql(s"ALTER TABLE $tableName ADD COLUMN txt STRING DEFAULT 'initial-text'")
+
+      val df2 = Seq((2, "hr"), (3, "software")).toDF("id", "dep")
+      df2.writeTo(tableName).append()
+
+      sql(s"ALTER TABLE $tableName ALTER COLUMN txt SET DEFAULT 'new-text'")
+
+      val df3 = Seq((4, "hr"), (5, "hr")).toDF("id", "dep")
+      df3.writeTo(tableName).append()
+
+      val df4 = Seq((6, "hr", null), (7, "hr", "explicit-text")).toDF("id", "dep", "txt")
+      df4.writeTo(tableName).append()
+
+      checkAnswer(
+        sql(s"SELECT * FROM $tableName"),
+        Seq(
+          Row(1, "hr", "initial-text"),
+          Row(2, "hr", "initial-text"),
+          Row(3, "software", "initial-text"),
+          Row(4, "hr", "new-text"),
+          Row(5, "hr", "new-text"),
+          Row(6, "hr", null),
+          Row(7, "hr", "explicit-text")))
+    }
+  }
 }