[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; the session-catalog path is unaffected because
Spark's V2SessionCatalog already reports this capability.

Bumps the pinned Unity Catalog SHA used by the UC integration tests to 7ad03012
(unitycatalog#1589), where UCSingleCatalog forwards the delegate's capabilities,
so the end-to-end path is exercised.

Tests: 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, date, array, map, and
struct columns.

Author: TimothyW553

project/scripts/setup_unitycatalog_main.sh

@@ -57,7 +57,7 @@ set -euo pipefail
 # The pin. Bump both lines together if UC's version.sbt changed at the new SHA. build.sbt's
 # `unityCatalogVersion` is obtained by running this script with `--print-version`, so these two
 # values are the single source of truth.
-UC_PIN_SHA=e1f6e52acc39b925fd6a42180d400ca4e0a3895f
+UC_PIN_SHA=7ad030127959453b9010285f4b814dd78b947070
 UC_BASE_VERSION=0.5.0-SNAPSHOT
 # ---------------------------------------------------------------------------------------------
 

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,12 @@ class AbstractDeltaCatalog extends DelegatingCatalogExtension
       AbstractDeltaCatalogClient.fromCatalogOptionsIfEnabled(name, options, super.loadTable)
   }
 
+  override def capabilities(): util.Set[TableCatalogCapability] = {
+    val capabilities = new util.HashSet[TableCatalogCapability](super.capabilities())
+    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,47 @@ 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',
+           |  arr_col ARRAY<INT> DEFAULT ARRAY(1, 2, 3),
+           |  map_col MAP<STRING, INT> DEFAULT MAP('a', 1, 'b', 2),
+           |  struct_col STRUCT<x: INT, y: STRING> DEFAULT NAMED_STRUCT('x', 7, 'y', 'foo')
+           |)
+           |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"),
+          Seq(1, 2, 3), Map("a" -> 1, "b" -> 2), Row(7, "foo")))
+    }
+  }
 }
 
 trait DeltaTableCreationColumnMappingSuiteBase extends DeltaColumnMappingSelectedTestMixin {