[Spark] Route Delta loadTable through the Delta REST API

When `deltaRestApi.enabled` is set on a UC catalog, AbstractDeltaCatalog routes table loads through UCDeltaCatalogClientImpl, backed by UCDeltaTokenBasedRestClient. Non-Delta tables fall back to the legacy delegate.

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

Author: yili-db

build.sbt

@@ -1212,6 +1212,14 @@ lazy val storage = (project in file("storage"))
         ExclusionRule(organization = "com.fasterxml.jackson.datatype"),
         ExclusionRule(organization = "com.fasterxml.jackson.dataformat")
       ),
+      // For UCDeltaTokenBasedRestClient credential vending via UCCredentialHadoopConfs.
+      "io.unitycatalog" % "unitycatalog-hadoop" % unityCatalogVersion excludeAll(
+        ExclusionRule(organization = "org.openapitools"),
+        ExclusionRule(organization = "com.fasterxml.jackson.core"),
+        ExclusionRule(organization = "com.fasterxml.jackson.module"),
+        ExclusionRule(organization = "com.fasterxml.jackson.datatype"),
+        ExclusionRule(organization = "com.fasterxml.jackson.dataformat")
+      ),
 
       // Test Deps
       "org.scalatest" %% "scalatest" % scalaTestVersion % "test",

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

