apache · difin · Jul 28, 2026
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -5536,6 +5536,14 @@ public static enum ConfVars {
         + ",s3.access-key-id"
         + ",s3.secret-access-key"
         + ",s3.session-token"
+        // Iceberg FileIO vended credential keys (GCS, ADLS, OSS)
+        + ",gcs.oauth2.token"
+        + ",adls.auth.shared-key.account.key"
+        + ",adls.sas-token."
+        + ",adls.token"
+        + ",client.access-key-id"
+        + ",client.access-key-secret"
+        + ",client.security-token"
         + ",dfs.adls.oauth2.credential"
         + ",fs.adl.oauth2.credential"
         + ",fs.azure.account.oauth2.client.secret"

diff --git a/...iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergVendedCredentialUtil.java b/...iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergVendedCredentialUtil.java
@@ -38,15 +38,15 @@
 import org.apache.iceberg.StaticTableOperations;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.aws.AwsClientProperties;
-import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.hive.IcebergCatalogProperties;
 import org.apache.iceberg.hive.rest.catalog.RestCatalogAccessDelegation;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.StorageCredential;
 import org.apache.iceberg.io.SupportsStorageCredentials;
 import org.apache.iceberg.mr.InputFormatConfig;
+import org.apache.iceberg.mr.hive.vended.VendedCredentialHadoopMapper;
+import org.apache.iceberg.mr.hive.vended.VendedCredentialSupport;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.util.SerializationUtil;
@@ -102,17 +102,23 @@ public static void propagateToJob(Table table, String catalogName, Map<String, S
   /**
    * Writes each key in one vended {@link StorageCredential} into job configuration.
    *
-   * <p>Derives the S3 bucket from {@link StorageCredential#prefix()} and delegates to
+   * <p>Derives storage scope from {@link StorageCredential#prefix()} and delegates to
    * {@link #addCredentialEntry} for every entry in {@link StorageCredential#config()}, which maps
-   * Iceberg keys to catalog-level and per-bucket S3A job properties or secrets.
+   * Iceberg keys to catalog-level and provider-specific Hadoop job properties or secrets.
    */
   private static void addCredentialEntries(String catalogName, StorageCredential credential,
       Map<String, String> jobProperties, Map<String, String> jobSecrets, Configuration conf) {
 
-    String bucket = bucketFromPrefix(credential.prefix());
+    VendedCredentialHadoopMapper mapper = VendedCredentialSupport.mapperFor(credential);
+    String scope = mapper != null ? mapper.scopeFromPrefix(credential.prefix()) :
+        VendedCredentialSupport.scopeFromPrefix(credential.prefix());
     for (Map.Entry<String, String> entry : credential.config().entrySet()) {
       addCredentialEntry(
-          catalogName, bucket, entry.getKey(), entry.getValue(), jobProperties, jobSecrets, conf);
+          catalogName, scope, mapper, entry.getKey(), entry.getValue(), jobProperties, jobSecrets, conf);
+    }
+    if (jobProperties != null) {
+      VendedCredentialSupport.additionalNonSecretHadoopProperties(mapper, scope, credential.config())
+          .forEach(jobProperties::putIfAbsent);
     }
   }
 
@@ -124,7 +130,8 @@ private static void addCredentialEntries(String catalogName, StorageCredential c
    * secret keys (access key, secret key, session token) to {@code jobSecrets}. Either map may be
    * {@code null} when {@link #propagateToJob} is called for only properties or only secrets.
    */
-  private static void addCredentialEntry(String catalogName, String bucket, String icebergKey, String value,
+  private static void addCredentialEntry(String catalogName, String scope,
+      VendedCredentialHadoopMapper mapper, String icebergKey, String value,
       Map<String, String> jobProperties, Map<String, String> jobSecrets, Configuration conf) {
 
     if (StringUtils.isBlank(value)) {
@@ -133,22 +140,22 @@ private static void addCredentialEntry(String catalogName, String bucket, String
     String resolvedValue = resolveCredentialValue(catalogName, icebergKey, value, conf);
 
     if (jobProperties != null && !isSecretKey(icebergKey, conf)) {
-      addNonSecretCredentialEntry(catalogName, bucket, icebergKey, resolvedValue, jobProperties);
+      addNonSecretCredentialEntry(catalogName, scope, mapper, icebergKey, resolvedValue, jobProperties);
     }
 
     if (jobSecrets != null && isSecretKey(icebergKey, conf)) {
-      addSecretCredentialEntry(bucket, icebergKey, resolvedValue, jobSecrets);
+      addSecretCredentialEntry(scope, mapper, icebergKey, resolvedValue, jobSecrets);
     }
   }
 
   /**
-   * Adds one non-secret vended value to {@code jobProperties} for Iceberg and Hadoop S3A.
+   * Adds one non-secret vended value to {@code jobProperties} for Iceberg and Hadoop.
    *
    * <p>When {@code catalogName} is set, writes {@code iceberg.catalog.&lt;catalog&gt;.&lt;key&gt;}.
-   * When {@code bucket} is set, also writes the matching {@code fs.s3a.bucket.&lt;bucket&gt;.*}
-   * key if {@link #toS3aBucketProperty} maps the Iceberg key.
+   * When a Hadoop mapper is present, also writes the matching provider-specific Hadoop key.
    */
-  private static void addNonSecretCredentialEntry(String catalogName, String bucket, String icebergKey, String value,
+  private static void addNonSecretCredentialEntry(String catalogName, String scope,
+      VendedCredentialHadoopMapper mapper, String icebergKey, String value,
       Map<String, String> jobProperties) {
 
     if (catalogName != null) {
@@ -157,23 +164,21 @@ private static void addNonSecretCredentialEntry(String catalogName, String bucke
       jobProperties.putIfAbsent(catalogConfigKey, value);
     }
 
-    if (bucket != null) {
-      String s3aKey = toS3aBucketProperty(bucket, icebergKey);
-      if (s3aKey != null) {
-        jobProperties.putIfAbsent(s3aKey, value);
+    if (mapper != null && scope != null) {
+      String hadoopKey = VendedCredentialSupport.toHadoopProperty(mapper, scope, icebergKey);
+      if (hadoopKey != null) {
+        jobProperties.putIfAbsent(hadoopKey, value);
       }
     }
   }
 
-  /** Writes Hadoop S3A per-bucket keys only; Iceberg secrets are carried in the serialized blob.
-   * First table wins on a shared bucket, matching the non-secret entries, so an endpoint and its
-   * key always come from the same credential. */
-  private static void addSecretCredentialEntry(String bucket, String icebergKey, String value,
-      Map<String, String> jobSecrets) {
-    if (bucket != null) {
-      String s3aSecretKey = toS3aBucketProperty(bucket, icebergKey);
-      if (s3aSecretKey != null) {
-        jobSecrets.putIfAbsent(s3aSecretKey, value);
+  /** Writes Hadoop keys only; Iceberg secrets are carried in the serialized blob. */
+  private static void addSecretCredentialEntry(String scope, VendedCredentialHadoopMapper mapper,
+      String icebergKey, String value, Map<String, String> jobSecrets) {
+    if (mapper != null && scope != null) {
+      String hadoopSecretKey = VendedCredentialSupport.toHadoopProperty(mapper, scope, icebergKey);
+      if (hadoopSecretKey != null) {
+        jobSecrets.putIfAbsent(hadoopSecretKey, value);
       }
     }
   }
@@ -386,50 +391,16 @@ static List<StorageCredential> extractCredentials(Table table) {
         return credentials;
       }
     }
-    return credentialsFromFileIoProperties(table, io);
-  }
-
-  private static List<StorageCredential> credentialsFromFileIoProperties(Table table, FileIO io) {
-    Map<String, String> props = io.properties();
-    if (props == null || StringUtils.isBlank(props.get(S3FileIOProperties.ACCESS_KEY_ID)) ||
-        StringUtils.isBlank(props.get(S3FileIOProperties.SECRET_ACCESS_KEY))) {
-      return List.of();
-    }
-    Map<String, String> config = new LinkedHashMap<>();
-    putIfPresent(config, props, S3FileIOProperties.ACCESS_KEY_ID);
-    putIfPresent(config, props, S3FileIOProperties.SECRET_ACCESS_KEY);
-    putIfPresent(config, props, S3FileIOProperties.SESSION_TOKEN);
-    putIfPresent(config, props, S3FileIOProperties.ENDPOINT);
-    putIfPresent(config, props, S3FileIOProperties.PATH_STYLE_ACCESS);
-    putIfPresent(config, props, AwsClientProperties.CLIENT_REGION);
-    return List.of(StorageCredential.create(credentialPrefix(table), config));
-  }
-
-  private static void putIfPresent(Map<String, String> target, Map<String, String> source, String key) {
-    if (source.containsKey(key) && StringUtils.isNotBlank(source.get(key))) {
-      target.put(key, source.get(key));
-    }
-  }
-
-  private static String credentialPrefix(Table table) {
-    String location = table.location();
-    if (StringUtils.isBlank(location)) {
-      return "";
-    }
-    String bucket = bucketFromPrefix(location);
-    if (bucket != null) {
-      return "s3://" + bucket + "/";
-    }
-    return location.endsWith("/") ? location : location + "/";
+    return VendedCredentialSupport.credentialsFromFileIoProperties(table, io);
   }
 
   /**
-   * REST catalogs vend credentials together with S3 connectivity settings such as endpoint
+   * REST catalogs vend credentials together with storage connectivity settings such as endpoint
    * and path-style access. These settings reflect the catalog's network view and may reference
    * hosts that are not reachable from Hive (for example, an internal {@code s3.ozone:9878}
    * hostname).
    *
-   * Catalog S3 properties configured in the Hive session (for example,
+   * Catalog properties configured in the Hive session (for example,
    * {@code iceberg.catalog.ice01.s3.endpoint}) override the corresponding vended connectivity
    * settings so the driver and executors use reachable endpoints. Vended credentials are
    * preserved; only non-secret connectivity properties are overridden.
@@ -470,31 +441,4 @@ private static String resolveCredentialValue(
         conf.get(IcebergCatalogProperties.catalogPropertyConfigKey(catalogName, icebergKey));
     return StringUtils.isNotBlank(override) ? override : vendedValue;
   }
-
-  /** Authority (bucket) of a storage prefix such as {@code s3://bucket/path}, any scheme.
-   * Plain string parse: {@code URI.getHost()} rejects legal bucket names with underscores. */
-  @SuppressWarnings("java:S1075") // storage URI path separator, not a filesystem path
-  private static String bucketFromPrefix(String prefix) {
-    int schemeEnd = prefix == null ? -1 : prefix.indexOf("://");
-    if (schemeEnd < 0) {
-      return null;
-    }
-    String withoutScheme = prefix.substring(schemeEnd + 3);
-    int slash = withoutScheme.indexOf('/');
-    String bucket = slash >= 0 ? withoutScheme.substring(0, slash) : withoutScheme;
-    return StringUtils.defaultIfBlank(bucket, null);
-  }
-
-  private static String toS3aBucketProperty(String bucket, String icebergKey) {
-    String bucketPrefix = "fs.s3a.bucket." + bucket + ".";
-    return switch (icebergKey) {
-      case S3FileIOProperties.ACCESS_KEY_ID -> bucketPrefix + "access.key";
-      case S3FileIOProperties.SECRET_ACCESS_KEY -> bucketPrefix + "secret.key";
-      case S3FileIOProperties.SESSION_TOKEN -> bucketPrefix + "session.token";
-      case S3FileIOProperties.ENDPOINT -> bucketPrefix + "endpoint";
-      case S3FileIOProperties.PATH_STYLE_ACCESS -> bucketPrefix + "path.style.access";
-      case AwsClientProperties.CLIENT_REGION -> bucketPrefix + "endpoint.region";
-      default -> null;
-    };
-  }
 }
diff --git a/...ler/src/main/java/org/apache/iceberg/mr/hive/vended/AdlsVendedCredentialHadoopMapper.java b/...ler/src/main/java/org/apache/iceberg/mr/hive/vended/AdlsVendedCredentialHadoopMapper.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.iceberg.mr.hive.vended;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+
+/** Maps Iceberg ADLS FileIO properties to Hadoop ABFS connector keys. */
+enum AdlsVendedCredentialHadoopMapper implements VendedCredentialHadoopMapper {
+  INSTANCE;
+
+  static final String ADLS_SAS_TOKEN_PREFIX = "adls.sas-token.";
+  static final String ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX = "adls.sas-token-expires-at-ms.";
+  static final String ADLS_SHARED_KEY_ACCOUNT_KEY = "adls.auth.shared-key.account.key";
+
+  private static final String DFS_CORE_WINDOWS_NET_SUFFIX = ".dfs.core.windows.net";
+
+  @Override
+  public boolean supportsPrefix(String prefix) {
+    String scheme = VendedCredentialPrefixUtil.schemeFromPrefix(prefix);
+    return "abfs".equals(scheme) || "abfss".equals(scheme) || "wasb".equals(scheme) || "wasbs".equals(scheme);
+  }
+
+  @Override
+  public boolean supportsConfigKey(String icebergKey) {
+    return icebergKey.startsWith("adls.");
+  }
+
+  @Override
+  public String scopeFromPrefix(String prefix) {
+    return VendedCredentialPrefixUtil.scopeFromPrefix(prefix);
+  }
+
+  @Override
+  public String toHadoopProperty(String scope, String icebergKey) {
+    // SAS token: the Iceberg key carries the account (adls.sas-token.<account>), so the
+    // account-scoped ABFS fixed-token key is derived directly from it. ABFS also requires
+    // fs.azure.account.auth.type.<account>...=SAS, emitted by additionalNonSecretHadoopProperties.
+    if (icebergKey.startsWith(ADLS_SAS_TOKEN_PREFIX) &&
+        !icebergKey.startsWith(ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX)) {
+      String account = icebergKey.substring(ADLS_SAS_TOKEN_PREFIX.length());
+      return "fs.azure.sas.fixed.token." + account + DFS_CORE_WINDOWS_NET_SUFFIX;
+    }
+    // Shared-key: the account name is not in this entry, so it is taken from the storage prefix.
+    // SharedKey is the ABFS default auth type, so no auth-type companion property is needed.
+    if (icebergKey.equals(ADLS_SHARED_KEY_ACCOUNT_KEY)) {
+      String account = accountFromScope(scope);
+      if (account != null) {
+        return "fs.azure.account.key." + account + DFS_CORE_WINDOWS_NET_SUFFIX;
+      }
+      return null;
+    }
+    // adls.token (OAuth bearer) has no plain ABFS config key — ABFS OAuth needs a provider type
+    // (client id/secret/endpoint or refresh token). It rides the Iceberg StorageCredential blob.
+    return null;
+  }
+
+  @Override
+  public Map<String, String> additionalNonSecretHadoopProperties(String scope, Map<String, String> config) {
+    Map<String, String> extra = new LinkedHashMap<>();
+    for (String key : config.keySet()) {
+      if (key.startsWith(ADLS_SAS_TOKEN_PREFIX) && !key.startsWith(ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX)) {
+        String account = key.substring(ADLS_SAS_TOKEN_PREFIX.length());
+        extra.put("fs.azure.account.auth.type." + account + DFS_CORE_WINDOWS_NET_SUFFIX, "SAS");
+      }
+    }
+    return extra;
+  }
+
+  /**
+   * Storage account from a credential prefix scope such as
+   * {@code container@account.dfs.core.windows.net} (or {@code account.dfs.core.windows.net}),
+   * with the {@code .dfs.core.windows.net} suffix stripped.
+   */
+  private static String accountFromScope(String scope) {
+    if (StringUtils.isBlank(scope)) {
+      return null;
+    }
+    int at = scope.indexOf('@');
+    String host = at < 0 ? scope : scope.substring(at + 1);
+    if (host.endsWith(DFS_CORE_WINDOWS_NET_SUFFIX)) {
+      host = host.substring(0, host.length() - DFS_CORE_WINDOWS_NET_SUFFIX.length());
+    }
+    return StringUtils.defaultIfBlank(host, null);
+  }
+}
diff --git a/...dler/src/main/java/org/apache/iceberg/mr/hive/vended/GcsVendedCredentialHadoopMapper.java b/...dler/src/main/java/org/apache/iceberg/mr/hive/vended/GcsVendedCredentialHadoopMapper.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.iceberg.mr.hive.vended;
+
+/** Maps Iceberg GCS FileIO properties to Hadoop Google Cloud Storage connector keys. */
+enum GcsVendedCredentialHadoopMapper implements VendedCredentialHadoopMapper {
+  INSTANCE;
+
+  static final String GCS_OAUTH2_TOKEN = "gcs.oauth2.token";
+  static final String GCS_OAUTH2_TOKEN_EXPIRES_AT = "gcs.oauth2.token-expires-at";
+  static final String GCS_PROJECT_ID = "gcs.project-id";
+  static final String GCS_SERVICE_HOST = "gcs.service.host";
+
+  @Override
+  public boolean supportsPrefix(String prefix) {
+    String scheme = VendedCredentialPrefixUtil.schemeFromPrefix(prefix);
+    return "gs".equals(scheme) || "gcs".equals(scheme);
+  }
+
+  @Override
+  public boolean supportsConfigKey(String icebergKey) {
+    return icebergKey.startsWith("gcs.");
+  }
+
+  @Override
+  public String scopeFromPrefix(String prefix) {
+    return VendedCredentialPrefixUtil.scopeFromPrefix(prefix);
+  }
+
+  @Override
+  public String toHadoopProperty(String bucket, String icebergKey) {
+    // The Google Cloud Storage Hadoop connector has no plain config key to inject a raw OAuth
+    // token; a vended token requires fs.gs.auth.type=ACCESS_TOKEN_PROVIDER plus an
+    // AccessTokenProvider implementation class. Until that is wired, gcs.oauth2.token and its
+    // expiry travel only in the serialized Iceberg StorageCredential blob and are consumed by
+    // Iceberg GCSFileIO. Only connectivity/config that maps to real connector keys is emitted.
+    return switch (icebergKey) {
+      case GCS_PROJECT_ID -> "fs.gs.project.id";
+      case GCS_SERVICE_HOST -> "fs.gs.storage.root.url";
+      default -> null;
+    };
+  }
+}