feat(spark): add OIDC federation token provider to the UC connector

build.sbt

@@ -185,6 +185,11 @@ lazy val client = (project in file("target/clients/java"))
     ),
     (Compile / compile) := ((Compile / compile) dependsOn generate).value,
 
+    // The javadoc tool crashes on the OpenAPI-generated sources (ClientCodeException wrapping a
+    // StringIndexOutOfBoundsException). The jar we consume needs no javadoc, so stop building it.
+    Compile / packageDoc / publishArtifact := false,
+    Compile / doc / sources := Seq.empty,
+
     // OpenAPI generation specs
     openApiInputSpec := (file(".") / "api" / "all.yaml").toString,
     openApiGeneratorName := "java",
@@ -556,6 +561,9 @@ lazy val spark = (project in file("connectors/spark"))
       "org.antlr" % "antlr4" % "4.9.3",
       "com.google.cloud.bigdataoss" % "util-hadoop" % "3.0.2" % Provided,
       "org.apache.hadoop" % "hadoop-azure" % "3.4.0" % Provided,
+      // S3VendedCredentialsProvider implements the SDK v2 AwsCredentialsProvider. Provided, like
+      // the GCS and ABFS SPIs above: hadoop-aws ships the SDK on the runtime classpath.
+      "software.amazon.awssdk" % "auth" % "2.24.0" % Provided,
     ),
     libraryDependencies ++= Seq(
       // Test dependencies

connectors/spark/src/main/scala/io/unitycatalog/spark/S3VendedCredentialsProvider.java

@@ -0,0 +1,235 @@
+package io.unitycatalog.spark;
+
+import io.unitycatalog.client.ApiClient;
+import io.unitycatalog.client.ApiException;
+import io.unitycatalog.client.api.TemporaryCredentialsApi;
+import io.unitycatalog.client.model.AwsCredentials;
+import io.unitycatalog.client.model.GenerateTemporaryTableCredential;
+import io.unitycatalog.client.model.TableOperation;
+import io.unitycatalog.client.model.TemporaryCredentials;
+import io.unitycatalog.spark.auth.AuthConfigUtils;
+import io.unitycatalog.spark.auth.catalog.UCTokenProvider;
+import java.net.URI;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.hadoop.conf.Configuration;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+
+/**
+ * S3A credentials provider backed by Unity Catalog vended credentials, renewing them before they
+ * expire.
+ *
+ * <p>Without this, the credentials UC vends at planning time are written verbatim into the plan as
+ * {@code fs.s3a.access.key}/{@code secret.key}/{@code session.token}. They are scoped to roughly an
+ * hour, so any stage still running past that point fails on S3 rather than on anything that looks
+ * like an auth problem. Azure and GCS already went through a provider; S3 was the one cloud left
+ * holding literal keys.
+ *
+ * <p>The provider re-requests credentials from UC on its own, which is why it is installed on the
+ * executors too: it needs the catalog coordinates and the auth configuration, both carried in the
+ * Hadoop configuration, and re-authenticates through {@link UCTokenProvider}.
+ *
+ * <p>Backport of the 0.3.x {@code AwsVendedTokenProvider} and its {@code GenericCredentialProvider}
+ * base, collapsed into one class: the 0.3.x split exists to share a cache across four clouds and
+ * five credential scopes, none of which applies to a table-scoped S3-only path.
+ */
+public class S3VendedCredentialsProvider implements AwsCredentialsProvider {
+
+  /** Bootstrap credentials vended on the driver, so the first S3 call needs no round trip. */
+  protected static final String INIT_ACCESS_KEY = "fs.s3a.uc.init.access.key";
+
+  protected static final String INIT_SECRET_KEY = "fs.s3a.uc.init.secret.key";
+  protected static final String INIT_SESSION_TOKEN = "fs.s3a.uc.init.session.token";
+  protected static final String INIT_EXPIRATION_TIME = "fs.s3a.uc.init.expiration.time";
+
+  /** Coordinates the provider needs to re-request credentials for the same scope. */
+  protected static final String UC_URI = "fs.s3a.uc.uri";
+
+  protected static final String UC_TABLE_ID = "fs.s3a.uc.table.id";
+  protected static final String UC_TABLE_OPERATION = "fs.s3a.uc.table.operation";
+
+  /** Auth configuration for the UC client, flattened under this prefix. */
+  protected static final String UC_AUTH_PREFIX = "fs.s3a.uc.auth.";
+
+  /** How long before expiry a credential is renewed. */
+  protected static final String RENEWAL_LEAD_TIME_MILLIS = "fs.s3a.uc.renewal.leadTimeMillis";
+
+  protected static final long DEFAULT_RENEWAL_LEAD_TIME_MILLIS = 60_000L;
+
+  /**
+   * Keyed by credential scope, and static because {@code generateCredentialProps} sets {@code
+   * fs.s3a.impl.disable.cache=true}: Spark builds a fresh S3AFileSystem, and so a fresh provider,
+   * on every resolution. An instance-level cache would never be reused, and each new FileSystem
+   * would re-authenticate and re-vend.
+   */
+  private static final Map<CredentialScope, VendedCredentials> CACHE = new ConcurrentHashMap<>();
+
+  private final Configuration conf;
+  private final CredentialScope scope;
+  private final long renewalLeadTimeMillis;
+
+  /** Constructor signature required by Hadoop's S3A provider instantiation. */
+  public S3VendedCredentialsProvider(URI uri, Configuration conf) {
+    this.conf = conf;
+    this.scope =
+        new CredentialScope(
+            conf.get(UC_URI), conf.get(UC_TABLE_ID), conf.get(UC_TABLE_OPERATION));
+    this.renewalLeadTimeMillis =
+        conf.getLong(RENEWAL_LEAD_TIME_MILLIS, DEFAULT_RENEWAL_LEAD_TIME_MILLIS);
+
+    VendedCredentials bootstrap = readBootstrapCredentials(conf);
+    if (bootstrap != null) {
+      CACHE.putIfAbsent(scope, bootstrap);
+    }
+  }
+
+  @Override
+  public software.amazon.awssdk.auth.credentials.AwsCredentials resolveCredentials() {
+    VendedCredentials credentials =
+        CACHE.compute(
+            scope,
+            (key, cached) -> {
+              if (cached != null && !cached.readyToRenew(renewalLeadTimeMillis)) {
+                return cached;
+              }
+              return vend(key);
+            });
+
+    return AwsSessionCredentials.builder()
+        .accessKeyId(credentials.accessKeyId)
+        .secretAccessKey(credentials.secretAccessKey)
+        .sessionToken(credentials.sessionToken)
+        .build();
+  }
+
+  private VendedCredentials vend(CredentialScope key) {
+    if (key.ucUri == null || key.tableId == null) {
+      throw new IllegalStateException(
+          "Cannot renew Unity Catalog credentials: "
+              + UC_URI
+              + " and "
+              + UC_TABLE_ID
+              + " are missing from the Hadoop configuration");
+    }
+
+    URI uri = URI.create(key.ucUri);
+    ApiClient apiClient =
+        new ApiClient().setHost(uri.getHost()).setPort(uri.getPort()).setScheme(uri.getScheme());
+
+    Map<String, String> authConfigs = AuthConfigUtils.buildAuthConfigs(prefixedAuthOptions());
+    if (!authConfigs.isEmpty()) {
+      UCTokenProvider tokenProvider = UCTokenProvider.create(authConfigs);
+      apiClient =
+          apiClient.setRequestInterceptor(
+              request -> request.header("Authorization", "Bearer " + tokenProvider.accessToken()));
+    }
+
+    TemporaryCredentials vended;
+    try {
+      vended =
+          new TemporaryCredentialsApi(apiClient)
+              .generateTemporaryTableCredentials(
+                  new GenerateTemporaryTableCredential()
+                      .tableId(key.tableId)
+                      .operation(TableOperation.fromValue(key.operation)));
+    } catch (ApiException e) {
+      throw new RuntimeException(
+          "Failed to renew Unity Catalog credentials for table " + key.tableId, e);
+    }
+
+    AwsCredentials aws = vended.getAwsTempCredentials();
+    if (aws == null) {
+      throw new IllegalStateException(
+          "Unity Catalog returned no AWS credentials for table " + key.tableId);
+    }
+    return new VendedCredentials(
+        aws.getAccessKeyId(),
+        aws.getSecretAccessKey(),
+        aws.getSessionToken(),
+        vended.getExpirationTime());
+  }
+
+  /**
+   * Re-keys the auth options from {@link #UC_AUTH_PREFIX} to the bare names {@link AuthConfigUtils}
+   * expects, so the catalog and the provider share one normalization path.
+   */
+  private Map<String, String> prefixedAuthOptions() {
+    Map<String, String> options = new HashMap<>();
+    conf.getPropsWithPrefix(UC_AUTH_PREFIX).forEach(options::put);
+    return options;
+  }
+
+  private static VendedCredentials readBootstrapCredentials(Configuration conf) {
+    String accessKey = conf.get(INIT_ACCESS_KEY);
+    String secretKey = conf.get(INIT_SECRET_KEY);
+    String sessionToken = conf.get(INIT_SESSION_TOKEN);
+    if (accessKey == null || secretKey == null || sessionToken == null) {
+      return null;
+    }
+    // Absent expiry means "never renew", matching how 0.3.x treats a static credential. UC only
+    // omits it for non-expiring credentials.
+    long expiration = conf.getLong(INIT_EXPIRATION_TIME, Long.MAX_VALUE);
+    return new VendedCredentials(accessKey, secretKey, sessionToken, expiration);
+  }
+
+  /** Test-only: the cache is static, so it outlives a single test. */
+  static void clearCache() {
+    CACHE.clear();
+  }
+
+  /** The resource a vended credential grants access to; two calls sharing it can share a credential. */
+  private static final class CredentialScope {
+    private final String ucUri;
+    private final String tableId;
+    private final String operation;
+
+    CredentialScope(String ucUri, String tableId, String operation) {
+      this.ucUri = ucUri;
+      this.tableId = tableId;
+      this.operation = operation == null ? TableOperation.READ_WRITE.getValue() : operation;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+      CredentialScope that = (CredentialScope) o;
+      return Objects.equals(ucUri, that.ucUri)
+          && Objects.equals(tableId, that.tableId)
+          && Objects.equals(operation, that.operation);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(ucUri, tableId, operation);
+    }
+  }
+
+  private static final class VendedCredentials {
+    private final String accessKeyId;
+    private final String secretAccessKey;
+    private final String sessionToken;
+    private final Long expirationTime;
+
+    VendedCredentials(
+        String accessKeyId, String secretAccessKey, String sessionToken, Long expirationTime) {
+      this.accessKeyId = accessKeyId;
+      this.secretAccessKey = secretAccessKey;
+      this.sessionToken = sessionToken;
+      this.expirationTime = expirationTime;
+    }
+
+    boolean readyToRenew(long leadTimeMillis) {
+      return expirationTime != null
+          && expirationTime <= Instant.now().toEpochMilli() + leadTimeMillis;
+    }
+  }
+}

connectors/spark/src/main/scala/io/unitycatalog/spark/UCSingleCatalog.scala

@@ -3,7 +3,8 @@ package io.unitycatalog.spark
 import io.unitycatalog.client.{ApiClient, ApiException}
 import io.unitycatalog.client.api.{SchemasApi, TablesApi, TemporaryCredentialsApi}
 import io.unitycatalog.client.model.{ColumnInfo, ColumnTypeName, CreateSchema, CreateTable, DataSourceFormat, GenerateTemporaryPathCredential, GenerateTemporaryTableCredential, ListTablesResponse, PathOperation, SchemaInfo, TableOperation, TableType, TemporaryCredentials}
-import io.unitycatalog.spark.auth.catalog.UCTokenProvider
+import io.unitycatalog.spark.auth.AuthConfigUtils
+import io.unitycatalog.spark.auth.catalog.{AuthConfigs, UCTokenProvider}
 
 import java.net.URI
 import java.util
@@ -43,24 +44,13 @@ class UCSingleCatalog extends TableCatalog with SupportsNamespaces with Logging
       .setHost(url.getHost)
       .setPort(url.getPort)
       .setScheme(url.getScheme)
-    // Backport (0.3.x formalism): resolve auth from options. Supports a static `token`
-    // or the OAuth 2.0 client-credentials keys `oauth.uri`/`oauth.clientId`/`oauth.clientSecret`
-    // (machine-to-machine). The interceptor is dynamic: `accessToken()` is called on every
-    // request, so `OAuthUCTokenProvider` refreshes the token transparently for long sessions.
-    val hasAuthConfig = options.get(UCTokenProvider.TOKEN) != null ||
-      options.get(UCTokenProvider.OAUTH_URI) != null ||
-      options.get(UCTokenProvider.OAUTH_CLIENT_ID) != null ||
-      options.get(UCTokenProvider.OAUTH_CLIENT_SECRET) != null
-    if (hasAuthConfig) {
-      // `options` is a CaseInsensitiveStringMap (keys lowercased), so read each key via `get`
-      // and rebuild a map with the exact keys the factory expects (e.g. `oauth.clientId`).
-      val authOptions = new util.HashMap[String, String]
-      Seq(UCTokenProvider.TOKEN, UCTokenProvider.OAUTH_URI, UCTokenProvider.OAUTH_CLIENT_ID,
-        UCTokenProvider.OAUTH_CLIENT_SECRET).foreach { key =>
-        val value = options.get(key)
-        if (value != null) authOptions.put(key, value)
-      }
-      val tokenProvider = UCTokenProvider.create(authOptions, "")
+    // Backport (0.3.x formalism): `AuthConfigUtils` normalizes the options into a flat, `type`-keyed
+    // map, accepting both the new `auth.*` keys and the legacy un-prefixed ones this connector
+    // shipped. The interceptor is dynamic: `accessToken()` is called on every request, so the
+    // providers refresh the token transparently for long sessions.
+    val authConfigs = AuthConfigUtils.buildAuthConfigs(options.asCaseSensitiveMap())
+    if (authConfigs.containsKey(AuthConfigs.TYPE)) {
+      val tokenProvider = UCTokenProvider.create(authConfigs)
       apiClient = apiClient.setRequestInterceptor { request =>
         request.header("Authorization", "Bearer " + tokenProvider.accessToken())
       }
@@ -72,7 +62,12 @@ class UCSingleCatalog extends TableCatalog with SupportsNamespaces with Logging
     // where external credential vending is disabled but the caller already has direct storage access.
     skipCredentialVending = java.lang.Boolean.parseBoolean(
       options.getOrDefault("skipCredentialVending", "false"))
-    val proxy = new UCProxy(apiClient, temporaryCredentialsApi, skipCredentialVending)
+    val proxy = new UCProxy(
+      apiClient,
+      temporaryCredentialsApi,
+      skipCredentialVending,
+      urlStr,
+      authConfigs.asScala.toMap)
     proxy.initialize(name, options)
     if (UCSingleCatalog.LOAD_DELTA_CATALOG.get()) {
       try {
@@ -188,12 +183,47 @@ object UCSingleCatalog {
   val LOAD_DELTA_CATALOG = ThreadLocal.withInitial[Boolean](() => true)
   val DELTA_CATALOG_LOADED = ThreadLocal.withInitial[Boolean](() => false)
 
+  /**
+   * What `S3VendedCredentialsProvider` needs to re-request credentials for the same scope: where UC
+   * lives, which table, which operation, and how to authenticate. Carried through the Hadoop
+   * configuration so the executors can renew too, since the driver's provider instance does not
+   * travel with the plan.
+   */
+  case class S3CredentialRenewal(
+      ucUri: String,
+      tableId: String,
+      operation: String,
+      authConfigs: Map[String, String]) {
+
+    def hadoopProps(temporaryCredentials: TemporaryCredentials): Map[String, String] = {
+      val bootstrap = Map(
+        S3VendedCredentialsProvider.INIT_ACCESS_KEY ->
+          temporaryCredentials.getAwsTempCredentials.getAccessKeyId,
+        S3VendedCredentialsProvider.INIT_SECRET_KEY ->
+          temporaryCredentials.getAwsTempCredentials.getSecretAccessKey,
+        S3VendedCredentialsProvider.INIT_SESSION_TOKEN ->
+          temporaryCredentials.getAwsTempCredentials.getSessionToken,
+        S3VendedCredentialsProvider.UC_URI -> ucUri,
+        S3VendedCredentialsProvider.UC_TABLE_ID -> tableId,
+        S3VendedCredentialsProvider.UC_TABLE_OPERATION -> operation,
+        "fs.s3a.aws.credentials.provider" -> classOf[S3VendedCredentialsProvider].getName
+      )
+      // UC omits the expiry only for non-expiring credentials; the provider then never renews.
+      val expiry = Option(temporaryCredentials.getExpirationTime)
+        .map(e => S3VendedCredentialsProvider.INIT_EXPIRATION_TIME -> e.toString)
+      bootstrap ++ expiry ++ authConfigs.map {
+        case (k, v) => (S3VendedCredentialsProvider.UC_AUTH_PREFIX + k) -> v
+      }
+    }
+  }
+
   def generateCredentialProps(
       scheme: String,
-      temporaryCredentials: TemporaryCredentials): Map[String, String] = {
+      temporaryCredentials: TemporaryCredentials,
+      renewal: Option[S3CredentialRenewal] = None): Map[String, String] = {
     if (scheme == "s3") {
       val awsCredentials = temporaryCredentials.getAwsTempCredentials
-      Map(
+      val base = Map(
         // TODO: how to support s3:// properly?
         "fs.s3a.access.key" -> awsCredentials.getAccessKeyId,
         "fs.s3a.secret.key" -> awsCredentials.getSecretAccessKey,
@@ -202,6 +232,10 @@ object UCSingleCatalog {
         "fs.s3.impl.disable.cache" -> "true",
         "fs.s3a.impl.disable.cache" -> "true"
       )
+      // Vended credentials last about an hour, and the keys above are literals frozen into the
+      // plan. Where the table coordinates are known, hand S3A a provider that can re-vend instead,
+      // so a stage outliving them does not fail on S3. Azure and GCS already work this way.
+      renewal.map(base ++ _.hadoopProps(temporaryCredentials)).getOrElse(base)
     } else if (scheme == "gs") {
       val gcsCredentials = temporaryCredentials.getGcpOauthToken
       Map(
@@ -232,7 +266,9 @@ object UCSingleCatalog {
 private class UCProxy(
     apiClient: ApiClient,
     temporaryCredentialsApi: TemporaryCredentialsApi,
-    skipCredentialVending: Boolean) extends TableCatalog with SupportsNamespaces {
+    skipCredentialVending: Boolean,
+    ucUri: String,
+    authConfigs: Map[String, String]) extends TableCatalog with SupportsNamespaces {
   private[this] var name: String = null
   private[this] var tablesApi: TablesApi = null
   private[this] var schemasApi: SchemasApi = null
@@ -300,7 +336,11 @@ private class UCProxy(
             )
         }
       }
-      UCSingleCatalog.generateCredentialProps(uri.getScheme, temporaryCredentials)
+      UCSingleCatalog.generateCredentialProps(
+        uri.getScheme,
+        temporaryCredentials,
+        Some(UCSingleCatalog.S3CredentialRenewal(
+          ucUri, tableId, TableOperation.READ_WRITE.getValue, authConfigs)))
     }
     val sparkTable = CatalogTable(
       identifier,