@@ -63,6 +63,7 @@ import org.apache.spark.sql.execution.datasources.{DataSource, PartitioningUtils
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.sources.InsertableRelation
 import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
 
 /**
@@ -83,6 +84,15 @@ class AbstractDeltaCatalog extends DelegatingCatalogExtension
 
   val spark = SparkSession.active
 
+  /** Non-null when the catalog opted into the Delta REST API path via `deltaRestApi.enabled`. */
+  private[catalog] var deltaCatalogClient: DeltaCatalogClient = null
+
+  override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {
+    super.initialize(name, options)
+    deltaCatalogClient =
+      UCDeltaCatalogClientImpl.fromCatalogOptionsIfEnabled(name, options, super.loadTable _)
+  }
+
   private lazy val isUnityCatalog: Boolean = {
     val delegateField = classOf[DelegatingCatalogExtension].getDeclaredField("delegate")
     delegateField.setAccessible(true)
@@ -290,7 +300,11 @@ class AbstractDeltaCatalog extends DelegatingCatalogExtension
       "DeltaCatalog", "loadTable") {
     setVariantBlockingConfigIfUC()
     try {
-      val table = super.loadTable(ident)
+      val table = if (deltaCatalogClient != null) {
+        deltaCatalogClient.loadTable(ident)
+      } else {
+        super.loadTable(ident)
+      }
 
       ServerSidePlannedTable.tryCreate(spark, ident, table, isUnityCatalog).foreach { sspt =>
         return sspt

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright (2026) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.catalog
+
+import org.apache.spark.sql.connector.catalog.{Identifier, Table}
+
+/**
+ * Backend hook used by [[AbstractDeltaCatalog]] for catalog-specific table loading; isolates
+ * catalog-client and credential-vending plumbing from `AbstractDeltaCatalog`.
+ */
+private[catalog] trait DeltaCatalogClient {
+
+  /**
+   * @throws org.apache.spark.sql.catalyst.analysis.NoSuchTableException if the catalog has
+   *   no record of this identifier
+   */
+  def loadTable(ident: Identifier): Table
+}

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

@@ -0,0 +1,216 @@
+/*
+ * Copyright (2026) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.catalog
+
+import java.net.URI
+import java.util
+import java.util.concurrent.atomic.AtomicLong
+import java.util.function.Supplier
+
+import scala.jdk.CollectionConverters._
+
+import io.delta.storage.commit.uccommitcoordinator.{
+  UCDeltaClient,
+  UCDeltaModels,
+  UCDeltaTokenBasedRestClient
+}
+import io.delta.storage.commit.uccommitcoordinator.UCDeltaModels.TableInfo
+import io.delta.storage.commit.uccommitcoordinator.exceptions.{
+  CredentialFetchFailedException,
+  UnsupportedTableFormatException,
+  NoSuchTableException => StorageNoSuchTableException
+}
+import org.apache.hadoop.conf.Configuration
+
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+import org.apache.spark.sql.catalyst.catalog.{
+  CatalogStorageFormat,
+  CatalogTable,
+  CatalogTableType
+}
+import org.apache.spark.sql.connector.catalog.{Identifier, Table, V1Table}
+import org.apache.spark.sql.delta.logging.DeltaLogKeys
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.types.{DataType, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * [[DeltaCatalogClient]] backed by a [[UCDeltaClient]]; translates between Spark/Delta types
+ * and the storage-side UC types.
+ */
+private[catalog] class UCDeltaCatalogClientImpl(
+    catalogName: String,
+    ucClient: UCDeltaClient,
+    serverSidePlanningEnabled: Boolean = false,
+    fallbackLoadTable: Identifier => Table = UCDeltaCatalogClientImpl.defaultFallbackLoadTable)
+  extends DeltaCatalogClient with Logging {
+
+  override def loadTable(ident: Identifier): Table = {
+    UCDeltaCatalogClientImpl.LOAD_TABLE_INVOCATIONS.incrementAndGet()
+    val (catalog, schema, table) = parseIdent(ident)
+    val info =
+      try ucClient.loadTable(catalog, schema, table)
+      catch {
+        case _: StorageNoSuchTableException => throw new NoSuchTableException(ident)
+        case e: UnsupportedTableFormatException =>
+          logInfo(log"Table ${MDC(DeltaLogKeys.TABLE_NAME, ident)} is not in Delta format; " +
+            log"falling back to the legacy catalog path. Cause: " +
+            log"${MDC(DeltaLogKeys.EXCEPTION, e.getMessage)}")
+          return fallbackLoadTable(ident)
+        case e: CredentialFetchFailedException if serverSidePlanningEnabled =>
+          logWarning(
+            s"Credential fetch failed for $catalog.$schema.$table; enabling server-side " +
+              s"planning fallback. Cause: ${e.getMessage}")
+          enableServerSidePlanningConfig(ident)
+          e.getTableInfoWithoutCredentials
+      }
+    if (info.getStorageProperties.isEmpty && serverSidePlanningEnabled) {
+      // Unrecognized scheme (e.g. file://) returns empty creds without throwing; enable SSP
+      // so Delta can still read.
+      enableServerSidePlanningConfig(ident)
+    }
+    UCDeltaCatalogClientImpl.SUCCESSFUL_DELTA_REST_API_LOADS.incrementAndGet()
+    toV1Table(ident, info)
+  }
+
+  private def enableServerSidePlanningConfig(ident: Identifier): Unit = {
+    SparkSession.getActiveSession match {
+      case Some(spark) =>
+        spark.conf.set(DeltaSQLConf.ENABLE_SERVER_SIDE_PLANNING.key, "true")
+        logInfo(log"Server-side planning enabled for table " +
+          log"${MDC(DeltaLogKeys.TABLE_NAME, ident)}; Delta will read via SSP with empty creds.")
+      case None =>
+        logWarning(log"Server-side planning requested for table " +
+          log"${MDC(DeltaLogKeys.TABLE_NAME, ident)} but no active SparkSession found.")
+    }
+  }
+
+  // ---------- conversions ----------
+
+  private def parseIdent(ident: Identifier): (String, String, String) = {
+    val ns = ident.namespace()
+    require(
+      ns.length == 1,
+      s"UC identifiers must be of the form <schema>.<table>; got namespace ${ns.mkString(".")}")
+    (catalogName, ns(0), ident.name())
+  }
+
+  private def toV1Table(ident: Identifier, info: TableInfo): V1Table = {
+    val m = info.getMetadata
+    val properties = Option(m.getConfiguration)
+      .map(_.asScala.toMap)
+      .getOrElse(Map.empty[String, String])
+    val partitionColumns = Option(m.getPartitionColumns)
+      .map(_.asScala.toSeq)
+      .getOrElse(Seq.empty[String])
+    val schema = Option(m.getSchemaString)
+      .map(DataType.fromJson(_).asInstanceOf[StructType])
+      .getOrElse(new StructType())
+    val storage = CatalogStorageFormat.empty.copy(
+      locationUri = Some(new URI(info.getLocation)),
+      properties = properties ++ info.getStorageProperties.asScala.toMap)
+    val catalogTable = CatalogTable(
+      identifier = TableIdentifier(ident.name(), ident.namespace().headOption, Some(catalogName)),
+      tableType = fromUcTableType(info.getTableType),
+      storage = storage,
+      schema = schema,
+      provider = Option(m.getProvider).map(_.toLowerCase(java.util.Locale.ROOT)),
+      partitionColumnNames = partitionColumns,
+      comment = Option(m.getDescription),
+      createTime = if (m.getCreatedTime != null) m.getCreatedTime else 0L,
+      tracksPartitionsInCatalog = false)
+    V1Table(catalogTable)
+  }
+
+  private def fromUcTableType(t: UCDeltaModels.TableType): CatalogTableType = t match {
+    case UCDeltaModels.TableType.MANAGED => CatalogTableType.MANAGED
+    case UCDeltaModels.TableType.EXTERNAL => CatalogTableType.EXTERNAL
+  }
+}
+
+object UCDeltaCatalogClientImpl extends Logging {
+  /** Bumped at every loadTable entry, regardless of outcome. */
+  val LOAD_TABLE_INVOCATIONS: AtomicLong = new AtomicLong(0L)
+
+  /**
+   * Bumped only when loadTable returned a Delta table from the Delta REST API (no fallback,
+   * no rethrow). Use this for "Delta REST actually served the load" assertions.
+   */
+  val SUCCESSFUL_DELTA_REST_API_LOADS: AtomicLong = new AtomicLong(0L)
+
+  private[catalog] val UCDeltaRestApiEnabledKey: String = "deltaRestApi.enabled"
+  private[catalog] val RenewCredentialEnabledKey: String = "renewCredential.enabled"
+  private[catalog] val CredScopedFsEnabledKey: String = "credScopedFs.enabled"
+  private[catalog] val ServerSidePlanningEnabledKey: String = "serverSidePlanning.enabled"
+
+  private[catalog] val defaultFallbackLoadTable: Identifier => Table = ident =>
+    throw new IllegalStateException(
+      s"Non-Delta table $ident cannot be served via the Delta REST API path and no " +
+        "fallback catalog was configured.")
+
+  /**
+   * Returns {@code null} when [[UCDeltaRestApiEnabledKey]] is not set; otherwise builds
+   * a client (requires {@code uri} + {@code token}). {@code fallbackLoadTable} is invoked when
+   * UC reports {@code UnsupportedTableFormatException}.
+   */
+  private[catalog] def fromCatalogOptionsIfEnabled(
+      catalogName: String,
+      options: CaseInsensitiveStringMap,
+      fallbackLoadTable: Identifier => Table = defaultFallbackLoadTable
+  ): UCDeltaCatalogClientImpl = {
+    if (!options.getBoolean(UCDeltaRestApiEnabledKey, false)) {
+      return null
+    }
+    val client = fromCatalogOptions(catalogName, options, fallbackLoadTable)
+    logInfo(log"Delta REST API loadTable path enabled for catalog " +
+      log"${MDC(DeltaLogKeys.CATALOG, catalogName)}; Delta table loads will route through " +
+      log"UCDeltaCatalogClientImpl, non-Delta tables will fall back to the legacy delegate.")
+    client
+  }
+
+  private def fromCatalogOptions(
+      catalogName: String,
+      options: CaseInsensitiveStringMap,
+      fallbackLoadTable: Identifier => Table
+  ): UCDeltaCatalogClientImpl = {
+    val uri = Option(options.get("uri")).getOrElse(throw new IllegalArgumentException(
+      s"'uri' is required when '$UCDeltaRestApiEnabledKey' is true " +
+        s"(catalog '$catalogName')"))
+    val token = Option(options.get("token")).getOrElse(throw new IllegalArgumentException(
+      s"'token' is required when '$UCDeltaRestApiEnabledKey' is true " +
+        s"(catalog '$catalogName')"))
+    val appVersions = new util.HashMap[String, String]()
+    val renewCredEnabled = options.getBoolean(RenewCredentialEnabledKey, true)
+    val credScopedFsEnabled = options.getBoolean(CredScopedFsEnabledKey, false)
+    val sspEnabled = options.getBoolean(ServerSidePlanningEnabledKey, false)
+    val hadoopConfSupplier: Supplier[Configuration] =
+      () => SparkSession.getActiveSession
+        .map(_.sparkContext.hadoopConfiguration)
+        .getOrElse(new Configuration())
+    val restClient = UCDeltaTokenBasedRestClient.forStaticToken(
+      uri,
+      token,
+      appVersions,
+      renewCredEnabled,
+      credScopedFsEnabled,
+      hadoopConfSupplier)
+    new UCDeltaCatalogClientImpl(catalogName, restClient, sspEnabled, fallbackLoadTable)
+  }
+}