[Spark] Support UnityCatalog OAuth authentication. (#5683)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

The related issue is #5509,
since we've already publish the unitycatalog-0.3.1 to the central maven
repo:
https://central.sonatype.com/artifact/io.unitycatalog/unitycatalog-client

And we've already introduced the abstracted
[TokenProvider](https://github.com/unitycatalog/unitycatalog/blob/main/clients/java/src/main/java/io/unitycatalog/client/auth/TokenProvider.java)
API in unitycatalog-client 0.3.1, then in this PR, we just use the
TokenProvider from unitycatalog-client 0.3.1 directly, and remove all
the hard-coded `static` token authentication in the oss-delta, so that
delta will have the consistent and abstracted authentication to
unitycatalog client, and with this approach, it also helps us to extend
to other custom TokenProvider implementation.


## How was this patch tested?

We use the locatly designed unit test.

## Does this PR introduce _any_ user-facing changes?

No.

---------

Signed-off-by: openinx <[email protected]>

Author: openinx

build.sbt

@@ -1023,6 +1023,13 @@ lazy val storage = (project in file("storage"))
       // Note that the org.apache.hadoop.fs.s3a.Listing::createFileStatusListingIterator 3.3.1 API
       // is not compatible with 3.3.2.
       "org.apache.hadoop" % "hadoop-aws" % hadoopVersion % "provided",
+      "io.unitycatalog" % "unitycatalog-client" % 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",

kernel-spark/src/main/java/io/delta/kernel/spark/snapshot/unitycatalog/UCTableInfo.java

@@ -17,23 +17,28 @@
 
 import static java.util.Objects.requireNonNull;
 
+import java.util.Collections;
+import java.util.Map;
+
 /**
  * Table information for Unity Catalog managed tables.
  *
  * <p>This POJO encapsulates all the information needed to interact with a Unity Catalog table
  * without requiring Spark dependencies.
  */
 public final class UCTableInfo {
+
   private final String tableId;
   private final String tablePath;
   private final String ucUri;
-  private final String ucToken;
+  private final Map<String, String> authConfig;
 
-  public UCTableInfo(String tableId, String tablePath, String ucUri, String ucToken) {
+  public UCTableInfo(
+      String tableId, String tablePath, String ucUri, Map<String, String> authConfig) {
     this.tableId = requireNonNull(tableId, "tableId is null");
     this.tablePath = requireNonNull(tablePath, "tablePath is null");
     this.ucUri = requireNonNull(ucUri, "ucUri is null");
-    this.ucToken = requireNonNull(ucToken, "ucToken is null");
+    this.authConfig = Collections.unmodifiableMap(requireNonNull(authConfig, "authConfig is null"));
   }
 
   public String getTableId() {
@@ -48,7 +53,7 @@ public String getUcUri() {
     return ucUri;
   }
 
-  public String getUcToken() {
-    return ucToken;
+  public Map<String, String> getAuthConfig() {
+    return authConfig;
   }
 }

kernel-spark/src/main/java/io/delta/kernel/spark/snapshot/unitycatalog/UCUtils.java

@@ -16,6 +16,7 @@
 package io.delta.kernel.spark.snapshot.unitycatalog;
 
 import static java.util.Objects.requireNonNull;
+import static scala.jdk.javaapi.CollectionConverters.asJava;
 
 import io.delta.kernel.spark.utils.CatalogTableUtils;
 import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorClient;
@@ -85,9 +86,8 @@ public static Optional<UCTableInfo> extractTableInfo(
 
     UCCatalogConfig config = configOpt.get();
     String ucUri = config.uri();
-    String ucToken = config.token();
 
-    return Optional.of(new UCTableInfo(tableId, tablePath, ucUri, ucToken));
+    return Optional.of(new UCTableInfo(tableId, tablePath, ucUri, asJava(config.authConfig())));
   }
 
   private static String extractUCTableId(CatalogTable catalogTable) {

kernel-spark/src/test/java/io/delta/kernel/spark/snapshot/unitycatalog/UCTableInfoTest.java

@@ -17,24 +17,33 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
+import java.util.HashMap;
+import java.util.Map;
 import org.junit.jupiter.api.Test;
 
 /** Tests for {@link UCTableInfo}. */
 class UCTableInfoTest {
 
   @Test
-  void testConstructor_ValidInputs_StoresAllFields() {
+  void testConstructor() {
     // Use distinctive values that would fail if implementation had hardcoded defaults
     String tableId = "uc_tbl_7f3a9b2c-e8d1-4f6a";
     String tablePath = "abfss://[email protected]/delta/v2";
     String ucUri = "https://uc-server.example.net/api/2.1/uc";
     String ucToken = "dapi_Kx9mN$2pQr#7vWz";
 
-    UCTableInfo info = new UCTableInfo(tableId, tablePath, ucUri, ucToken);
+    Map<String, String> authConfig = new HashMap<>();
+    authConfig.put("type", "static");
+    authConfig.put("token", ucToken);
+
+    UCTableInfo info = new UCTableInfo(tableId, tablePath, ucUri, authConfig);
 
     assertEquals(tableId, info.getTableId(), "Table ID should be stored correctly");
     assertEquals(tablePath, info.getTablePath(), "Table path should be stored correctly");
     assertEquals(ucUri, info.getUcUri(), "UC URI should be stored correctly");
-    assertEquals(ucToken, info.getUcToken(), "UC token should be stored correctly");
+
+    Map<String, String> ret = info.getAuthConfig();
+    assertEquals("static", ret.get("type"), "Type should be static");
+    assertEquals(ucToken, ret.get("token"), "UC token should be stored correctly in configMap");
   }
 }

kernel-spark/src/test/scala/io/delta/kernel/spark/snapshot/unitycatalog/UCUtilsSuite.scala

@@ -164,7 +164,13 @@ class UCUtilsSuite extends SparkFunSuite with SharedSparkSession {
         info.getTablePath == TABLE_PATH_ALPHA,
         s"Table path mismatch: got ${info.getTablePath}")
       assert(info.getUcUri == UC_URI_ALPHA, s"UC URI mismatch: got ${info.getUcUri}")
-      assert(info.getUcToken == UC_TOKEN_ALPHA, s"UC token mismatch: got ${info.getUcToken}")
+      val configMap = info.getAuthConfig
+      assert(
+        configMap.get("type") == "static",
+        s"Type should be static: got ${configMap.get("type")}")
+      assert(
+        configMap.get("token") == UC_TOKEN_ALPHA,
+        s"UC token mismatch: got ${configMap.get("token")}")
     }
   }
 
@@ -190,7 +196,7 @@ class UCUtilsSuite extends SparkFunSuite with SharedSparkSession {
       // catalogGamma config (should NOT be used)
       s"spark.sql.catalog.$catalogGamma" -> UC_CATALOG_CONNECTOR,
       s"spark.sql.catalog.$catalogGamma.uri" -> ucUriGamma,
-      s"spark.sql.catalog.$catalogGamma.token" -> ucTokenGamma,
+      s"spark.sql.catalog.$catalogGamma.token" -> ucTokenBeta,
       // catalogBeta config (should be used)
       s"spark.sql.catalog.$catalogBeta" -> UC_CATALOG_CONNECTOR,
       s"spark.sql.catalog.$catalogBeta.uri" -> ucUriBeta,
@@ -208,7 +214,11 @@ class UCUtilsSuite extends SparkFunSuite with SharedSparkSession {
       assert(
         info.getUcUri == ucUriBeta,
         s"Should use catalogBeta's URI, got: ${info.getUcUri}")
-      assert(info.getUcToken == ucTokenBeta, s"Should use catalogBeta's token, got: ${info.getUcToken}")
+      val configMap = info.getAuthConfig
+      assert(configMap.get("type") == "static", s"Type should be static")
+      assert(
+        configMap.get("token") == ucTokenBeta,
+        s"Should use catalogBeta's token, got: ${configMap.get("token")}")
       assert(info.getTableId == tableIdBeta, s"Should extract tableIdBeta, got: ${info.getTableId}")
       assert(
         info.getTablePath == tablePathBeta,