[Spark] Make deltaRestApi.enabled default to true

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

Author: yili-db

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

@@ -627,7 +627,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) }
@@ -655,8 +654,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;
   }