[Spark] Make deltaRestApi.enabled default to true

Signed-off-by: Yi Li <yi.li@databricks.com>

Author: yili-db

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=ae0a40caa8b2b0627e88b93cb84fe6be255575db
+UC_PIN_SHA=9196fa45948adbebb9276140d5d16c60ec7ae7a3
 UC_BASE_VERSION=0.5.0-SNAPSHOT
 # ---------------------------------------------------------------------------------------------
 

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

@@ -86,20 +86,38 @@ class AbstractDeltaCatalog extends DelegatingCatalogExtension
 
   val spark = SparkSession.active
 
+  // Captured from `initialize` so `deltaCatalogClient` can be built lazily (see its docs). Using
+  // the `initialize` name directly rather than `name()`, which delegates to the (possibly unset)
+  // delegate catalog.
+  private var catalogName: String = _
+  private var catalogOptions: CaseInsensitiveStringMap = _
+
   /**
    * When defined, table operations are routed through this client instead of through the
    * [[org.apache.spark.sql.connector.catalog.DelegatingCatalogExtension]] delegate that
    * `AbstractDeltaCatalog` normally relies on. This lets the catalog inject custom
    * interactions (e.g. talking to a REST endpoint, catalog-specific property handling,
    * storage-credential vending) rather than going through the Spark
    * [[org.apache.spark.sql.connector.catalog.TableCatalog]] API.
+   *
+   * Evaluated lazily rather than in `initialize` because the decision depends on
+   * [[isUnityCatalog]], which reads the [[DelegatingCatalogExtension]] delegate. Spark's
+   * `CatalogManager` calls `initialize` *before* `setDelegateCatalog` for the session catalog,
+   * so the delegate isn't available yet during `initialize`; it is by the time any table
+   * operation (the only reader of this client) runs.
    */
