[Spark] Advertise SUPPORT_COLUMN_DEFAULT_VALUE catalog capability

DeltaCatalog now overrides capabilities() to advertise
TableCatalogCapability.SUPPORT_COLUMN_DEFAULT_VALUE on top of the delegate's
capabilities. Without this, Spark's analyzer (validateCatalogForDefaultValue)
rejects column DEFAULT clauses on tables created through a non-session Delta
catalog such as Unity Catalog.

Adds a unit test asserting the capability is added on top of a delegate that
reports none, and an end-to-end test creating a Delta table with DEFAULT values
across string, int, bigint, boolean, double, decimal, and date columns.

Author: TimothyW553

spark/src/main/scala/org/apache/spark/sql/delta/catalog/AbstractDeltaCatalog.scala

@@ -55,7 +55,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchNamespaceException, NoSuchTableException, UnresolvedAttribute, UnresolvedFieldName, UnresolvedFieldPosition}
 import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogTable, CatalogTableType, CatalogUtils, SessionCatalog}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, QualifiedColType, QualifiedColTypeShims, SyncIdentity}
-import org.apache.spark.sql.connector.catalog.{DelegatingCatalogExtension, Identifier, StagedTable, StagingTableCatalog, SupportsWrite, Table, TableCapability, TableCatalog, TableChange, V1Table}
+import org.apache.spark.sql.connector.catalog.{DelegatingCatalogExtension, Identifier, StagedTable, StagingTableCatalog, SupportsWrite, Table, TableCapability, TableCatalog, TableCatalogCapability, TableChange, V1Table}
 import org.apache.spark.sql.connector.catalog.TableCapability._
 import org.apache.spark.sql.connector.catalog.TableChange._
 import org.apache.spark.sql.connector.expressions.{FieldReference, IdentityTransform, Literal, NamedReference, Transform}
@@ -102,6 +102,14 @@ class AbstractDeltaCatalog extends DelegatingCatalogExtension
       AbstractDeltaCatalogClient.fromCatalogOptionsIfEnabled(name, options, super.loadTable)
   }
 
+  override def capabilities(): util.Set[TableCatalogCapability] = {
+    val capabilities = new util.HashSet[TableCatalogCapability](super.capabilities())
+    // Delta tables support column DEFAULT values, so advertise this catalog capability on top of
+    // whatever the delegate catalog already supports.
+    capabilities.add(TableCatalogCapability.SUPPORT_COLUMN_DEFAULT_VALUE)
+    capabilities
+  }
+
   private lazy val isUnityCatalog: Boolean = {
     val delegateField = classOf[DelegatingCatalogExtension].getDeclaredField("delegate")
     delegateField.setAccessible(true)

spark/src/test/scala/org/apache/spark/sql/delta/DeltaTableCreationTests.scala

@@ -28,8 +28,9 @@ import org.apache.spark.sql.delta.DeltaOperations.ManualUpdate
 import org.apache.spark.sql.delta.actions.Metadata
 import org.apache.spark.sql.delta.actions.Protocol
 import org.apache.spark.sql.delta.actions.TableFeatureProtocolUtils
+import org.apache.spark.sql.delta.catalog.DeltaCatalog
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
-import org.apache.spark.sql.delta.test.{DeltaColumnMappingSelectedTestMixin, DeltaSQLCommandTest}
+import org.apache.spark.sql.delta.test.{DeltaColumnMappingSelectedTestMixin, DeltaSQLCommandTest, DummyCatalog}
 import org.apache.spark.sql.delta.test.DeltaTestImplicits._
 import org.apache.commons.io.FileUtils
 import org.apache.hadoop.fs.Path
@@ -41,7 +42,7 @@ import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
 import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType, ExternalCatalogUtils, SessionCatalog}
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumnsUtils
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, Table, TableCatalog, TableCatalogCapability}
 import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
@@ -2715,6 +2716,43 @@ class DeltaTableCreationSuite
       }
     }
   }
+
+  test("Default column values: Delta catalog advertises SUPPORT_COLUMN_DEFAULT_VALUE") {
+    // Spark only accepts column DEFAULT clauses on tables created through a (non-session) V2
+    // catalog when the catalog reports SUPPORT_COLUMN_DEFAULT_VALUE. The override has to add the
+    // capability on top of whatever the delegate reports, so wrap a delegate that reports none.
+    val deltaCatalog = new DeltaCatalog()
+    deltaCatalog.setDelegateCatalog(new DummyCatalog())
+    assert(deltaCatalog.capabilities().contains(
+      TableCatalogCapability.SUPPORT_COLUMN_DEFAULT_VALUE))
+  }
+
+  test("Default column values: defaults of different column types are materialized") {
+    val tbl = "default_column_types"
+    withTable(tbl) {
+      sql(
+        s"""CREATE TABLE $tbl (
+           |  id INT,
+           |  str_col STRING DEFAULT 'hello',
+           |  int_col INT DEFAULT 42,
+           |  long_col BIGINT DEFAULT 1234567890123,
+           |  bool_col BOOLEAN DEFAULT true,
+           |  double_col DOUBLE DEFAULT 3.14,
+           |  decimal_col DECIMAL(5, 2) DEFAULT 9.99,
+           |  date_col DATE DEFAULT DATE'2024-01-01'
+           |)
+           |USING DELTA
+           |TBLPROPERTIES ('delta.feature.allowColumnDefaults' = 'supported')""".stripMargin)
+
+      // Provide only the column without a default; the rest fall back to their DEFAULT values.
+      sql(s"INSERT INTO $tbl (id) VALUES (1)")
+
+      checkAnswer(
+        sql(s"SELECT * FROM $tbl"),
+        Row(1, "hello", 42, 1234567890123L, true, 3.14,
+          new java.math.BigDecimal("9.99"), java.sql.Date.valueOf("2024-01-01")))
+    }
+  }
 }
 
 trait DeltaTableCreationColumnMappingSuiteBase extends DeltaColumnMappingSelectedTestMixin {