HIVE-29359: Support Credential Vending in Hive Iceberg REST Catalog Client

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -5600,6 +5600,11 @@ public static enum ConfVars {
         + ",fs.s3a.access.key"
         + ",fs.s3a.secret.key"
         + ",fs.s3a.proxy.password"
+        + ",iceberg.vended.storage.credentials"
+        // Iceberg FileIO vended credential keys (S3FileIOProperties)
+        + ",s3.access-key-id"
+        + ",s3.secret-access-key"
+        + ",s3.session-token"
         + ",dfs.adls.oauth2.credential"
         + ",fs.adl.oauth2.credential"
         + ",fs.azure.account.oauth2.client.secret"

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergCatalogProperties.java

@@ -27,6 +27,7 @@
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.hive.client.RestAccessDelegationMode;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 
 public class IcebergCatalogProperties {
@@ -38,6 +39,7 @@ public class IcebergCatalogProperties {
   public static final String ICEBERG_HADOOP_TABLE_NAME = "location_based_table";
   public static final String ICEBERG_DEFAULT_CATALOG_NAME = "default_iceberg";
   public static final String NO_CATALOG_TYPE = "no catalog";
+  public static final String REST_ACCESS_DELEGATION_HEADER_PROPERTY = "header.X-Iceberg-Access-Delegation";
 
   private IcebergCatalogProperties() {
 
@@ -104,6 +106,19 @@ public static String catalogPropertyConfigKey(String catalogName, String catalog
     return String.format("%s%s.%s", CATALOG_CONFIG_PREFIX, catalogName, catalogProperty);
   }
 
+  /**
+   * Returns true when the catalog is configured to request REST vended storage credentials via
+   * {@link #REST_ACCESS_DELEGATION_HEADER_PROPERTY}.
+   */
+  public static boolean requestsVendedCredentials(String catalogName, Configuration conf) {
+    if (conf == null || StringUtils.isEmpty(catalogName)) {
+      return false;
+    }
+    String headerValue =
+        conf.get(catalogPropertyConfigKey(catalogName, REST_ACCESS_DELEGATION_HEADER_PROPERTY));
+    return RestAccessDelegationMode.headerRequestsVendedCredentials(headerValue);
+  }
+
   /**
    * Return the catalog type based on the catalog name.
    * <p>

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/client/RestAccessDelegationMode.java

@@ -0,0 +1,80 @@
+/*
+ * 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.hive.client;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.iceberg.hive.IcebergCatalogProperties;
+
+/**
+ * Values for the Iceberg REST catalog {@code X-Iceberg-Access-Delegation} request header. The header
+ * accepts a comma-separated list of these modes; configure via
+ * {@link IcebergCatalogProperties#REST_ACCESS_DELEGATION_HEADER_PROPERTY}.
+ *
+ * @see <a href="https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml">REST catalog spec</a>
+ */
+public enum RestAccessDelegationMode {
+  VENDED_CREDENTIALS("vended-credentials"),
+  REMOTE_SIGNING("remote-signing");
+
+  private final String modeName;
+
+  RestAccessDelegationMode(String modeName) {
+    this.modeName = modeName;
+  }
+
+  /** Spec-defined header token for this delegation mode. */
+  public String modeName() {
+    return modeName;
+  }
+
+  /** Comma-separated list suitable for {@link IcebergCatalogProperties#REST_ACCESS_DELEGATION_HEADER_PROPERTY}. */
+  public static String toHeaderValue(RestAccessDelegationMode... modes) {
+    return Arrays.stream(modes).map(RestAccessDelegationMode::modeName).collect(Collectors.joining(","));
+  }
+
+  /** Parses a single mode name (case-insensitive); throws if unknown. */
+  public static RestAccessDelegationMode fromModeName(String modeName) {
+    for (RestAccessDelegationMode mode : values()) {
+      if (mode.modeName.equalsIgnoreCase(modeName.trim())) {
+        return mode;
+      }
+    }
+    throw new IllegalArgumentException(
+        String.format(
+            "Unknown REST access delegation mode: %s. Valid values are: %s",
+            modeName,
+            Arrays.stream(values()).map(RestAccessDelegationMode::modeName).collect(Collectors.joining(", "))));
+  }
+
+  /** Returns true if the {@code X-Iceberg-Access-Delegation} header value includes vended credentials. */
+  public static boolean headerRequestsVendedCredentials(String headerValue) {
+    if (StringUtils.isBlank(headerValue)) {
+      return false;
+    }
+    for (String token : headerValue.split(",")) {
+      if (VENDED_CREDENTIALS.modeName.equalsIgnoreCase(token.trim())) {
+        return true;
+      }
+    }
+    return false;
+  }
+}

iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/client/TestRestAccessDelegationMode.java

@@ -0,0 +1,75 @@
+/*
+ * 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.hive.client;
+
+import org.apache.iceberg.hive.IcebergCatalogProperties;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class TestRestAccessDelegationMode {
+
+  @Test
+  void vendedCredentialsModeName() {
+    assertThat(RestAccessDelegationMode.VENDED_CREDENTIALS.modeName()).isEqualTo("vended-credentials");
+  }
+
+  @Test
+  void toHeaderValueJoinsModes() {
+    assertThat(
+            RestAccessDelegationMode.toHeaderValue(
+                RestAccessDelegationMode.VENDED_CREDENTIALS, RestAccessDelegationMode.REMOTE_SIGNING))
+        .isEqualTo("vended-credentials,remote-signing");
+  }
+
+  @Test
+  void fromModeNameParsesCaseInsensitive() {
+    assertThat(RestAccessDelegationMode.fromModeName("VENDED-CREDENTIALS"))
+        .isEqualTo(RestAccessDelegationMode.VENDED_CREDENTIALS);
+  }
+
+  @Test
+  void fromModeNameRejectsUnknown() {
+    assertThatThrownBy(() -> RestAccessDelegationMode.fromModeName("unknown"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("unknown");
+  }
+
+  @Test
+  void headerRequestsVendedCredentials() {
+    assertThat(RestAccessDelegationMode.headerRequestsVendedCredentials(null)).isFalse();
+    assertThat(RestAccessDelegationMode.headerRequestsVendedCredentials("remote-signing")).isFalse();
+    assertThat(RestAccessDelegationMode.headerRequestsVendedCredentials("vended-credentials")).isTrue();
+    assertThat(RestAccessDelegationMode.headerRequestsVendedCredentials("VENDED-CREDENTIALS,remote-signing"))
+        .isTrue();
+  }
+
+  @Test
+  void requestsVendedCredentialsFromConfiguration() {
+    org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
+    assertThat(IcebergCatalogProperties.requestsVendedCredentials("ice01", conf)).isFalse();
+
+    conf.set(
+        "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation",
+        RestAccessDelegationMode.VENDED_CREDENTIALS.modeName());
+    assertThat(IcebergCatalogProperties.requestsVendedCredentials("ice01", conf)).isTrue();
+  }
+}

iceberg/iceberg-handler/pom.xml

@@ -34,6 +34,14 @@
       <artifactId>hive-iceberg-catalog</artifactId>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <!-- compile-time only, for S3FileIOProperties/AwsClientProperties string constants
+           (inlined by javac); the classes are not referenced at runtime -->
+      <groupId>org.apache.iceberg</groupId>
+      <artifactId>iceberg-aws</artifactId>
+      <version>${iceberg.version}</version>
+      <scope>provided</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client</artifactId>

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/Catalogs.java

@@ -229,16 +229,27 @@ public static void renameTable(Configuration conf, Properties props, TableIdenti
   }
 
   static Optional<Catalog> loadCatalog(Configuration conf, String catalogName) {
-    String catalogType = IcebergCatalogProperties.getCatalogType(conf, catalogName);
+    String resolvedName = catalogName != null ? catalogName : configuredDefaultCatalogName(conf);
+    String catalogType = IcebergCatalogProperties.getCatalogType(conf, resolvedName);
     if (NO_CATALOG_TYPE.equalsIgnoreCase(catalogType)) {
       return Optional.empty();
     } else {
-      String name = catalogName == null ? ICEBERG_DEFAULT_CATALOG_NAME : catalogName;
+      String name = resolvedName == null ? ICEBERG_DEFAULT_CATALOG_NAME : resolvedName;
       return Optional.of(CatalogUtil.buildIcebergCatalog(name,
           IcebergCatalogProperties.getCatalogProperties(conf, name), conf));
     }
   }
 
+  /**
+   * Tables created without an explicit catalog in their properties belong to the session default
+   * catalog when the configuration defines one (e.g. a REST catalog); returns {@code null}
+   * otherwise, preserving the legacy default-catalog behavior.
+   */
+  private static String configuredDefaultCatalogName(Configuration conf) {
+    String defaultCatalogName = IcebergCatalogProperties.getCatalogName(conf);
+    return IcebergCatalogProperties.getCatalogType(conf, defaultCatalogName) != null ? defaultCatalogName : null;
+  }
+
   /**
    * Parse the table schema from the properties
    * @param props the controlling properties

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java

@@ -80,6 +80,18 @@ private InputFormatConfig() {
 
   public static final String CATALOG_CONFIG_PREFIX = "iceberg.catalog.";
 
+  /**
+   * Base64-serialized list of {@link org.apache.iceberg.io.StorageCredential} for Tez/LLAP executors.
+   * Stored in {@code TableDesc#jobSecrets} (HIVE-20651), not in job properties. REST catalogs mint
+   * credentials per table, so entries are keyed via {@link #vendedCredentialsKey(String)}; this
+   * constant is the key prefix (and the {@code hive.conf.hidden.list} entry, matched by prefix).
+   */
+  public static final String VENDED_STORAGE_CREDENTIALS = "iceberg.vended.storage.credentials";
+
+  public static String vendedCredentialsKey(String tableName) {
+    return VENDED_STORAGE_CREDENTIALS + "." + tableName;
+  }
+
   public static final String SORT_ORDER = "sort.order";
   public static final String SORT_COLUMNS = "sort.columns";
   public static final String ZORDER = "ZORDER";

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java

@@ -175,9 +175,12 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
                   referencedDataFiles.addAll(files.referencedDataFiles());
                   rewrittenDeleteFiles.addAll(files.rewrittenDeleteFiles());
                 }
+                // the writers' FileIO carries the table storage credentials; the table deserialized
+                // from the committer conf does not, since that conf is built at DAG time (secret-free)
+                FileIO io = writers.get(output).get(0).io();
                 createFileForCommit(
                     new FilesForCommit(dataFiles, deleteFiles, replacedDataFiles, referencedDataFiles,
-                        rewrittenDeleteFiles, mergedPaths), fileForCommitLocation, table.io());
+                        rewrittenDeleteFiles, mergedPaths), fileForCommitLocation, io);
               } else {
                 LOG.info("CommitTask found no writer for specific table: {}, attemptID: {}", output, attemptID);
                 createFileForCommit(FilesForCommit.empty(), fileForCommitLocation, table.io());
@@ -313,6 +316,9 @@ private static Multimap<OutputTable, JobContext> collectOutputs(List<JobContext>
             .orElseGet(() -> HiveTableUtil.deserializeTable(jobContext.getJobConf(), output));
         if (table != null) {
           String catalogName = catalogName(jobContext.getJobConf(), output);
+          // pass the per-table catalog name: the job-level CATALOG_NAME is absent in the DAG conf,
+          // and without it the session endpoint override is not applied to the commit-side FileIO
+          IcebergVendedCredentialUtil.applyFromJobConf(table, catalogName, jobContext.getJobConf());
           outputs.put(new OutputTable(catalogName, output, table), jobContext);
         } else {
           LOG.info("Found no table object in QueryState or conf for: {}. Skipping job commit.", output);