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

Author: Dmitriy Fingerman

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

@@ -32,6 +32,15 @@
 public class IcebergCatalogProperties {
   public static final String CATALOG_NAME = "iceberg.catalog";
   public static final String CATALOG_CONFIG_PREFIX = "iceberg.catalog.";
+
+  /**
+   * Optional comma-separated Iceberg REST access delegation modes for catalog HTTP requests (for example
+   * {@code vended-credentials}). Hive maps this to the {@code X-Iceberg-Access-Delegation} header when initializing the
+   * REST catalog client, unless {@code iceberg.catalog.<name>.header.X-Iceberg-Access-Delegation} is set explicitly.
+   *
+   * @see <a href="https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml">REST catalog spec</a>
+   */
+  public static final String REST_ACCESS_DELEGATION = "rest.access-delegation";
   public static final String CATALOG_WAREHOUSE_TEMPLATE = "iceberg.catalog.%s.warehouse";
   public static final String CATALOG_IMPL_TEMPLATE = "iceberg.catalog.%s.catalog-impl";
   public static final String CATALOG_DEFAULT_CONFIG_PREFIX = "iceberg.catalog-default.";

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

@@ -64,6 +64,12 @@ public class HiveRESTCatalogClient extends BaseMetaStoreClient {
   public static final String DB_OWNER = "owner";
   public static final String DB_OWNER_TYPE = "ownerType";
 
+  /**
+   * Iceberg REST catalog property prefix recognized by {@link org.apache.iceberg.rest.RESTUtil#configHeaders}; values
+   * are sent as HTTP headers on REST requests.
+   */
+  public static final String ICEBERG_ACCESS_DELEGATION_HEADER_PROPERTY = "header.X-Iceberg-Access-Delegation";
+
   private static final Logger LOG = LoggerFactory.getLogger(HiveRESTCatalogClient.class);
 
   private RESTCatalog restCatalog;
@@ -81,10 +87,44 @@ public HiveRESTCatalogClient(Configuration conf) {
   public void reconnect()  {
     close();
     String catName = MetaStoreUtils.getDefaultCatalog(conf);
-    Map<String, String> properties = IcebergCatalogProperties.getCatalogProperties(conf);
+    Map<String, String> properties =
+        applyAccessDelegationHeader(IcebergCatalogProperties.getCatalogProperties(conf));
     restCatalog = (RESTCatalog) CatalogUtil.buildIcebergCatalog(catName, properties, null);
   }
 
+  /**
+   * Maps Hive catalog property {@link IcebergCatalogProperties#REST_ACCESS_DELEGATION} to the Iceberg REST
+   * {@code X-Iceberg-Access-Delegation} header so any spec-compliant catalog may attach vended (or other delegated)
+   * storage credentials to load responses. An explicit {@link #ICEBERG_ACCESS_DELEGATION_HEADER_PROPERTY} entry
+   * always wins and is left unchanged.
+   */
+  static Map<String, String> applyAccessDelegationHeader(Map<String, String> catalogProps) {
+    if (catalogProps.containsKey(ICEBERG_ACCESS_DELEGATION_HEADER_PROPERTY)) {
+      return catalogProps;
+    }
+    String delegation = trimToNull(catalogProps.get(IcebergCatalogProperties.REST_ACCESS_DELEGATION));
+    if (delegation == null) {
+      return catalogProps;
+    }
+    ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+    catalogProps.forEach(
+        (key, value) -> {
+          if (!IcebergCatalogProperties.REST_ACCESS_DELEGATION.equals(key)) {
+            builder.put(key, value);
+          }
+        });
+    builder.put(ICEBERG_ACCESS_DELEGATION_HEADER_PROPERTY, delegation);
+    return builder.build();
+  }
+
+  private static String trimToNull(String value) {
+    if (value == null) {
+      return null;
+    }
+    String trimmed = value.trim();
+    return trimmed.isEmpty() ? null : trimmed;
+  }
+
   @Override
   public void close() {
     try {

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

@@ -44,6 +44,7 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.hive.HiveSchemaUtil;
+import org.apache.iceberg.hive.IcebergCatalogProperties;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.LocationProvider;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -136,6 +137,37 @@ public void after() {
 
   }
 
+  @Test
+  public void applyAccessDelegationHeaderMapsToIcebergRestHeader() {
+    Map<String, String> in = Maps.newHashMap();
+    in.put("uri", "http://localhost");
+    in.put(IcebergCatalogProperties.REST_ACCESS_DELEGATION, "vended-credentials");
+    Map<String, String> out = HiveRESTCatalogClient.applyAccessDelegationHeader(in);
+    assertThat(out.get(HiveRESTCatalogClient.ICEBERG_ACCESS_DELEGATION_HEADER_PROPERTY))
+        .isEqualTo("vended-credentials");
+    assertThat(out.containsKey(IcebergCatalogProperties.REST_ACCESS_DELEGATION)).isFalse();
+    assertThat(out.get("uri")).isEqualTo("http://localhost");
+  }
+
+  @Test
+  public void applyAccessDelegationHeaderExplicitHeaderWins() {
+    Map<String, String> in = Maps.newHashMap();
+    in.put(IcebergCatalogProperties.REST_ACCESS_DELEGATION, "vended-credentials");
+    in.put(HiveRESTCatalogClient.ICEBERG_ACCESS_DELEGATION_HEADER_PROPERTY, "remote-signing");
+    Map<String, String> out = HiveRESTCatalogClient.applyAccessDelegationHeader(in);
+    assertThat(out).isSameAs(in);
+    assertThat(out.get(HiveRESTCatalogClient.ICEBERG_ACCESS_DELEGATION_HEADER_PROPERTY))
+        .isEqualTo("remote-signing");
+  }
+
+  @Test
+  public void applyAccessDelegationHeaderNoOpWhenUnset() {
+    Map<String, String> in = Maps.newHashMap();
+    in.put("uri", "http://localhost");
+    Map<String, String> out = HiveRESTCatalogClient.applyAccessDelegationHeader(in);
+    assertThat(out).isSameAs(in);
+  }
+
   @Test
   public void testGetTable() throws TException {
     spyHiveRESTCatalogClient.getTable("default", "tableName");

iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q

@@ -3,6 +3,10 @@
 --! qt:replace:/(\s+neededVirtualColumns:\s)(.*)/$1#Masked#/
 -- Mask random uuid
 --! qt:replace:/(\s+'uuid'=')\S+('\s*)/$1#Masked#$2/
+-- Mask Iceberg metadata file name (sequence id + UUID) in show create table
+--! qt:replace:/('metadata_location'=')[^']+(')/$1#Masked#$2/
+-- Mask metadata_location path in describe formatted ('|' delimiter: regex contains s3:// so '/' breaks QTestReplaceHandler).
+--! qt:replace:|(metadata_location)(\s+)(s3://\S+)|$1$2#Masked#|
 -- Mask random uuid
 --! qt:replace:/(\s+uuid\s+)\S+(\s*)/$1#Masked#$2/
 -- Mask a random snapshot id
@@ -26,9 +30,10 @@ set hive.stats.autogather=false;
 set metastore.client.impl=org.apache.iceberg.hive.client.HiveRESTCatalogClient;
 set metastore.catalog.default=ice01;
 set iceberg.catalog.ice01.type=rest;
+set iceberg.catalog.ice01.rest.access-delegation=vended-credentials;
 
---! This config is set in the driver setup (see TestIcebergRESTCatalogLlapLocalCliDriver.java)   
---! conf.set('iceberg.catalog.ice01.uri', <RESTServer URI>);
+--! REST URI, OAuth, MinIO + Gravitino S3 warehouse / credential vending, and host S3A are set in
+--! TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.
 
 create database ice_rest;
 use ice_rest;
@@ -66,11 +71,6 @@ show create table ice_orc2;
 insert into ice_orc2 partition (company_id=100) 
 VALUES ('fn1','ln1', 1, 10), ('fn2','ln2', 2, 20), ('fn3','ln3', 3, 30);
 
---! In CI, Testcontainers' .withFileSystemBind() is not able to bind the same host path to the same container path,
---! so as a workaround, the .metadata.json files from container are manually synced in a daemon process,
---! since the sync can take some time, need to wait for it to happen after the insert operation.
-! sleep 20;
-
 describe formatted ice_orc2;
 select * from ice_orc2;
 

iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out

@@ -79,15 +79,15 @@ STORED BY
 WITH SERDEPROPERTIES ( 
   'serialization.format'='1')
 LOCATION
-#### A masked pattern was here ####
+  's3://iceberg-vend/warehouse/ice_rest/ice_orc2'
 TBLPROPERTIES (
   'bucketing_version'='2', 
   'current-schema'='{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"first_name","required":false,"type":"string"},{"id":2,"name":"last_name","required":false,"type":"string"},{"id":3,"name":"dept_id","required":false,"type":"long"},{"id":4,"name":"team_id","required":false,"type":"long"},{"id":5,"name":"company_id","required":false,"type":"long"}]}', 
   'default-partition-spec'='{"spec-id":0,"fields":[{"name":"company_id","transform":"identity","source-id":5,"field-id":1000}]}', 
   'format-version'='2', 
   'iceberg.catalog'='ice01', 
   'iceberg.orc.files.only'='true', 
-#### A masked pattern was here ####
+  'metadata_location'='#Masked#', 
   'name'='ice_rest.ice_orc2', 
   'parquet.compression'='zstd', 
   'serialization.format'='1', 
@@ -143,7 +143,7 @@ Table Parameters:
 	format-version      	2                   
 	iceberg.catalog     	ice01               
 	iceberg.orc.files.only	true                
-#### A masked pattern was here ####
+	metadata_location   	#Masked#
 	name                	ice_rest.ice_orc2   
 	numFiles            	1                   
 	numRows             	3                   
@@ -219,6 +219,7 @@ PREHOOK: type: SHOWDATABASES
 POSTHOOK: query: show databases
 POSTHOOK: type: SHOWDATABASES
 ice_rest
+vend_cred_chk
 PREHOOK: query: drop database ice_rest
 PREHOOK: type: DROPDATABASE
 PREHOOK: Input: database:ice_rest
@@ -231,3 +232,4 @@ PREHOOK: query: show databases
 PREHOOK: type: SHOWDATABASES
 POSTHOOK: query: show databases
 POSTHOOK: type: SHOWDATABASES
+vend_cred_chk

itests/qtest-iceberg/pom.xml

@@ -37,6 +37,12 @@
       <artifactId>metrics-core</artifactId>
       <version>${dropwizard.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.minio</groupId>
+      <artifactId>minio</artifactId>
+      <version>8.5.11</version>
+      <scope>test</scope>
+    </dependency>
     <!-- dependencies are always listed in sorted order by groupId, artifactId -->
     <!-- test intra-project -->
     <dependency>
@@ -140,6 +146,11 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>com.sun.jersey</groupId>
       <artifactId>jersey-servlet</artifactId>
@@ -156,6 +167,11 @@
       <artifactId>hadoop-archives</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-aws</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>
@@ -364,6 +380,12 @@
       <artifactId>hbase-mapreduce</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.iceberg</groupId>
+      <artifactId>iceberg-aws</artifactId>
+      <version>${iceberg.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.tez</groupId>
       <artifactId>tez-tests</artifactId>
@@ -496,6 +518,12 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>bundle</artifactId>
+      <version>${aws-java-sdk.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>testcontainers</artifactId>