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

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

@@ -5570,6 +5570,7 @@ public static enum ConfVars {
         + ",fs.s3a.access.key"
         + ",fs.s3a.secret.key"
         + ",fs.s3a.proxy.password"
+        + ",iceberg.mr.vended.storage.credentials"
         + ",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

@@ -39,6 +39,16 @@ public class IcebergCatalogProperties {
   public static final String ICEBERG_DEFAULT_CATALOG_NAME = "default_iceberg";
   public static final String NO_CATALOG_TYPE = "no catalog";
 
+  /**
+   * Iceberg REST catalog property for the {@code X-Iceberg-Access-Delegation} HTTP header (via
+   * {@code header.*} keys recognized by {@link org.apache.iceberg.rest.RESTUtil#configHeaders}). Value is a
+   * comma-separated list of {@link RestAccessDelegationMode} tokens, for example
+   * {@link RestAccessDelegationMode#VENDED_CREDENTIALS}.
+   *
+   * <p>Hive configuration: {@code iceberg.catalog.<name>.header.X-Iceberg-Access-Delegation}
+   */
+  public static final String REST_ACCESS_DELEGATION_HEADER_PROPERTY = "header.X-Iceberg-Access-Delegation";
+
   private IcebergCatalogProperties() {
 
   }
@@ -104,6 +114,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(Configuration conf, String catalogName) {
+    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/RestAccessDelegationMode.java

@@ -0,0 +1,79 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * 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/TestRestAccessDelegationMode.java

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

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

@@ -80,6 +80,12 @@ 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.
+   */
+  public static final String VENDED_STORAGE_CREDENTIALS = "iceberg.mr.vended.storage.credentials";
+
   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

@@ -312,6 +312,7 @@ private static Multimap<OutputTable, JobContext> collectOutputs(List<JobContext>
             // fall back to getting the serialized table from the config
             .orElseGet(() -> HiveTableUtil.deserializeTable(jobContext.getJobConf(), output));
         if (table != null) {
+          IcebergVendedCredentialUtil.applyFromJobConf(jobContext.getJobConf(), table);
           String catalogName = catalogName(jobContext.getJobConf(), output);
           outputs.put(new OutputTable(catalogName, output, table), jobContext);
         } else {

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

@@ -284,6 +284,21 @@ public HiveAuthorizationProvider getAuthorizationProvider() {
     return null;
   }
 
+  @Override
+  public void configureInputJobCredentials(TableDesc tableDesc, Map<String, String> secrets) {
+    if (!IcebergVendedCredentialUtil.requestsVendedCredentials(conf, tableDesc.getProperties())) {
+      return;
+    }
+    try {
+      Table table =
+          IcebergVendedCredentialUtil.tableWithVendedCredentials(conf, tableDesc.getProperties());
+      String catalogName = tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
+      IcebergVendedCredentialUtil.propagateToJob(conf, table, catalogName, null, secrets);
+    } catch (NoSuchTableException ex) {
+      // Table may not exist yet for CTAS; credentials will not be available.
+    }
+  }
+
   @Override
   public void configureInputJobProperties(TableDesc tableDesc, Map<String, String> map) {
     overlayTableProperties(conf, tableDesc, map);
@@ -335,32 +350,10 @@ public void commitJob(JobContext originalContext) {
   @Override
   public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
     setCommonJobConf(jobConf);
-    if (tableDesc != null && tableDesc.getProperties() != null &&
-        tableDesc.getProperties().get(InputFormatConfig.OPERATION_TYPE_PREFIX + tableDesc.getTableName()) != null) {
-      String tableName = tableDesc.getTableName();
-      String opKey = InputFormatConfig.OPERATION_TYPE_PREFIX + tableName;
-      // set operation type into job conf too
-      jobConf.set(opKey, tableDesc.getProperties().getProperty(opKey));
-      Preconditions.checkArgument(!tableName.contains(TABLE_NAME_SEPARATOR),
-          "Can not handle table " + tableName + ". Its name contains '" + TABLE_NAME_SEPARATOR + "'");
-      if (HiveCustomStorageHandlerUtils.getWriteOperation(tableDesc.getProperties()::getProperty, tableName) != null) {
-        HiveCustomStorageHandlerUtils.setWriteOperation(jobConf, tableName,
-            Operation.valueOf(tableDesc.getProperties().getProperty(
-                HiveCustomStorageHandlerUtils.WRITE_OPERATION_CONFIG_PREFIX + tableName)));
-      }
-      boolean isMergeTaskEnabled = Boolean.parseBoolean(tableDesc.getProperty(
-          HiveCustomStorageHandlerUtils.MERGE_TASK_ENABLED + tableName));
-      if (isMergeTaskEnabled) {
-        HiveCustomStorageHandlerUtils.setMergeTaskEnabled(jobConf, tableName, true);
-      }
-      String tables = jobConf.get(InputFormatConfig.OUTPUT_TABLES);
-      tables = (tables == null) ? tableName : tables + TABLE_NAME_SEPARATOR + tableName;
-      jobConf.set(InputFormatConfig.OUTPUT_TABLES, tables);
-
-      String catalogName = tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
-      if (catalogName != null) {
-        jobConf.set(InputFormatConfig.TABLE_CATALOG_PREFIX + tableName, catalogName);
-      }
+    configureOutputTableJobConf(tableDesc, jobConf);
+    if (IcebergVendedCredentialUtil.requestsVendedCredentials(conf, tableDesc.getProperties())) {
+      IcebergVendedCredentialUtil.refreshVendedCredentialsIfMissing(conf, tableDesc, jobConf);
+      IcebergVendedCredentialUtil.applyJobSecretsToJobConf(tableDesc, jobConf);
     }
     try {
       if (!jobConf.getBoolean(ConfVars.HIVE_IN_TEST_IDE.varname, false)) {
@@ -373,6 +366,37 @@ public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
     }
   }
 
+  private static void configureOutputTableJobConf(TableDesc tableDesc, JobConf jobConf) {
+    if (tableDesc == null || tableDesc.getProperties() == null ||
+        tableDesc.getProperties().get(InputFormatConfig.OPERATION_TYPE_PREFIX + tableDesc.getTableName()) == null) {
+      return;
+    }
+    String tableName = tableDesc.getTableName();
+    String opKey = InputFormatConfig.OPERATION_TYPE_PREFIX + tableName;
+    // set operation type into job conf too
+    jobConf.set(opKey, tableDesc.getProperties().getProperty(opKey));
+    Preconditions.checkArgument(!tableName.contains(TABLE_NAME_SEPARATOR),
+        "Can not handle table " + tableName + ". Its name contains '" + TABLE_NAME_SEPARATOR + "'");
+    if (HiveCustomStorageHandlerUtils.getWriteOperation(tableDesc.getProperties()::getProperty, tableName) != null) {
+      HiveCustomStorageHandlerUtils.setWriteOperation(jobConf, tableName,
+          Operation.valueOf(tableDesc.getProperties().getProperty(
+              HiveCustomStorageHandlerUtils.WRITE_OPERATION_CONFIG_PREFIX + tableName)));
+    }
+    boolean isMergeTaskEnabled = Boolean.parseBoolean(tableDesc.getProperty(
+        HiveCustomStorageHandlerUtils.MERGE_TASK_ENABLED + tableName));
+    if (isMergeTaskEnabled) {
+      HiveCustomStorageHandlerUtils.setMergeTaskEnabled(jobConf, tableName, true);
+    }
+    String tables = jobConf.get(InputFormatConfig.OUTPUT_TABLES);
+    tables = (tables == null) ? tableName : tables + TABLE_NAME_SEPARATOR + tableName;
+    jobConf.set(InputFormatConfig.OUTPUT_TABLES, tables);
+
+    String catalogName = tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
+    if (catalogName != null) {
+      jobConf.set(InputFormatConfig.TABLE_CATALOG_PREFIX + tableName, catalogName);
+    }
+  }
+
   @Override
   public boolean directInsert() {
     return true;
@@ -1667,7 +1691,11 @@ static void overlayTableProperties(Configuration configuration, TableDesc tableD
     PartitionSpec spec;
     String bytes;
     try {
-      Table table = IcebergTableUtil.getTable(configuration, props);
+      boolean vendedCredentials =
+          IcebergVendedCredentialUtil.requestsVendedCredentials(configuration, props);
+      Table table = vendedCredentials ?
+          IcebergVendedCredentialUtil.tableWithVendedCredentials(configuration, props) :
+          IcebergTableUtil.getTable(configuration, props);
       location = table.location();
       // set table format-version and write-mode information from tableDesc
       bytes = HiveTableUtil.serializeTable(table, configuration, props,
@@ -1677,6 +1705,11 @@ static void overlayTableProperties(Configuration configuration, TableDesc tableD
       schema = table.schema();
       spec = table.spec();
 
+      String catalogName = props.getProperty(InputFormatConfig.CATALOG_NAME);
+      if (vendedCredentials) {
+        IcebergVendedCredentialUtil.propagateToJob(configuration, table, catalogName, map, null);
+      }
+
       // For intra-txn read-after-write: if the table has in-memory metadata with no metadata file
       // (i.e. uncommitted changes from a prior statement in the same txn), write the metadata to
       // a file so the Tez side can reconstruct the table with the updated state.

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

@@ -244,6 +244,7 @@ public static Table deserializeTable(Configuration config, String name) {
       table = readTableObjectFromFile(location, config);
     }
     checkAndSetIoConfig(config, table);
+    IcebergVendedCredentialUtil.applyFromJobConf(config, table);
 
     // For intra-txn read-after-write: if a metadata file was written for uncommitted in-txn state,
     // reconstruct a BaseTable from it so the Tez side sees changes from prior statements.