[Storage] Extend UCDeltaClient with table-loading ops + exceptions

- loadTable / createStagingTable / createTable now take TableIdentifier; createTable takes AbstractMetadata + AbstractProtocol (mirrors commit()).
- TableInfo realigned with StagingTableInfo (tableId field + ordering).
- New typed exceptions: CredentialFetchFailedException, NoSuchTableException, UnsupportedTableFormatException.
- build.sbt: conditional unitycatalog-hadoop dep gated on UC version >= 0.5.0 via a small isAtLeastVersion helper.

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

Author: yili-db

build.sbt

@@ -791,6 +791,23 @@ val unityCatalogVersion: String = sys.props.getOrElse(
   if (useDefaultUnityCatalogReleaseVersion) defaultUnityCatalogReleaseVersion
   else unityCatalogReleaseVersion.getOrElse(pinnedUnityCatalogVersion))
 
+/**
+ * Returns true when `current` is at least `target`. Numeric segments only; suffix after
+ * the first `-` (e.g. `-SNAPSHOT-abc1234`) is stripped before comparison.
+ */
+def isAtLeastVersion(current: String, target: String): Boolean = {
+  def parts(v: String): Seq[Int] =
+    v.takeWhile(_ != '-').split('.').iterator
+      .map(p => scala.util.Try(p.toInt).getOrElse(0)).toSeq
+  val cur = parts(current)
+  val tgt = parts(target)
+  val n = math.max(cur.length, tgt.length)
+  (0 until n).iterator
+    .map(i => (cur.lift(i).getOrElse(0), tgt.lift(i).getOrElse(0)))
+    .find { case (a, b) => a != b }
+    .forall { case (a, b) => a >= b }
+}
+
 val sparkUnityCatalogJacksonVersion = "2.15.4" // We are using Spark 4.0's Jackson version 2.15.x, to override Unity Catalog 0.3.0's version 2.18.x
 
 // Publishes the pinned UC jars to ~/.ivy2/local if they're not already cached there. Hooked
@@ -1217,6 +1234,19 @@ lazy val storage = (project in file("storage"))
       "org.scalatest" %% "scalatest" % scalaTestVersion % "test",
       // Jackson datatype module needed for UC SDK tests (excluded from main compile scope)
       "com.fasterxml.jackson.datatype" % "jackson-datatype-jsr310" % "2.15.4" % "test",
+    ) ++ (
+      // unitycatalog-hadoop ships from UC 0.5.0 onward; older versions don't publish the
+      // artifact, so resolving it would fail. Used by UCDeltaTokenBasedRestClient for
+      // credential vending via UCCredentialHadoopConfs.
+      if (isAtLeastVersion(unityCatalogVersion, "0.5.0")) {
+        Seq("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")
+        ))
+      } else Nil
     ),
 
     // Publish the pinned UC jars before sbt tries to resolve them. storage is the transitive

storage/src/main/java/io/delta/storage/commit/uccommitcoordinator/UCDeltaClient.java

@@ -16,8 +16,10 @@
 
 package io.delta.storage.commit.uccommitcoordinator;
 
+import io.delta.storage.commit.TableIdentifier;
 import io.delta.storage.commit.actions.AbstractMetadata;
 import io.delta.storage.commit.uccommitcoordinator.UCDeltaModels.StagingTableInfo;
+import io.delta.storage.commit.uccommitcoordinator.UCDeltaModels.TableInfo;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
@@ -31,13 +33,12 @@ public interface UCDeltaClient extends UCClient {
   /**
    * Loads a table's metadata from Unity Catalog.
    *
-   * @param catalog the catalog name
-   * @param schema  the schema name
-   * @param table   the table name
-   * @return the table's {@link AbstractMetadata}
+   * @param tableIdentifier catalog + schema namespace and table name
+   * @return the table's {@link TableInfo}, carrying the catalog-supplied storage location and
+   *         metadata
    * @throws IOException on network or API errors
    */
-  AbstractMetadata loadTable(String catalog, String schema, String table) throws IOException;
+  TableInfo loadTable(TableIdentifier tableIdentifier) throws IOException;
 
   /**
    * Reserves a staging slot for a new Delta table. The returned response contains the table ID,

storage/src/main/java/io/delta/storage/commit/uccommitcoordinator/UCDeltaModels.java

@@ -16,16 +16,18 @@
 
 package io.delta.storage.commit.uccommitcoordinator;
 
+import io.delta.storage.commit.actions.AbstractMetadata;
 import io.delta.storage.commit.actions.AbstractProtocol;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.UUID;
 
 /**
- * Delta-owned models for the UC Delta REST Catalog API. These decouple the {@link UCDeltaClient}
+ * Delta-owned models for the UC Delta REST API. These decouple the {@link UCDeltaClient}
  * interface from any generated SDK types.
  */
 public final class UCDeltaModels {
@@ -101,9 +103,54 @@ public int hashCode() {
     }
   }
 
+  /** Result of {@link UCDeltaClient#loadTable}. */
+  public static final class TableInfo {
+
+    private final UUID tableId;
+    private final TableType tableType;
+    private final String location;
+    private final AbstractMetadata metadata;
+    private final Map<String, String> storageProperties;
+
+    public TableInfo(
+        UUID tableId,
+        TableType tableType,
+        String location,
+        AbstractMetadata metadata,
+        Map<String, String> storageProperties) {
+      this.tableId = tableId;
+      this.tableType = tableType;
+      this.location = location;
+      this.metadata = metadata;
+      this.storageProperties = storageProperties;
+    }
+
+    /** UC's {@code table_uuid}; distinct from {@link AbstractMetadata#getId()} (the Delta id). */
+    public UUID getTableId() {
+      return tableId;
+    }
+
+    public TableType getTableType() {
+      return tableType;
+    }
+
+    public String getLocation() {
+      return location;
+    }
+
+    public AbstractMetadata getMetadata() {
+      return metadata;
+    }
+
+    /** Hadoop-style storage options (e.g. catalog-vended credentials). */
+    public Map<String, String> getStorageProperties() {
+      return storageProperties == null ? Collections.emptyMap() : storageProperties;
+    }
+  }
+
   public static final class StagingTableInfo {
 
-    private final String tableId;
+    private final UUID tableId;
     private final TableType tableType;
     private final String location;
     private final DeltaProtocol requiredProtocol;
@@ -112,7 +159,7 @@ public static final class StagingTableInfo {
     private final Map<String, String> suggestedProperties;
 
     public StagingTableInfo(
-        String tableId,
+        UUID tableId,
         TableType tableType,
         String location,
         DeltaProtocol requiredProtocol,
@@ -128,7 +175,7 @@ public StagingTableInfo(
       this.suggestedProperties = suggestedProperties;
     }
 
-    public String getTableId() {
+    public UUID getTableId() {
       return tableId;
     }