-  private[catalog] var deltaCatalogClient: Option[AbstractDeltaCatalogClient] = None
+  private[catalog] lazy val deltaCatalogClient: Option[AbstractDeltaCatalogClient] =
+    if (isUnityCatalog) {
+      AbstractDeltaCatalogClient.fromCatalogOptionsIfEnabled(
+        catalogName, catalogOptions, super.loadTable)
+    } else {
+      None
+    }
 
   override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {
     super.initialize(name, options)
-    deltaCatalogClient =
-      AbstractDeltaCatalogClient.fromCatalogOptionsIfEnabled(name, options, super.loadTable)
+    catalogName = name
+    catalogOptions = options
   }
 
   private lazy val isUnityCatalog: Boolean = {

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

@@ -136,17 +136,19 @@ private[catalog] object AbstractDeltaCatalogClient extends Logging {
     "org.apache.spark.sql.delta.catalog.UCDeltaCatalogClientImpl"
 
   /**
-   * Returns a [[AbstractDeltaCatalogClient]] wrapped in [[Some]] when the catalog opted in via
-   * `deltaRestApi.enabled`, else [[None]]. The concrete impl is loaded reflectively so
-   * [[AbstractDeltaCatalog]] doesn't compile-depend on it. If opt-in is explicit but reflective
-   * loading fails, throws [[IllegalStateException]] rather than silently degrading.
+   * Returns a [[AbstractDeltaCatalogClient]] wrapped in [[Some]] unless the catalog has
+   * opted out via `deltaRestApi.enabled=false`, in which case returns [[None]]. The flag
+   * defaults to `true` when absent, so any UC catalog goes through the UC Delta API path
+   * unless the operator explicitly disables it. The concrete impl is loaded reflectively so
+   * [[AbstractDeltaCatalog]] doesn't compile-depend on it; if loading fails, throws
+   * [[IllegalStateException]] rather than silently degrading.
    */
   def fromCatalogOptionsIfEnabled(
       catalogName: String,
       options: CaseInsensitiveStringMap,
       fallbackLoadTableFunc: Identifier => Table): Option[AbstractDeltaCatalogClient] = {
     val key = UCTokenBasedRestClientFactory.DELTA_REST_API_ENABLED_KEY
-    if (!options.getBoolean(key, false)) {
+    if (!options.getBoolean(key, true)) {
       return None
     }
     val factory = try {

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

@@ -637,7 +637,6 @@ object UCDeltaCatalogClientImpl extends AbstractDeltaCatalogClientFactory with L
     // case-insensitive so defaults don't create duplicate keys.
     val merged = new java.util.HashMap[String, String](options.asCaseSensitiveMap())
     Seq(
-      UCTokenBasedRestClientFactory.DELTA_REST_API_ENABLED_KEY -> "true",
       UCTokenBasedRestClientFactory.RENEW_CREDENTIAL_ENABLED_KEY -> "true",
       UCTokenBasedRestClientFactory.CRED_SCOPED_FS_ENABLED_KEY -> "false"
     ).foreach { case (k, v) => if (!options.containsKey(k)) merged.put(k, v) }
@@ -665,8 +664,8 @@ object UCDeltaCatalogClientImpl extends AbstractDeltaCatalogClientFactory with L
     val hasLegacyToken = options.get(LegacyTokenKey) != null
     if (!hasAuthPrefix && !hasLegacyToken) {
       throw new IllegalArgumentException(
-        s"auth configuration is required when 'deltaRestApi.enabled' is true " +
-          s"(catalog '$catalogName'). Set either '${AuthPrefix}type' (with the corresponding " +
+        s"auth configuration is required (catalog '$catalogName'). " +
+          s"Set either '${AuthPrefix}type' (with the corresponding " +
           s"$AuthPrefix* keys) or the legacy '$LegacyTokenKey' option.")
     }
   }

spark/src/main/scala/org/apache/spark/sql/delta/coordinatedcommits/UCCommitCoordinatorBuilder.scala

@@ -280,8 +280,8 @@ trait UCClientFactory {
  * Recognised ucConfig keys:
  *  - `uri` (required) -- the UC server endpoint.
  *  - `auth.*` / `token` (legacy) -- authentication parameters for [[TokenProvider]].
- *  - `deltaRestApi.enabled` -- if `"true"`, uses [[UCDeltaTokenBasedRestClient]];
- *    otherwise uses [[UCTokenBasedRestClient]].
+ *  - `deltaRestApi.enabled` -- defaults to `"true"`; uses [[UCDeltaTokenBasedRestClient]]
+ *    unless explicitly set to `"false"`, in which case [[UCTokenBasedRestClient]] is used.
  *  - `appVersions.*` -- caller-supplied version entries merged with defaults; e.g.
  *    `appVersions.Kernel -> "0.7.0"` adds a `"Kernel"` entry to the version map.
  */
@@ -309,7 +309,7 @@ object UCTokenBasedRestClientFactory extends UCClientFactory {
     val tokenProvider = TokenProvider.create(authConfig.asJava)
 
     val className =
-      if (Option(ucConfig.get(DELTA_REST_API_ENABLED_KEY)).exists(_.equalsIgnoreCase("true"))) {
+      if (ucConfig.getOrDefault(DELTA_REST_API_ENABLED_KEY, "true").toBoolean) {
         DELTA_UC_CLIENT_CLASS
       } else {
         DEFAULT_UC_CLIENT_CLASS

spark/src/test/scala/org/apache/spark/sql/delta/catalog/AbstractDeltaCatalogClientRoutingSuite.scala

@@ -52,11 +52,19 @@ class AbstractDeltaCatalogClientRoutingSuite extends QueryTest with DeltaSQLComm
 
   test("deltaRestApi.enabled=false leaves deltaCatalogClient empty") {
     val catalog = new AbstractDeltaCatalog
-    catalog.initialize("test_cat", options())
+    catalog.initialize("test_cat", options("deltaRestApi.enabled" -> "false"))
     assert(catalog.deltaCatalogClient.isEmpty,
       "UC Delta API client should not be constructed when the catalog opts out")
   }
 
+  test("deltaRestApi.enabled defaults to true: requires uri when flag absent") {
+    val catalog = new AbstractDeltaCatalog
+    val e = intercept[IllegalArgumentException] {
+      catalog.initialize("test_cat", options())
+    }
+    assert(e.getMessage.contains("'uri' is required"))
+  }
+
   test("deltaRestApi.enabled=true requires uri") {
     val catalog = new AbstractDeltaCatalog
     val e = intercept[IllegalArgumentException] {
@@ -97,7 +105,7 @@ class AbstractDeltaCatalogClientRoutingSuite extends QueryTest with DeltaSQLComm
 
   test("AbstractDeltaCatalogClient.fromCatalogOptionsIfEnabled returns None when flag is off") {
     val result = AbstractDeltaCatalogClient.fromCatalogOptionsIfEnabled(
-      "test_cat", options(), noFallback)
+      "test_cat", options("deltaRestApi.enabled" -> "false"), noFallback)
     assert(result.isEmpty)
   }
 
@@ -109,6 +117,16 @@ class AbstractDeltaCatalogClientRoutingSuite extends QueryTest with DeltaSQLComm
     assert(result.isDefined)
   }
 
+  test("AbstractDeltaCatalogClient.fromCatalogOptionsIfEnabled returns Some when flag absent " +
+      "(default is on)") {
+    val result = AbstractDeltaCatalogClient.fromCatalogOptionsIfEnabled(
+      "test_cat",
+      options("uri" -> "http://uc", "token" -> "tok"),
+      noFallback)
+    assert(result.isDefined,
+      "absent deltaRestApi.enabled should default to true and construct the client")
+  }
+
   private val noFallback: Identifier => Table =
     _ => throw new UnsupportedOperationException("fallback not expected in this test")
 

spark/unitycatalog/src/test/java/io/sparkuctest/UCDeltaTableIntegrationBaseTest.java

@@ -169,14 +169,15 @@ private SparkConf configureSparkWithUnityCatalog(SparkConf conf) {
         conf.set("spark.sql.catalog." + catalogName, "io.unitycatalog.spark.UCSingleCatalog")
             .set("spark.sql.catalog." + catalogName + ".uri", uc.serverUri())
             .set("spark.sql.catalog." + catalogName + ".token", uc.serverToken());
-    if (useDeltaRestApiForTests()) {
+    if (!useDeltaRestApiForTests()) {
+      // Default is true. Tests can opt out.
       conf =
           conf.set(
               "spark.sql.catalog."
                   + catalogName
                   + "."
                   + UCTokenBasedRestClientFactory.DELTA_REST_API_ENABLED_KEY(),
-              "true");
+              "false");
     }
     return conf;
   }