diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java index b10509b85f13..65ad8a82bbef 100644 --- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java +++ b/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" diff --git a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergCatalogProperties.java b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergCatalogProperties.java index 424f8e10c350..6d7f047c05d4 100644 --- a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergCatalogProperties.java +++ b/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. *

diff --git a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/client/RestAccessDelegationMode.java b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/client/RestAccessDelegationMode.java new file mode 100644 index 000000000000..89f685131ce8 --- /dev/null +++ b/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 REST catalog spec + */ +public enum RestAccessDelegationMode { + VENDED_CREDENTIALS("vended-credentials"), + REMOTE_SIGNING("remote-signing"); // Part of REST Catalog API, not implement yet in Hive + + 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; + } +} diff --git a/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/client/TestRestAccessDelegationMode.java b/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/client/TestRestAccessDelegationMode.java new file mode 100644 index 000000000000..dce5106b5cfc --- /dev/null +++ b/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(); + } +} diff --git a/iceberg/iceberg-handler/pom.xml b/iceberg/iceberg-handler/pom.xml index ef20ab1aab4d..f00415ecbe27 100644 --- a/iceberg/iceberg-handler/pom.xml +++ b/iceberg/iceberg-handler/pom.xml @@ -34,6 +34,14 @@ hive-iceberg-catalog true + + + org.apache.iceberg + iceberg-aws + ${iceberg.version} + provided + org.apache.hadoop hadoop-client diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/Catalogs.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/Catalogs.java index 9a416f8277ab..7b6f0810461f 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/Catalogs.java +++ b/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 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 diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java index 34ba9f2404ca..46c281ce0f04 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/InputFormatConfig.java +++ b/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"; diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java index 9fabe10488b1..2a8701379086 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java +++ b/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 collectOutputs(List .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); diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java index 0fa610e6f3da..2875890efdc4 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java +++ b/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 secrets) { + if (!IcebergVendedCredentialUtil.requestsVendedCredentials(tableDesc.getProperties(), conf)) { + return; + } + try { + Table table = + IcebergVendedCredentialUtil.getTableWithVendedCredentials(tableDesc.getProperties(), conf); + String catalogName = tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME); + IcebergVendedCredentialUtil.propagateToJob(table, catalogName, null, secrets, conf); + } catch (NoSuchTableException ex) { + // Table may not exist yet for CTAS; credentials will not be available. + } + } + @Override public void configureInputJobProperties(TableDesc tableDesc, Map map) { overlayTableProperties(conf, tableDesc, map); @@ -335,32 +350,9 @@ 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(tableDesc.getProperties(), conf)) { + IcebergVendedCredentialUtil.refreshVendedCredentialsIfMissing(tableDesc, jobConf, conf); } try { if (!jobConf.getBoolean(ConfVars.HIVE_IN_TEST_IDE.varname, false)) { @@ -373,6 +365,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; @@ -1681,13 +1704,12 @@ static void overlayTableProperties(Configuration configuration, TableDesc tableD PartitionSpec spec; String bytes; try { - Table table = IcebergTableUtil.getTable(configuration, props); + boolean isVendedCredentials = + IcebergVendedCredentialUtil.requestsVendedCredentials(props, configuration); + Table table = isVendedCredentials ? + IcebergVendedCredentialUtil.getTableWithVendedCredentials(props, configuration) : + IcebergTableUtil.getTable(configuration, props); location = table.location(); - // set table format-version and write-mode information from tableDesc - bytes = HiveTableUtil.serializeTable(table, configuration, props, - ImmutableList.of( - TableProperties.FORMAT_VERSION, - TableProperties.DELETE_MODE, TableProperties.UPDATE_MODE, TableProperties.MERGE_MODE)); schema = table.schema(); spec = table.spec(); @@ -1704,6 +1726,20 @@ static void overlayTableProperties(Configuration configuration, TableDesc tableD } } + if (isVendedCredentials) { + String catalogName = props.getProperty(InputFormatConfig.CATALOG_NAME); + IcebergVendedCredentialUtil.propagateToJob(table, catalogName, map, null, configuration); + // Serializing the table as-is would embed its FileIO, and with it the vended credentials, + // into plain job properties: ship a copy over a secret-free FileIO instead; executors + // restore the credentials from the Credentials channel (setCredentials). + table = IcebergVendedCredentialUtil.secretFreeCopy(table, configuration); + } + // set table format-version and write-mode information from tableDesc + bytes = HiveTableUtil.serializeTable(table, configuration, props, + ImmutableList.of( + TableProperties.FORMAT_VERSION, + TableProperties.DELETE_MODE, TableProperties.UPDATE_MODE, TableProperties.MERGE_MODE)); + } catch (NoSuchTableException ex) { if (!HiveTableUtil.isCtas(props)) { throw ex; diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java index a129266f6db0..94e7217cefc4 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveTableUtil.java @@ -246,6 +246,7 @@ public static Table deserializeTable(Configuration config, String name) { table = readTableObjectFromFile(location, config); } checkAndSetIoConfig(config, table); + IcebergVendedCredentialUtil.applyFromJobConf(table, config); // 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. @@ -362,8 +363,10 @@ static void cleanupTableObjectFile(String location, Configuration configuration) try { FileSystem fs = Util.getFs(toDelete, configuration); fs.delete(toDelete, true); - } catch (IOException ex) { - throw new UncheckedIOException(ex); + } catch (IOException e) { + // best effort: never fail the commit over the temp table-object file; with catalog-vended + // credentials the Hadoop filesystem has no keys and the query-scoped file is left behind + LOG.warn("Could not remove temp table object file: {}", filePath, e); } } diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergVendedCredentialUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergVendedCredentialUtil.java new file mode 100644 index 000000000000..8d110f40f706 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergVendedCredentialUtil.java @@ -0,0 +1,499 @@ +/* + * 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; + +import java.util.Base64; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConfUtil; +import org.apache.hadoop.hive.ql.plan.TableDesc; +import org.apache.hadoop.mapred.JobConf; +import org.apache.iceberg.BaseMetadataTable; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.MetadataTableType; +import org.apache.iceberg.MetadataTableUtils; +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.io.FileIO; +import org.apache.iceberg.io.StorageCredential; +import org.apache.iceberg.io.SupportsStorageCredentials; +import org.apache.iceberg.mr.InputFormatConfig; +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; + +/** + * Propagates vended storage credentials from an Iceberg {@link Table}'s {@link FileIO} to Hive job + * configuration so Tez/LLAP executors can access object storage without static catalog keys. + * + *

Limitation: credentials are minted at compile/job launch and are not refreshed — work that + * outlives the vended token lifetime fails authentication. + */ +public final class IcebergVendedCredentialUtil { + + private IcebergVendedCredentialUtil() { + } + + /** + * Copies vended credentials from the table FileIO into Hive job configuration. + * + *

Follows the HIVE-20651 split used by {@code JdbcStorageHandler}: sensitive values (including + * the serialized {@link StorageCredential} list) go to {@code jobSecrets}; non-secret config such + * as endpoint and path-style access go to {@code jobProperties}. + * + * @param table loaded Iceberg table; its catalog-qualified name keys the credentials blob, + * since REST catalogs mint credentials per table (the table pointer carries the same + * name, so executors resolve the matching key) + * @param catalogName Hive catalog name ({@link InputFormatConfig#CATALOG_NAME}) + * @param jobProperties Tez/MR non-secret job properties; may be {@code null} + * @param jobSecrets sensitive keys and serialized credentials; may be {@code null} + * @param conf session conf used to preserve host-side endpoint overrides + */ + public static void propagateToJob(Table table, String catalogName, Map jobProperties, + Map jobSecrets, Configuration conf) { + + List credentials = + withConfigurationOverrides(catalogName, extractCredentials(table), conf); + + if (credentials.isEmpty()) { + return; + } + + if (jobSecrets != null) { + jobSecrets.put( + InputFormatConfig.vendedCredentialsKey(table.name()), + serializeToSingleLineBase64(Lists.newArrayList(credentials))); + } + + for (StorageCredential credential : credentials) { + addCredentialEntries(catalogName, credential, jobProperties, jobSecrets, conf); + } + } + + /** + * Writes each key in one vended {@link StorageCredential} into job configuration. + * + *

Derives the S3 bucket 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. + */ + private static void addCredentialEntries(String catalogName, StorageCredential credential, + Map jobProperties, Map jobSecrets, Configuration conf) { + + String bucket = bucketFromPrefix(credential.prefix()); + for (Map.Entry entry : credential.config().entrySet()) { + addCredentialEntry( + catalogName, bucket, entry.getKey(), entry.getValue(), jobProperties, jobSecrets, conf); + } + } + + /** + * Routes one Iceberg credential config entry into job properties or secrets. + * + *

Skips blank values, applies session catalog overrides via {@link #resolveCredentialValue}, + * then sends non-secret keys (endpoint, path-style access, etc.) to {@code jobProperties} and + * 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, + Map jobProperties, Map jobSecrets, Configuration conf) { + + if (StringUtils.isBlank(value)) { + return; + } + String resolvedValue = resolveCredentialValue(catalogName, icebergKey, value, conf); + + if (jobProperties != null && !isSecretKey(icebergKey, conf)) { + addNonSecretCredentialEntry(catalogName, bucket, icebergKey, resolvedValue, jobProperties); + } + + if (jobSecrets != null && isSecretKey(icebergKey, conf)) { + addSecretCredentialEntry(bucket, icebergKey, resolvedValue, jobSecrets); + } + } + + /** + * Adds one non-secret vended value to {@code jobProperties} for Iceberg and Hadoop S3A. + * + *

When {@code catalogName} is set, writes {@code iceberg.catalog.<catalog>.<key>}. + * When {@code bucket} is set, also writes the matching {@code fs.s3a.bucket.<bucket>.*} + * key if {@link #toS3aBucketProperty} maps the Iceberg key. + */ + private static void addNonSecretCredentialEntry(String catalogName, String bucket, String icebergKey, String value, + Map jobProperties) { + + if (catalogName != null) { + String catalogConfigKey = + IcebergCatalogProperties.catalogPropertyConfigKey(catalogName, icebergKey); + jobProperties.putIfAbsent(catalogConfigKey, value); + } + + if (bucket != null) { + String s3aKey = toS3aBucketProperty(bucket, icebergKey); + if (s3aKey != null) { + jobProperties.putIfAbsent(s3aKey, 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 jobSecrets) { + if (bucket != null) { + String s3aSecretKey = toS3aBucketProperty(bucket, icebergKey); + if (s3aSecretKey != null) { + jobSecrets.putIfAbsent(s3aSecretKey, value); + } + } + } + + /** + * Applies vended credentials to the table FileIO, merging session/catalog conf overrides (e.g. S3 endpoint). + * Used on executors after deserialization and on HS2 commit when the table is taken from query state. + */ + public static void applyFromJobConf(Table table, Configuration conf) { + applyFromJobConf(table, conf != null ? conf.get(InputFormatConfig.CATALOG_NAME) : null, conf); + } + + /** Variant for callers that resolve the catalog name per table (HS2 commit paths, where the + * job-level {@link InputFormatConfig#CATALOG_NAME} is not set). */ + public static void applyFromJobConf(Table table, String catalogName, Configuration conf) { + if (table == null || conf == null) { + return; + } + + if (shouldSkipApplyFromJobConf(table.name(), catalogName, conf)) { + return; + } + + FileIO io = table.io(); + if (!(io instanceof SupportsStorageCredentials credentialIo)) { + return; + } + + List credentials = resolveCredentialsForApply(table, credentialIo, conf); + if (!credentials.isEmpty()) { + credentialIo.setCredentials(withConfigurationOverrides(catalogName, credentials, conf)); + } + } + + /** + * Returns true when the job carries no serialized vended credentials and the catalog is not + * configured for credential vending (no {@code vended-credentials} REST delegation header). + * Otherwise apply may restore credentials from the job conf or from the table FileIO. + */ + private static boolean shouldSkipApplyFromJobConf(String tableName, String catalogName, Configuration conf) { + return StringUtils.isBlank(conf.get(InputFormatConfig.vendedCredentialsKey(tableName))) && + !IcebergCatalogProperties.requestsVendedCredentials(catalogName, conf); + } + + /** + * Chooses which vended credentials {@link #applyFromJobConf} should install on the FileIO. + * + *

Uses the first non-empty source: credentials already on the FileIO (typical on HS2 after + * table load), else the base64 list from {@link InputFormatConfig#VENDED_STORAGE_CREDENTIALS} on + * the task {@code conf} (restored from the HIVE-20651 Credentials channel into table properties + * by {@code Utilities#copyJobSecretToTableProperties} and copied to the task-local conf), else + * {@link #extractCredentials(Table)} from the table (including FileIO property fallbacks). + */ + private static List resolveCredentialsForApply( + Table table, SupportsStorageCredentials credentialIo, Configuration conf) { + + List credentials = credentialIo.credentials(); + if (credentials != null && !credentials.isEmpty()) { + return credentials; + } + String serialized = conf.get(InputFormatConfig.vendedCredentialsKey(table.name())); + if (StringUtils.isNotBlank(serialized)) { + return SerializationUtil.deserializeFromBase64(serialized); + } + return extractCredentials(table); + } + + /** + * Returns true when the table catalog is configured for REST vended storage credentials. + */ + static boolean requestsVendedCredentials(Properties properties, Configuration configuration) { + if (properties == null) { + return false; + } + return IcebergCatalogProperties.requestsVendedCredentials( + properties.getProperty(InputFormatConfig.CATALOG_NAME), configuration); + } + + /** + * Loads a table and, if needed, bypasses the query-level cache so REST vended credentials are present on the FileIO. + * When vended credentials are not requested for the catalog, returns the cached table without an extra load. + */ + static Table getTableWithVendedCredentials(Properties properties, Configuration configuration) { + Table table = IcebergTableUtil.getTable(configuration, properties); + if (requestsVendedCredentials(properties, configuration) && extractCredentials(table).isEmpty()) { + table = IcebergTableUtil.getTable(configuration, properties, true); + } + return table; + } + + /** + * Reloads vended credentials at job launch when compile-time propagation missed them. + * Non-secret config is written to {@code jobConf}; secrets are merged into {@code TableDesc#getJobSecrets()} + * for {@link org.apache.hadoop.hive.ql.plan.PlanUtils#configureJobConf} (HIVE-20651). + */ + static void refreshVendedCredentialsIfMissing(TableDesc tableDesc, JobConf jobConf, Configuration configuration) { + if (tableDesc == null || tableDesc.getProperties() == null || + hasSerializedCredentials(tableDesc.getJobSecrets())) { + return; + } + + Properties props = tableDesc.getProperties(); + if (!requestsVendedCredentials(props, configuration)) { + return; + } + + String catalogName = props.getProperty(InputFormatConfig.CATALOG_NAME); + if (catalogName == null) { + return; + } + + try { + Table table = getTableWithVendedCredentials(props, configuration); + Map jobProps = new LinkedHashMap<>(); + Map secrets = new LinkedHashMap<>(); + propagateToJob(table, catalogName, jobProps, secrets, configuration); + jobProps.forEach(jobConf::set); + mergeJobSecrets(tableDesc, secrets); + } catch (NoSuchTableException ex) { + // Table may not exist yet for CTAS; credentials will not be available. + } + } + + /** + * Returns a copy of the table over a fresh secret-free {@link FileIO}, safe to serialize into + * job properties. {@code SerializableTable.copyOf} embeds {@code table.io()} — and with it the + * vended credentials — into the serialized bytes, so the copy is rebuilt from the same table + * metadata over a FileIO that keeps only the allowlisted non-secret properties: unknown keys + * never ship, whatever the storage provider. Executors restore the credentials from the + * HIVE-20651 Credentials channel via {@link #applyFromJobConf} + * ({@code SupportsStorageCredentials#setCredentials}). + */ + static Table secretFreeCopy(Table table, Configuration conf) { + if (table instanceof BaseMetadataTable metadataTable) { + Table base = secretFreeCopy(metadataTable.table(), conf); + return MetadataTableUtils.createMetadataTableInstance(base, metadataTableType(metadataTable)); + } + Preconditions.checkState(table instanceof HasTableOperations, + "Cannot build a secret-free copy of %s (%s)", table.name(), table.getClass().getName()); + TableMetadata metadata = ((HasTableOperations) table).operations().current(); + + FileIO io = table.io(); + Map ioProps = new LinkedHashMap<>(); + io.properties().forEach((k, v) -> { + if (!isSecretKey(k, conf)) { + ioProps.put(k, v); + } + }); + FileIO cleanIo = CatalogUtil.loadFileIO(io.getClass().getName(), ioProps, conf); + + return new BaseTable( + new StaticTableOperations(metadata, cleanIo, table.locationProvider()), table.name()); + } + + /** {@code metadataTableType()} is package-private; the name suffix is the type by construction. */ + private static MetadataTableType metadataTableType(BaseMetadataTable metadataTable) { + String name = metadataTable.name(); + MetadataTableType type = MetadataTableType.from(name.substring(name.lastIndexOf('.') + 1)); + Preconditions.checkState(type != null, "Cannot resolve metadata table type from %s", name); + return type; + } + + /** + * Single-line base64, unlike {@link SerializationUtil#serializeToBase64} which MIME-wraps with + * CR/LF. The blob is restored into table properties and copied to the task conf through + * {@code Utilities#copyTablePropertiesToConf}, whose {@code escapeJava} turns CR/LF into literal + * {@code \r\n} — and {@code r}/{@code n} are valid base64 alphabet, corrupting the decoded + * stream. {@link SerializationUtil#deserializeFromBase64} uses the MIME decoder, which accepts + * unwrapped input. + */ + private static String serializeToSingleLineBase64(Object obj) { + return Base64.getEncoder().encodeToString(SerializationUtil.serializeToBytes(obj)); + } + + private static boolean hasSerializedCredentials(Map jobSecrets) { + return jobSecrets != null && jobSecrets.keySet().stream() + .anyMatch(key -> key.startsWith(InputFormatConfig.VENDED_STORAGE_CREDENTIALS)); + } + + private static void mergeJobSecrets(TableDesc tableDesc, Map secrets) { + if (secrets.isEmpty()) { + return; + } + Map existing = tableDesc.getJobSecrets(); + if (existing == null) { + tableDesc.setJobSecrets(new LinkedHashMap<>(secrets)); + } else { + secrets.forEach(existing::putIfAbsent); + } + } + + /** + * A vended config key is secret — routed to the Credentials channel, never job properties, and + * stripped from the serialized table's FileIO — exactly when {@code hive.conf.hidden.list} covers + * it. That is Hive's own registry of sensitive configuration (the same one behind + * {@code HiveConf#isHiddenConfig}) that masks values in EXPLAIN, the Tez UI, and ATS. + */ + private static boolean isSecretKey(String key, Configuration conf) { + return HiveConfUtil.getHiddenSet(conf).stream().anyMatch(key::startsWith); + } + + static List extractCredentials(Table table) { + if (table == null) { + return List.of(); + } + FileIO io = table.io(); + if (io instanceof SupportsStorageCredentials credentialIo) { + List credentials = credentialIo.credentials(); + if (credentials != null && !credentials.isEmpty()) { + return credentials; + } + } + return credentialsFromFileIoProperties(table, io); + } + + private static List credentialsFromFileIoProperties(Table table, FileIO io) { + Map 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 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 target, Map 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 + "/"; + } + + /** + * REST catalogs vend credentials together with S3 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, + * {@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. + */ + private static List withConfigurationOverrides( + String catalogName, List credentials, Configuration conf) { + + if (credentials.isEmpty() || conf == null || catalogName == null) { + return credentials; + } + + List updated = Lists.newArrayListWithCapacity(credentials.size()); + for (StorageCredential credential : credentials) { + Map credsConfig = new LinkedHashMap<>(credential.config()); + applyCatalogConfigOverrides(catalogName, credsConfig, conf); + updated.add(StorageCredential.create(credential.prefix(), credsConfig)); + } + + return updated; + } + + /** + * Applies session-level catalog overrides to every entry of the given credential configuration, + * through the same {@link #resolveCredentialValue} used for the job-property entries — one + * resolver, one scope, so both channels always carry the same values. + */ + private static void applyCatalogConfigOverrides( + String catalogName, Map config, Configuration conf) { + config.replaceAll((icebergKey, value) -> resolveCredentialValue(catalogName, icebergKey, value, conf)); + } + + private static String resolveCredentialValue( + String catalogName, String icebergKey, String vendedValue, Configuration conf) { + if (conf == null || catalogName == null) { + return vendedValue; + } + String override = + 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/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriter.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriter.java index 0a3cce06611b..d7dfc8b95395 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriter.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriter.java @@ -24,6 +24,7 @@ import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapred.Reporter; import org.apache.iceberg.data.Record; +import org.apache.iceberg.io.FileIO; import org.apache.iceberg.mr.hive.FilesForCommit; import org.apache.iceberg.mr.mapred.Container; @@ -32,6 +33,8 @@ public interface HiveIcebergWriter extends FileSinkOperator.RecordWriter, FilesForCommit files(); + FileIO io(); + default void close(Reporter reporter) throws IOException { close(false); } diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriterBase.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriterBase.java index ad026a83040b..6267a3b99885 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriterBase.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergWriterBase.java @@ -69,6 +69,11 @@ abstract class HiveIcebergWriterBase implements HiveIcebergWriter { this.writer = writer; } + @Override + public FileIO io() { + return io; + } + @Override public void close(boolean abort) throws IOException { writer.close(); diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java index fa66b25b3490..89f888255b89 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java @@ -162,6 +162,8 @@ public List getSplits(JobContext context) { .orElseGet(() -> { Table tbl = Catalogs.loadTable(conf); conf.set(InputFormatConfig.TABLE_IDENTIFIER, tbl.name()); + // planning-local conf only (never shipped): for credential-vending catalogs the loaded + // table's FileIO carries secrets, which must not reach a serialized job configuration conf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tbl.name(), SerializationUtil.serializeToBase64(tbl)); return tbl; }); diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestIcebergVendedCredentialUtil.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestIcebergVendedCredentialUtil.java new file mode 100644 index 000000000000..1d80244e70d9 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestIcebergVendedCredentialUtil.java @@ -0,0 +1,885 @@ +/* + * 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; + +import java.nio.charset.StandardCharsets; +import java.security.PrivilegedExceptionAction; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.plan.PlanUtils; +import org.apache.hadoop.hive.ql.plan.TableDesc; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.MetadataTableType; +import org.apache.iceberg.MetadataTableUtils; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SerializableTable; +import org.apache.iceberg.StaticTableOperations; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableMetadataParser; +import org.apache.iceberg.aws.s3.S3FileIOProperties; +import org.apache.iceberg.hadoop.HadoopFileIO; +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.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.SerializationUtil; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link IcebergVendedCredentialUtil}: copying REST-vended storage credentials into Hive + * job configuration at plan time and re-applying them on executors. + */ +public class TestIcebergVendedCredentialUtil { + + /** + * Tests {@link IcebergVendedCredentialUtil#requestsVendedCredentials(Properties, Configuration)}. + * + *

Setup: table properties name catalog {@code ice01}; session {@code conf} has no + * delegation header, then gains {@code vended-credentials}. + * + *

Expects: {@code false} without the header, {@code true} with it — propagation and + * apply logic must not run for ordinary REST catalogs. + * + *

Why: Iceberg REST only vends storage credentials when the client sends + * {@code X-Iceberg-Access-Delegation: vended-credentials}; Hive mirrors that opt-in via catalog + * config so static catalog keys are not replaced unnecessarily. + */ + @Test + public void requestsVendedCredentialsRequiresDelegationHeader() { + Configuration conf = new Configuration(); + Properties props = new java.util.Properties(); + props.setProperty(InputFormatConfig.CATALOG_NAME, "ice01"); + + assertThat(IcebergVendedCredentialUtil.requestsVendedCredentials(props, conf)).isFalse(); + + conf.set( + "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation", + "vended-credentials"); + assertThat(IcebergVendedCredentialUtil.requestsVendedCredentials(props, conf)).isTrue(); + } + + /** + * Tests {@link IcebergVendedCredentialUtil#propagateToJob} when both {@code jobProperties} and + * {@code jobSecrets} are non-null (full compile-time path after both storage-handler hooks). + * + *

Setup: vended credential with internal endpoint {@code http://minio:9000}; session + * {@code conf} sets {@code iceberg.catalog.ice01.s3.endpoint} to {@code http://host:9000}. + * + *

Expects: endpoint and path-style in {@code jobProps} (Iceberg catalog keys and S3A + * per-bucket keys) with host endpoint; access/secret and serialized blob in {@code jobSecrets} + * only; blob endpoint also host; no secret material in {@code jobProps}. + * + *

Why: HIVE-20651 keeps secrets in {@code TableDesc#jobSecrets} (later Hadoop + * {@code Credentials}) and non-secrets in job conf. Catalogs often vend an internal S3 endpoint; + * HS2 overrides with a reachable host endpoint at propagation time so Tez tasks and Iceberg FileIO + * agree on connectivity while still using vended keys. + */ + @Test + public void propagateToJobMapsIcebergAndS3aProperties() { + Configuration conf = new Configuration(); + conf.set("iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT, "http://host:9000"); + + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of( + S3FileIOProperties.ACCESS_KEY_ID, "access", + S3FileIOProperties.SECRET_ACCESS_KEY, "secret", + S3FileIOProperties.ENDPOINT, "http://minio:9000", + S3FileIOProperties.PATH_STYLE_ACCESS, "true")))); + + Table table = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO), "db.t"); + Map jobProps = Maps.newHashMap(); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", jobProps, jobSecrets, conf); + + assertThat(jobProps) + .containsEntry( + "iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT, + "http://host:9000") + .containsEntry( + "fs.s3a.bucket.my-bucket.endpoint", + "http://host:9000") + .containsEntry( + "fs.s3a.bucket.my-bucket.path.style.access", + "true") + .doesNotContainKey(InputFormatConfig.vendedCredentialsKey("db.t")) + .doesNotContainKey("fs.s3a.bucket.my-bucket.access.key") + .doesNotContainKey( + "iceberg.catalog.ice01." + S3FileIOProperties.ACCESS_KEY_ID); + + assertThat(jobSecrets) + .containsEntry("fs.s3a.bucket.my-bucket.access.key", "access") + .containsEntry("fs.s3a.bucket.my-bucket.secret.key", "secret") + .doesNotContainKey( + "iceberg.catalog.ice01." + S3FileIOProperties.ACCESS_KEY_ID) + .doesNotContainKey( + "iceberg.catalog.ice01." + S3FileIOProperties.SECRET_ACCESS_KEY) + .satisfies(map -> + assertThat(map.get(InputFormatConfig.vendedCredentialsKey("db.t"))) + .isNotBlank()); + + List serialized = + SerializationUtil.deserializeFromBase64( + jobSecrets.get(InputFormatConfig.vendedCredentialsKey("db.t"))); + + assertThat(serialized.getFirst().prefix()).isEqualTo("s3://my-bucket/"); + assertThat(serialized.getFirst().config()) + .containsEntry(S3FileIOProperties.ENDPOINT, "http://host:9000") + .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access") + .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret") + .containsEntry(S3FileIOProperties.PATH_STYLE_ACCESS, "true"); + } + + /** + * Tests {@link IcebergVendedCredentialUtil#propagateToJob} with {@code jobSecrets == null}. + * + *

Setup: same mixed secret/non-secret credential as the full test; only + * {@code jobProperties} map is passed (models {@code configureInputJobProperties} / + * {@code overlayTableProperties}). + * + *

Expects: overridden endpoint in {@code jobProps}; no serialized blob and no Iceberg + * access-key property in {@code jobProps}. + * + *

Why: Hive calls the job-properties hook without a secrets map. This hook must add + * connectivity settings for S3A and Iceberg but must not place keys or the credential blob in + * plain job properties — those are filled later by {@code configureInputJobCredentials}. + */ + @Test + public void propagateNonSecretsOnlyWhenJobSecretsNull() { + Configuration conf = new Configuration(); + conf.set("iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT, "http://host:9000"); + + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of( + S3FileIOProperties.ACCESS_KEY_ID, "access", + S3FileIOProperties.SECRET_ACCESS_KEY, "secret", + S3FileIOProperties.ENDPOINT, "http://minio:9000", + S3FileIOProperties.PATH_STYLE_ACCESS, "true")))); + + Table table = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO), "db.t"); + Map jobProps = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", jobProps, null, conf); + + assertThat(jobProps) + .containsEntry( + "iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT, + "http://host:9000") + .containsEntry("fs.s3a.bucket.my-bucket.endpoint", "http://host:9000") + .containsEntry("fs.s3a.bucket.my-bucket.path.style.access", "true") + .doesNotContainKey(InputFormatConfig.vendedCredentialsKey("db.t")) + .doesNotContainKey( + "iceberg.catalog.ice01." + S3FileIOProperties.ACCESS_KEY_ID); + } + + /** + * Tests {@link IcebergVendedCredentialUtil#extractCredentials(Table)} when + * {@link SupportsStorageCredentials#credentials()} is empty. + * + *

Setup: FileIO initialized with {@code s3.access-key-id} / {@code s3.secret-access-key} + * in {@link FileIO#properties()} but an empty credential list. + * + *

Expects: one synthetic {@link StorageCredential} with the access key from properties. + * + *

Why: after {@code loadTable}, some REST clients populate {@code FileIO} properties + * before the credential list; propagation must still find vended keys or plan/launch would skip + * credential vending even though the table load succeeded. + */ + @Test + public void extractCredentialsFromFileIoPropertiesWhenCredentialListEmpty() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.initialize( + Map.of( + S3FileIOProperties.ACCESS_KEY_ID, "access", + S3FileIOProperties.SECRET_ACCESS_KEY, "secret", + S3FileIOProperties.ENDPOINT, "http://minio:9000")); + Schema schema = new Schema(Types.NestedField.required(1, "x", Types.IntegerType.get())); + TableMetadata metadata = + TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(), "s3://my-bucket/warehouse/t", Map.of()); + Table table = new BaseTable(new StaticTableOperations(metadata, fileIO), "db.t"); + + StorageCredential extracted = IcebergVendedCredentialUtil.extractCredentials(table).getFirst(); + assertThat(extracted.prefix()).isEqualTo("s3://my-bucket/"); + assertThat(extracted.config()) + .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access") + .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret") + .containsEntry(S3FileIOProperties.ENDPOINT, "http://minio:9000"); + } + + /** + * Tests {@link IcebergVendedCredentialUtil#propagateToJob} with {@code jobProperties == null}. + * + *

Setup: credential with access key and secret only; only {@code jobSecrets} map is + * passed (models {@code configureInputJobCredentials} for input and output tables). + * + *

Expects: serialized blob and {@code fs.s3a.bucket.*.access.key/secret.key} in + * {@code jobSecrets}; no catalog-prefixed secret keys, no endpoint in {@code jobSecrets}. + * + *

Why: the credentials hook receives only the secrets map. Propagation must still run + * without NPE, write material for {@code PlanUtils#configureJobConf} to move into + * {@code Credentials}, and avoid duplicating non-secrets or catalog-level secret keys that belong + * in job properties or only in the serialized blob. + */ + @Test + public void propagateSecretsOnlyWhenJobPropertiesNull() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of(S3FileIOProperties.ACCESS_KEY_ID, "access", + S3FileIOProperties.SECRET_ACCESS_KEY, "secret")))); + Table table = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO), "db.t"); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", null, jobSecrets, new Configuration()); + + assertThat(jobSecrets) + .containsEntry("fs.s3a.bucket.my-bucket.access.key", "access") + .containsEntry("fs.s3a.bucket.my-bucket.secret.key", "secret") + .doesNotContainKey( + "iceberg.catalog.ice01." + S3FileIOProperties.ACCESS_KEY_ID) + .doesNotContainKey( + "iceberg.catalog.ice01." + S3FileIOProperties.SECRET_ACCESS_KEY) + .doesNotContainKey("fs.s3a.bucket.my-bucket.endpoint") + .satisfies(map -> + assertThat(map.get(InputFormatConfig.vendedCredentialsKey("db.t"))) + .isNotBlank()); + + List serialized = + SerializationUtil.deserializeFromBase64( + jobSecrets.get(InputFormatConfig.vendedCredentialsKey("db.t"))); + + assertThat(serialized.getFirst().prefix()).isEqualTo("s3://my-bucket/"); + assertThat(serialized.getFirst().config()) + .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access") + .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret"); + } + + /** + * Tests {@link IcebergVendedCredentialUtil#applyFromJobConf} for a non-vending catalog. + * + *

Setup: task {@code conf} has catalog name only — no serialized credentials blob and + * no {@code vended-credentials} delegation header. + * + *

Expects: FileIO credential list stays empty after apply. + * + *

Why: {@link IcebergVendedCredentialUtil#shouldSkipApplyFromJobConf} must return early + * so tables that use static catalog S3 configuration are not cleared or rewritten on executors. + */ + @Test + public void applyFromJobConfSkipsWhenVendedCredentialsNotRequested() { + CredentialFileIO fileIO = new CredentialFileIO(); + Table table = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO), "db.t"); + + Configuration conf = new Configuration(); + conf.set("iceberg.catalog", "ice01"); + + IcebergVendedCredentialUtil.applyFromJobConf(table, conf); + + assertThat(fileIO.credentials()).isEmpty(); + } + + /** + * Tests {@link IcebergVendedCredentialUtil#applyFromJobConf} endpoint override on an existing FileIO. + * + *

Setup: FileIO already has vended credentials with {@code http://minio:9000}; task + * {@code conf} requests vending and sets host endpoint {@code http://host:9000}. + * + *

Expects: after apply, FileIO credential endpoint is {@code http://host:9000}. + * + *

Why: HS2 may retain credentials from {@code loadTable} with the catalog-internal + * endpoint; apply must re-merge session catalog S3 settings so commit paths and tasks use the + * same reachable endpoint as plan-time propagation. + */ + @Test + public void applyFromJobConfOverridesEndpointOnExistingCredentials() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of( + S3FileIOProperties.ACCESS_KEY_ID, "access", + S3FileIOProperties.SECRET_ACCESS_KEY, "secret", + S3FileIOProperties.ENDPOINT, "http://minio:9000")))); + Table table = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO), "db.t"); + + Configuration conf = new Configuration(); + conf.set("iceberg.catalog", "ice01"); + conf.set( + "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation", + "vended-credentials"); + conf.set("iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT, "http://host:9000"); + + IcebergVendedCredentialUtil.applyFromJobConf(table, conf); + + StorageCredential applied = + ((SupportsStorageCredentials) table.io()).credentials().getFirst(); + assertThat(applied.prefix()).isEqualTo("s3://my-bucket/"); + assertThat(applied.config()) + .containsEntry(S3FileIOProperties.ENDPOINT, "http://host:9000") + .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access") + .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret"); + } + + /** + * Tests {@link IcebergVendedCredentialUtil#applyFromJobConf} on a task with an empty FileIO. + * + *

Setup: {@code propagateToJob} fills job maps on HS2; keys are copied into a task + * {@code Configuration} as on Tez/LLAP; executor table is deserialized with a fresh FileIO (no + * credentials). + * + *

Expects: after apply, executor FileIO has one credential with the vended access key. + * + *

Why: executors do not re-run REST {@code loadTable}; they rebuild the table from job + * conf. {@link IcebergVendedCredentialUtil#resolveCredentialsForApply} must read the serialized + * blob (or S3A secret keys) from the task conf and call {@code setCredentials} so Iceberg I/O + * can access S3 during the task. + */ + @Test + public void applyFromJobConfRestoresCredentialsOnExecutor() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of(S3FileIOProperties.ACCESS_KEY_ID, "access", + S3FileIOProperties.SECRET_ACCESS_KEY, "secret")))); + Table table = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO), "db.t"); + + Map jobProps = Maps.newHashMap(); + Map jobSecrets = Maps.newHashMap(); + IcebergVendedCredentialUtil.propagateToJob( + table, "ice01", jobProps, jobSecrets, new Configuration()); + + Configuration taskConf = new Configuration(); + jobProps.forEach(taskConf::set); + jobSecrets.forEach(taskConf::set); + + CredentialFileIO executorIo = new CredentialFileIO(); + Table executorTable = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", executorIo), "db.t"); + IcebergVendedCredentialUtil.applyFromJobConf(executorTable, taskConf); + + assertThat(((SupportsStorageCredentials) executorTable.io()).credentials()).hasSize(1); + StorageCredential applied = + ((SupportsStorageCredentials) executorTable.io()).credentials().getFirst(); + assertThat(applied.prefix()).isEqualTo("s3://my-bucket/"); + assertThat(applied.config()) + .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access") + .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret"); + } + + /** + * Tests the full HIVE-20651 secret round trip: {@code propagateToJob} → + * {@link org.apache.hadoop.hive.ql.plan.PlanUtils#configureJobConf} (secrets → Hadoop + * {@code Credentials}, cleared from the TableDesc) → task UGI → + * {@code Utilities#copyJobSecretToTableProperties} → {@code Utilities#copyTablePropertiesToConf} + * → {@link IcebergVendedCredentialUtil#applyFromJobConf}. + * + *

Setup: the exact chain Hive runs — HS2 fills {@code TableDesc#jobSecrets}, PlanUtils + * moves them to the job {@code Credentials} (asserted: nothing lands in the plain JobConf), the + * task restores them into table properties and copies them into the task-local conf the same way + * {@code MapRecordProcessor}/{@code FileSinkOperator} do. + * + *

Expects: the executor FileIO receives the vended credential. + * + *

Why: this is the only sanctioned transport for secrets — they must survive the real + * restore path, including {@code copyTablePropertiesToConf}'s {@code escapeJava}, which corrupts + * MIME-wrapped base64 (CR/LF become literal {@code \r\n} whose {@code r}/{@code n} bytes are + * valid base64 alphabet). The blob must therefore be single-line base64. + */ + @Test + public void applyFromJobConfRestoresCredentialsViaCredentialsChannel() throws Exception { + CredentialFileIO hs2Io = new CredentialFileIO(); + hs2Io.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of(S3FileIOProperties.ACCESS_KEY_ID, "access", + S3FileIOProperties.SECRET_ACCESS_KEY, "secret")))); + Table hs2Table = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", hs2Io), "db.t"); + + TableDesc tableDesc = new TableDesc(); + Properties tableProps = new Properties(); + tableProps.setProperty(hive_metastoreConstants.META_TABLE_NAME, "db.t"); + tableDesc.setProperties(tableProps); + + Map jobSecrets = Maps.newHashMap(); + IcebergVendedCredentialUtil.propagateToJob(hs2Table, "ice01", null, jobSecrets, new Configuration()); + tableDesc.setJobSecrets(jobSecrets); + + JobConf jobConf = new JobConf(); + PlanUtils.configureJobConf(tableDesc, jobConf); + assertThat(jobConf.get(InputFormatConfig.vendedCredentialsKey("db.t"))).isNull(); + assertThat(tableDesc.getJobSecrets()).isEmpty(); + + UserGroupInformation taskUgi = UserGroupInformation.createRemoteUser("task-user"); + taskUgi.addCredentials(jobConf.getCredentials()); + + JobConf taskConf = new JobConf(); + taskConf.set(InputFormatConfig.CATALOG_NAME, "ice01"); + taskConf.set( + "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation", + "vended-credentials"); + + CredentialFileIO executorIo = new CredentialFileIO(); + Table executorTable = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", executorIo), "db.t"); + + taskUgi.doAs((PrivilegedExceptionAction) () -> { + Utilities.copyJobSecretToTableProperties(tableDesc); + Utilities.copyTablePropertiesToConf(tableDesc, taskConf); + IcebergVendedCredentialUtil.applyFromJobConf(executorTable, taskConf); + return null; + }); + + assertThat(executorIo.credentials()).hasSize(1); + StorageCredential applied = executorIo.credentials().getFirst(); + assertThat(applied.prefix()).isEqualTo("s3://my-bucket/"); + assertThat(applied.config()) + .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access") + .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret"); + } + + /** + * Tests {@link HiveIcebergStorageHandler#configureJobConf} secret hygiene. + * + *

Setup: vending-enabled catalog; {@code TableDesc#jobSecrets} already holds the + * serialized blob and a per-bucket S3A secret key (filled at compile time by + * {@code configureInputJobCredentials}). + * + *

Expects: after {@code configureJobConf}, neither the blob nor the S3A secret key + * appears in the {@code JobConf}; the jobSecrets map is left for + * {@code PlanUtils#configureJobConf} to move into Hadoop {@code Credentials}. + * + *

Why: {@code TezTask} strips {@code hive.conf.hidden.list} entries before + * storage handlers run ({@code createConfiguration} precedes + * {@code configureJobConfAndExtractJars}), so any secret written to the JobConf here ships in + * plaintext with the DAG plan. Secrets may only travel via the Credentials channel. + */ + @Test + public void configureJobConfKeepsSecretsOutOfJobConf() { + Configuration conf = new Configuration(); + conf.set( + "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation", + "vended-credentials"); + HiveIcebergStorageHandler handler = new HiveIcebergStorageHandler(); + handler.setConf(conf); + + Properties props = new Properties(); + props.setProperty(InputFormatConfig.CATALOG_NAME, "ice01"); + TableDesc tableDesc = new TableDesc(); + tableDesc.setProperties(props); + + String secretS3aKey = "fs.s3a.bucket.my-bucket.secret.key"; + Map jobSecrets = Maps.newHashMap(); + jobSecrets.put(InputFormatConfig.vendedCredentialsKey("db.t"), "blob"); + jobSecrets.put(secretS3aKey, "secret"); + tableDesc.setJobSecrets(jobSecrets); + + JobConf jobConf = new JobConf(); + jobConf.setBoolean(HiveConf.ConfVars.HIVE_IN_TEST_IDE.varname, true); + + handler.configureJobConf(tableDesc, jobConf); + + assertThat(jobConf.get(InputFormatConfig.vendedCredentialsKey("db.t"))).isNull(); + assertThat(jobConf.get(secretS3aKey)).isNull(); + assertThat(tableDesc.getJobSecrets()) + .containsEntry(InputFormatConfig.vendedCredentialsKey("db.t"), "blob") + .containsEntry(secretS3aKey, "secret"); + } + + /** + * Tests per-table credential resolution when two vended tables publish into one configuration. + * + *

Setup: REST catalogs mint credentials per table; two tables on different buckets go + * through the full chain — {@code propagateToJob} → {@code PlanUtils#configureJobConf} → one + * task UGI → per-table restore — and both publish into a shared conf the way merged + * map-works do. {@code Utilities#copyTableJobPropertiesToConf} is skip-if-present for table + * properties, so with a single global blob key the second table would read the first table's + * credentials. + * + *

Expects: each executor table's FileIO receives the credential vended for it — + * table 2 gets the bucket-b credential, not bucket-a's. + * + *

Why: the blob key must be table-scoped (like {@code iceberg.mr.serialized.table.*}) + * for multi-table queries; key collisions would silently apply another table's credentials. + */ + @Test + public void applyFromJobConfResolvesPerTableCredentialsInSharedConf() throws Exception { + TableDesc tableDesc1 = tableDescNamed("db.t1"); + Map secrets1 = Maps.newHashMap(); + IcebergVendedCredentialUtil.propagateToJob( + tableWithCredential("s3://bucket-a/", "access-a", "db.t1"), "ice01", null, secrets1, new Configuration()); + tableDesc1.setJobSecrets(secrets1); + + TableDesc tableDesc2 = tableDescNamed("db.t2"); + Map secrets2 = Maps.newHashMap(); + IcebergVendedCredentialUtil.propagateToJob( + tableWithCredential("s3://bucket-b/", "access-b", "db.t2"), "ice01", null, secrets2, new Configuration()); + tableDesc2.setJobSecrets(secrets2); + + JobConf jobConf = new JobConf(); + PlanUtils.configureJobConf(tableDesc1, jobConf); + PlanUtils.configureJobConf(tableDesc2, jobConf); + + UserGroupInformation taskUgi = UserGroupInformation.createRemoteUser("task-user"); + taskUgi.addCredentials(jobConf.getCredentials()); + + JobConf sharedConf = new JobConf(); + sharedConf.set(InputFormatConfig.CATALOG_NAME, "ice01"); + sharedConf.set( + "iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation", + "vended-credentials"); + + CredentialFileIO executorIo1 = new CredentialFileIO(); + Table executorTable1 = + new BaseTable(new StaticTableOperations("s3://bucket-a/t1", executorIo1), "db.t1"); + CredentialFileIO executorIo2 = new CredentialFileIO(); + Table executorTable2 = + new BaseTable(new StaticTableOperations("s3://bucket-b/t2", executorIo2), "db.t2"); + + taskUgi.doAs((PrivilegedExceptionAction) () -> { + Utilities.copyJobSecretToTableProperties(tableDesc1); + Utilities.copyJobSecretToTableProperties(tableDesc2); + Utilities.copyTableJobPropertiesToConf(tableDesc1, sharedConf); + Utilities.copyTableJobPropertiesToConf(tableDesc2, sharedConf); + IcebergVendedCredentialUtil.applyFromJobConf(executorTable1, sharedConf); + IcebergVendedCredentialUtil.applyFromJobConf(executorTable2, sharedConf); + return null; + }); + + assertThat(executorIo1.credentials()).hasSize(1); + assertThat(executorIo1.credentials().getFirst().prefix()).isEqualTo("s3://bucket-a/"); + assertThat(executorIo1.credentials().getFirst().config()) + .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access-a"); + + assertThat(executorIo2.credentials()).hasSize(1); + assertThat(executorIo2.credentials().getFirst().prefix()).isEqualTo("s3://bucket-b/"); + assertThat(executorIo2.credentials().getFirst().config()) + .containsEntry(S3FileIOProperties.ACCESS_KEY_ID, "access-b"); + } + + private static TableDesc tableDescNamed(String tableName) { + Properties tableProps = new Properties(); + tableProps.setProperty(hive_metastoreConstants.META_TABLE_NAME, tableName); + TableDesc tableDesc = new TableDesc(); + tableDesc.setProperties(tableProps); + return tableDesc; + } + + private static Table tableWithCredential(String prefix, String accessKey, String tableName) { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + prefix, + Map.of(S3FileIOProperties.ACCESS_KEY_ID, accessKey, + S3FileIOProperties.SECRET_ACCESS_KEY, accessKey + "-secret")))); + return new BaseTable(new StaticTableOperations(prefix + "t", fileIO), tableName); + } + + /** + * Tests the strip → ship → setCredentials flow for vended catalogs. + * + *

Setup: HS2-side table with a credentialed FileIO (secret in the credential list and + * in the FileIO properties) over a real metadata file on local disk. {@code secretFreeCopy} + * rebuilds the table over a fresh FileIO keeping only allowlisted non-secret properties before + * {@code serializeTable}; credentials travel separately via {@code propagateToJob} (modelling + * the HIVE-20651 restore); {@code HiveTableUtil#deserializeTable} restores them on the + * deserialized FileIO. + * + *

Expects: the serialized table carries no secret bytes; the deserialized table has + * the original qualified name, schema, FileIO class with non-secret properties only, and the + * vended credentials restored from the Credentials-channel key. + * + *

Why: a serialized table embeds its FileIO, so the copy shipped through job + * properties must carry a FileIO with no secret state; executors re-attach the credentials + * through {@code SupportsStorageCredentials#setCredentials}. + */ + @Test + public void vendedSerializedTableShipsSecretFreeAndRestoresCredentials() throws Exception { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.initialize( + Map.of( + S3FileIOProperties.SECRET_ACCESS_KEY, "prop-secret-11bb", + S3FileIOProperties.ENDPOINT, "http://minio:9000")); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of(S3FileIOProperties.ACCESS_KEY_ID, "access", + S3FileIOProperties.SECRET_ACCESS_KEY, "cred-secret-11bb")))); + + Schema schema = new Schema(Types.NestedField.required(1, "x", Types.IntegerType.get())); + TableMetadata metadata = + TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(), "s3://my-bucket/warehouse/t", Map.of()); + String metadataLocation = "file://" + + java.nio.file.Files.createTempDirectory("iceberg-strip-test").resolve("v1.metadata.json"); + TableMetadataParser.write(metadata, fileIO.newOutputFile(metadataLocation)); + Table hs2Table = new BaseTable(new StaticTableOperations(metadataLocation, fileIO), "ice01.db.t"); + + Configuration conf = new Configuration(); + String serialized = HiveTableUtil.serializeTable( + IcebergVendedCredentialUtil.secretFreeCopy(hs2Table, conf), conf, new Properties(), null); + + String rawBytes = new String( + java.util.Base64.getMimeDecoder().decode(serialized), StandardCharsets.ISO_8859_1); + assertThat(rawBytes) + .doesNotContain("prop-secret-11bb") + .doesNotContain("cred-secret-11bb") + .contains("http://minio:9000"); + + Map jobSecrets = Maps.newHashMap(); + IcebergVendedCredentialUtil.propagateToJob(hs2Table, "ice01", null, jobSecrets, new Configuration()); + + JobConf taskConf = new JobConf(); + taskConf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + "db.t", serialized); + jobSecrets.forEach(taskConf::set); + taskConf.set(InputFormatConfig.CATALOG_NAME, "ice01"); + + Table executorTable = HiveTableUtil.deserializeTable(taskConf, "db.t"); + + assertThat(executorTable).isNotNull(); + assertThat(executorTable.name()).isEqualTo("ice01.db.t"); + assertThat(executorTable.schema().asStruct()).isEqualTo(schema.asStruct()); + assertThat(executorTable.io()).isInstanceOf(CredentialFileIO.class); + assertThat(executorTable.io().properties()) + .containsEntry(S3FileIOProperties.ENDPOINT, "http://minio:9000") + .doesNotContainKey(S3FileIOProperties.SECRET_ACCESS_KEY); + assertThat(((SupportsStorageCredentials) executorTable.io()).credentials()).hasSize(1); + } + + /** + * Tests the strip → ship → setCredentials flow for metadata tables ({@code db.t.files}-style + * queries). + * + *

Setup: a {@code FILES} metadata table over a credentialed base; the secret-free + * copy is serialized and the table restored through {@code HiveTableUtil#deserializeTable}. + * + *

Expects: the serialized bytes carry no secrets; the deserialized table keeps the + * metadata-table name and schema over a secret-free FileIO with the credentials restored. + * + *

Why: metadata tables share the base table FileIO and would leak the same way; the + * secret-free copy must recreate the metadata-table view over the clean FileIO. + */ + @Test + public void vendedMetadataTableSerializesSecretFreeAndRestoresCredentials() throws Exception { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.initialize(Map.of(S3FileIOProperties.SECRET_ACCESS_KEY, "prop-secret-3fd2")); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of(S3FileIOProperties.SECRET_ACCESS_KEY, "cred-secret-3fd2")))); + + Schema schema = new Schema(Types.NestedField.required(1, "x", Types.IntegerType.get())); + TableMetadata metadata = + TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(), "s3://my-bucket/warehouse/t", Map.of()); + String metadataLocation = "file://" + + java.nio.file.Files.createTempDirectory("iceberg-strip-meta-test").resolve("v1.metadata.json"); + TableMetadataParser.write(metadata, fileIO.newOutputFile(metadataLocation)); + Table baseTable = new BaseTable(new StaticTableOperations(metadataLocation, fileIO), "ice01.db.t"); + Table filesTable = MetadataTableUtils.createMetadataTableInstance(baseTable, MetadataTableType.FILES); + + Configuration conf = new Configuration(); + String serialized = HiveTableUtil.serializeTable( + IcebergVendedCredentialUtil.secretFreeCopy(filesTable, conf), conf, new Properties(), null); + + String rawBytes = new String( + java.util.Base64.getMimeDecoder().decode(serialized), StandardCharsets.ISO_8859_1); + assertThat(rawBytes).doesNotContain("cred-secret-3fd2").doesNotContain("prop-secret-3fd2"); + + Map jobSecrets = Maps.newHashMap(); + IcebergVendedCredentialUtil.propagateToJob(filesTable, "ice01", null, jobSecrets, new Configuration()); + + JobConf taskConf = new JobConf(); + taskConf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + "db.t.files", serialized); + jobSecrets.forEach(taskConf::set); + taskConf.set(InputFormatConfig.CATALOG_NAME, "ice01"); + + Table executorTable = HiveTableUtil.deserializeTable(taskConf, "db.t.files"); + + assertThat(executorTable).isNotNull(); + assertThat(executorTable.name()).isEqualTo(filesTable.name()); + assertThat(executorTable.schema().asStruct()).isEqualTo(filesTable.schema().asStruct()); + assertThat(executorTable.io().properties()) + .doesNotContainKey(S3FileIOProperties.SECRET_ACCESS_KEY); + assertThat(((SupportsStorageCredentials) executorTable.io()).credentials()).hasSize(1); + } + + /** + * Tests the HS2 commit-path variant: {@code collectOutputs} resolves the catalog name per table + * (the job-level {@code CATALOG_NAME} is not set in the DAG conf) and session endpoint overrides + * must still reach the live table's FileIO through {@code setCredentials}. + */ + @Test + public void applyFromJobConfWithExplicitCatalogAppliesEndpointOverride() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of(S3FileIOProperties.ACCESS_KEY_ID, "commit-access", + S3FileIOProperties.SECRET_ACCESS_KEY, "commit-secret-4fa1", + S3FileIOProperties.ENDPOINT, "http://minio:9000")))); + Schema schema = new Schema(Types.NestedField.required(1, "x", Types.IntegerType.get())); + TableMetadata metadata = + TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(), "s3://my-bucket/warehouse/t", Map.of()); + Table table = new BaseTable(new StaticTableOperations(metadata, fileIO), "ice01.db.t"); + + JobConf jobConf = new JobConf(); + jobConf.set("iceberg.catalog.ice01.header.X-Iceberg-Access-Delegation", "vended-credentials"); + jobConf.set("iceberg.catalog.ice01." + S3FileIOProperties.ENDPOINT, "http://localhost:9100"); + + IcebergVendedCredentialUtil.applyFromJobConf(table, "ice01", jobConf); + + assertThat(fileIO.credentials()).hasSize(1); + assertThat(fileIO.credentials().get(0).config()) + .containsEntry(S3FileIOProperties.ENDPOINT, "http://localhost:9100") + .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "commit-secret-4fa1"); + } + + /** + * Pins the leak that makes the secret-free copy before serialization necessary. + * + *

Setup: {@code SerializableTable.copyOf} stores a reference to the live + * table's FileIO in a non-transient field, and Java serialization walks it — so serializing an + * unsanitized copy embeds the FileIO's credential list and secret properties in the blob. The + * stub mirrors {@code S3FileIO}'s serialized shape: {@code properties} and + * {@code storageCredentials} are both non-transient there too (the real implementation is + * covered by the Gravitino qtest). + * + *

Expects: secrets ARE present in the raw serialized bytes when no sanitization runs. + * + *

Why: guards the premise. If Iceberg ever stops serializing FileIO secret state, + * this test fails and serializing tables as-is becomes safe again. + */ + @Test + public void serializableTableCopyLeaksIoSecretsWithoutSanitization() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.initialize(Map.of(S3FileIOProperties.SECRET_ACCESS_KEY, "prop-secret-77aa")); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "s3://my-bucket/", + Map.of(S3FileIOProperties.ACCESS_KEY_ID, "cred-access-77aa", + S3FileIOProperties.SECRET_ACCESS_KEY, "cred-secret-77aa")))); + Schema schema = new Schema(Types.NestedField.required(1, "x", Types.IntegerType.get())); + TableMetadata metadata = + TableMetadata.newTableMetadata(schema, PartitionSpec.unpartitioned(), "s3://my-bucket/warehouse/t", Map.of()); + Table table = new BaseTable(new StaticTableOperations(metadata, fileIO), "db.t"); + + String unsanitized = SerializationUtil.serializeToBase64(SerializableTable.copyOf(table)); + + String rawBytes = new String( + java.util.Base64.getMimeDecoder().decode(unsanitized), StandardCharsets.ISO_8859_1); + assertThat(rawBytes) + .contains("cred-access-77aa") + .contains("cred-secret-77aa") + .contains("prop-secret-77aa"); + } + + /** Public with a no-arg constructor: the secret-free copy instantiates it reflectively via + * {@code CatalogUtil.loadFileIO}. File operations delegate to local Hadoop IO so metadata + * files can be written and read in tests. */ + public static class CredentialFileIO implements FileIO, SupportsStorageCredentials { + private List credentials = List.of(); + private Map properties = Map.of(); + private transient HadoopFileIO localFiles; + + @Override + public void initialize(Map props) { + this.properties = props; + } + + private HadoopFileIO localFiles() { + if (localFiles == null) { + localFiles = new HadoopFileIO(new Configuration()); + } + return localFiles; + } + + @Override + public org.apache.iceberg.io.InputFile newInputFile(String path) { + return localFiles().newInputFile(path); + } + + @Override + public org.apache.iceberg.io.OutputFile newOutputFile(String path) { + return localFiles().newOutputFile(path); + } + + @Override + public void deleteFile(String path) { + localFiles().deleteFile(path); + } + + @Override + public Map properties() { + return properties; + } + + @Override + public void close() { + // No-op: test stub does not hold resources. + } + + @Override + public List credentials() { + return credentials; + } + + @Override + public void setCredentials(List creds) { + this.credentials = creds; + } + } +} diff --git a/iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q b/iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q index 91aa14fdc170..66d11bbd8701 100644 --- a/iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q +++ b/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 @@ -24,9 +28,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.header.X-Iceberg-Access-Delegation=vended-credentials; ---! This config is set in the driver setup (see TestIcebergRESTCatalogLlapLocalCliDriver.java) ---! conf.set('iceberg.catalog.ice01.uri', ); +--! REST URI, OAuth, MinIO + Gravitino S3 warehouse / credential vending, and host S3A are set in +--! TestIcebergRESTCatalogGravitinoLlapLocalCliDriver. create database ice_rest; use ice_rest; @@ -64,11 +69,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; diff --git a/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out b/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out index 5821007cd3cb..db4e5c2129a4 100644 --- a/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out +++ b/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_rest_catalog_gravitino.q.out @@ -77,14 +77,14 @@ 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', -#### A masked pattern was here #### + 'metadata_location'='#Masked#', 'name'='ice_rest.ice_orc2', 'parquet.compression'='zstd', 'serialization.format'='1', @@ -139,7 +139,7 @@ Table Parameters: default-partition-spec {\"spec-id\":0,\"fields\":[{\"name\":\"company_id\",\"transform\":\"identity\",\"source-id\":5,\"field-id\":1000}]} format-version 2 iceberg.catalog ice01 -#### A masked pattern was here #### + metadata_location #Masked# name ice_rest.ice_orc2 numFiles 1 numRows 3 @@ -220,7 +220,7 @@ Table Type: VIRTUAL_VIEW Table Parameters: 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\"}]} -#### A masked pattern was here #### + metadata_location #Masked# storage_handler org.apache.iceberg.mr.hive.HiveIcebergStorageHandler table_type ICEBERG-VIEW type rest @@ -279,7 +279,7 @@ Table Type: VIRTUAL_VIEW Table Parameters: 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\"}]} -#### A masked pattern was here #### + metadata_location #Masked# storage_handler org.apache.iceberg.mr.hive.HiveIcebergStorageHandler table_type ICEBERG-VIEW type rest @@ -336,7 +336,7 @@ Table Type: VIRTUAL_VIEW Table Parameters: bucketing_version 2 current-schema {\"type\":\"struct\",\"schema-id\":1,\"fields\":[{\"id\":1,\"name\":\"_c0\",\"required\":false,\"type\":\"string\"}]} -#### A masked pattern was here #### + metadata_location #Masked# storage_handler org.apache.iceberg.mr.hive.HiveIcebergStorageHandler table_type ICEBERG-VIEW type rest @@ -405,7 +405,7 @@ Table Type: VIRTUAL_VIEW Table Parameters: bucketing_version 2 current-schema {\"type\":\"struct\",\"schema-id\":0,\"fields\":[{\"id\":1,\"name\":\"first_name\",\"required\":false,\"type\":\"string\"},{\"id\":2,\"name\":\"_c1\",\"required\":false,\"type\":\"string\"}]} -#### A masked pattern was here #### + metadata_location #Masked# storage_handler org.apache.iceberg.mr.hive.HiveIcebergStorageHandler table_type ICEBERG-VIEW type rest diff --git a/itests/qtest-iceberg/pom.xml b/itests/qtest-iceberg/pom.xml index 3dc736007f4e..035c79a14453 100644 --- a/itests/qtest-iceberg/pom.xml +++ b/itests/qtest-iceberg/pom.xml @@ -140,6 +140,11 @@ junit test + + org.assertj + assertj-core + test + com.sun.jersey jersey-servlet @@ -156,6 +161,11 @@ hadoop-archives test + + org.apache.hadoop + hadoop-aws + test + org.apache.hadoop hadoop-common @@ -271,6 +281,10 @@ hadoop-mapreduce-client-core test + + org.apache.hive + hive-iceberg-handler + org.apache.hive hive-iceberg-handler @@ -364,6 +378,12 @@ hbase-mapreduce test + + org.apache.iceberg + iceberg-aws + ${iceberg.version} + test + org.apache.tez tez-tests @@ -496,6 +516,12 @@ + + software.amazon.awssdk + bundle + ${aws-java-sdk.version} + test + org.testcontainers testcontainers diff --git a/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/OzoneS3GatewayContainers.java b/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/OzoneS3GatewayContainers.java new file mode 100644 index 000000000000..f77254fff3ea --- /dev/null +++ b/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/OzoneS3GatewayContainers.java @@ -0,0 +1,198 @@ +/* + * 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.hadoop.hive.cli; + +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.containers.wait.strategy.WaitAllStrategy; +import org.testcontainers.utility.DockerImageName; + +/** + * Minimal Apache Ozone cluster with S3 Gateway for q-tests that need an S3-compatible endpoint. + * + *

Layout matches {@code packaging/src/docker/storage/ozone/docker-compose.yml}: scm, om, datanode, + * and s3g on a shared Docker network. S3G is reachable as {@link #S3_DOCKER_ALIAS}:{@link #S3G_PORT} + * from other containers (for example Gravitino) and via the published host port from the test JVM. + */ +public final class OzoneS3GatewayContainers { + + private static final Logger LOG = LoggerFactory.getLogger(OzoneS3GatewayContainers.class); + private static final DockerImageName OZONE_IMAGE = DockerImageName.parse("apache/ozone:2.1.0"); + + private static final int SCM_PORT = 9876; + private static final int OM_PORT = 9874; + /** Ozone 2.1.0 HDDS datanode client RPC port ({@code hdds.datanode.client.port}). */ + private static final int DATANODE_CLIENT_PORT = 19864; + + private static final String S3_BUCKET_READY_MARKER = "OZONE_S3_BUCKET_READY"; + + public static final int S3G_PORT = 9878; + public static final String S3_DOCKER_ALIAS = "s3.ozone"; + public static final String S3_DOCKER_ENDPOINT = String.format("http://%s:%d", S3_DOCKER_ALIAS, S3G_PORT); + + private static final Duration STARTUP_TIMEOUT = Duration.ofMinutes(5); + + private GenericContainer scm; + private GenericContainer om; + private GenericContainer datanode; + private GenericContainer s3g; + + /** + * Starts scm → om → datanode → s3g on {@code network} and creates {@code bucketName} under the + * default S3 volume ({@code /s3v}). + */ + @SuppressWarnings("resource") + public void start(Network network, String bucketName) { + Map commonEnv = commonEnv(); + scm = startScm(network, commonEnv); + om = startOm(network, commonEnv); + datanode = startDatanode(network, commonEnv); + s3g = startS3g(network, commonEnv, bucketName); + LOG.info("Ozone S3G ready at {} (host {}:{})", S3_DOCKER_ENDPOINT, s3g.getHost(), s3g.getMappedPort(S3G_PORT)); + } + + public void stop() { + stopQuietly(s3g); + stopQuietly(datanode); + stopQuietly(om); + stopQuietly(scm); + s3g = null; + datanode = null; + om = null; + scm = null; + } + + public String getHost() { + return s3g.getHost(); + } + + public int getMappedPort() { + return s3g.getMappedPort(S3G_PORT); + } + + private static Map commonEnv() { + Map env = new HashMap<>(); + env.put("OZONE-SITE.XML_hdds.datanode.dir", "/data/hdds"); + env.put("OZONE-SITE.XML_ozone.metadata.dirs", "/data/metadata"); + env.put("OZONE-SITE.XML_ozone.om.address", "om"); + env.put("OZONE-SITE.XML_ozone.om.http-address", "om:" + OM_PORT); + env.put("OZONE-SITE.XML_ozone.replication", "1"); + env.put("OZONE-SITE.XML_ozone.scm.block.client.address", "scm"); + env.put("OZONE-SITE.XML_ozone.scm.client.address", "scm"); + env.put("OZONE-SITE.XML_ozone.scm.datanode.id.dir", "/data/metadata"); + env.put("OZONE-SITE.XML_ozone.scm.names", "scm"); + env.put("OZONE-SITE.XML_hdds.scm.safemode.min.datanode", "1"); + env.put("OZONE-SITE.XML_hdds.scm.safemode.healthy.pipeline.pct", "0"); + env.put("OZONE-SITE.XML_hdds.scm.safemode.enabled", "false"); + // Path-style only: host JVM and Gravitino use http://host:9878/bucket/... . Setting + // ozone.s3g.domain.name would require virtual-host Host headers (e.g. bucket.s3.ozone). + // Keep CI/docker-desktop friendly (see itests/test-docker/helm/ozone/values.yaml). + env.put("OZONE-SITE.XML_hdds.datanode.volume.min.free.space", "256MB"); + env.put("OZONE-SITE.XML_ozone.scm.container.size", "128MB"); + env.put("OZONE-SITE.XML_ozone.scm.block.size", "32MB"); + env.put("no_proxy", "om,scm,s3g,localhost,127.0.0.1"); + return env; + } + + @SuppressWarnings("resource") + private static GenericContainer startScm(Network network, Map commonEnv) { + GenericContainer container = new GenericContainer<>(OZONE_IMAGE) + .withNetwork(network) + .withNetworkAliases("scm") + .withExposedPorts(SCM_PORT) + .withEnv(commonEnv) + .withEnv("ENSURE_SCM_INITIALIZED", "/data/metadata/scm/current/VERSION") + .withCommand("ozone", "scm") + .waitingFor(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT)) + .withLogConsumer(outputFrame -> LOG.debug("[ozone-scm] {}", outputFrame.getUtf8String().trim())); + container.start(); + return container; + } + + @SuppressWarnings("resource") + private static GenericContainer startOm(Network network, Map commonEnv) { + GenericContainer container = new GenericContainer<>(OZONE_IMAGE) + .withNetwork(network) + .withNetworkAliases("om") + .withExposedPorts(OM_PORT) + .withEnv(commonEnv) + .withEnv("CORE-SITE.XML_hadoop.proxyuser.hadoop.hosts", "*") + .withEnv("CORE-SITE.XML_hadoop.proxyuser.hadoop.groups", "*") + .withEnv("ENSURE_OM_INITIALIZED", "/data/metadata/om/current/VERSION") + .withEnv("WAITFOR", "scm:" + SCM_PORT) + .withCommand("ozone", "om") + .waitingFor(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT)) + .withLogConsumer(outputFrame -> LOG.debug("[ozone-om] {}", outputFrame.getUtf8String().trim())); + container.start(); + return container; + } + + @SuppressWarnings("resource") + private static GenericContainer startDatanode(Network network, Map commonEnv) { + GenericContainer container = new GenericContainer<>(OZONE_IMAGE) + .withNetwork(network) + .withExposedPorts(DATANODE_CLIENT_PORT) + .withEnv(commonEnv) + .withCommand("ozone", "datanode") + .waitingFor(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT)) + .withLogConsumer(outputFrame -> LOG.debug("[ozone-dn] {}", outputFrame.getUtf8String().trim())); + container.start(); + return container; + } + + @SuppressWarnings("resource") + private static GenericContainer startS3g(Network network, Map commonEnv, String bucketName) { + String bootstrap = "" + + "set -e\n" + + "ozone s3g &\n" + + "s3g_pid=$!\n" + + "until ozone sh volume list >/dev/null 2>&1; do echo 'waiting for OM...' && sleep 1; done\n" + + "ozone sh volume create /s3v || true\n" + + "ozone sh bucket delete /s3v/" + bucketName + " || true\n" + + "ozone sh bucket create /s3v/" + bucketName + "\n" + + "echo " + S3_BUCKET_READY_MARKER + "\n" + + "wait \"$s3g_pid\"\n"; + GenericContainer container = new GenericContainer<>(OZONE_IMAGE) + .withNetwork(network) + .withNetworkAliases(S3_DOCKER_ALIAS) + .withExposedPorts(S3G_PORT) + .withEnv(commonEnv) + .withEnv("WAITFOR", "om:" + OM_PORT) + .withCommand("sh", "-c", bootstrap) + .waitingFor(new WaitAllStrategy() + .withStrategy(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT)) + .withStrategy(Wait.forLogMessage(".*" + S3_BUCKET_READY_MARKER + ".*\\n", 1) + .withStartupTimeout(STARTUP_TIMEOUT))) + .withLogConsumer(outputFrame -> LOG.info("[ozone-s3g] {}", outputFrame.getUtf8String().trim())); + container.start(); + return container; + } + + private static void stopQuietly(GenericContainer container) { + if (container != null) { + container.stop(); + } + } +} diff --git a/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java b/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java index 89545f83a074..4c04578e2443 100644 --- a/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java +++ b/itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java @@ -18,9 +18,6 @@ package org.apache.hadoop.hive.cli; -import com.github.dockerjava.api.command.CopyArchiveFromContainerCmd; -import org.apache.commons.compress.archivers.tar.TarArchiveEntry; -import org.apache.commons.compress.archivers.tar.TarArchiveInputStream; import org.apache.commons.io.FileUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hive.cli.control.CliAdapter; @@ -28,7 +25,10 @@ import org.apache.hadoop.hive.metastore.conf.MetastoreConf; import org.apache.hadoop.hive.ql.session.SessionState; import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.aws.AwsClientProperties; +import org.apache.iceberg.aws.s3.S3FileIOProperties; import org.apache.iceberg.hive.IcebergCatalogProperties; +import org.apache.iceberg.hive.client.RestAccessDelegationMode; import org.apache.iceberg.hive.client.HiveRESTCatalogClient; import org.apache.iceberg.rest.extension.OAuth2AuthorizationServer; import org.junit.After; @@ -42,13 +42,13 @@ import org.junit.runners.Parameterized.Parameters; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.Network; import org.testcontainers.containers.output.Slf4jLogConsumer; import org.testcontainers.containers.wait.strategy.Wait; import org.testcontainers.containers.wait.strategy.WaitAllStrategy; import org.testcontainers.utility.DockerImageName; import org.testcontainers.utility.MountableFile; -import org.testcontainers.containers.GenericContainer; import java.io.File; import java.io.IOException; @@ -59,22 +59,30 @@ import java.nio.file.Paths; import java.time.Duration; import java.util.List; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; +/** + * LLAP {@link CliAdapter} qtests for Hive against the Gravitino Iceberg REST server image + * ({@link #GRAVITINO_IMAGE}), with OAuth2 on the catalog HTTP API and an Iceberg warehouse on + * Apache Ozone S3 Gateway using Gravitino {@code s3-secret-key} credential vending (see + * {@link #GRAVITINO_S3_CONF_TEMPLATE}). + * + *

Table metadata and data live under {@code s3://} in {@link #S3_BUCKET}. The host temp directory + * {@link #warehouseDir} is used to render the Gravitino server configuration and as the source path for + * {@code .withCopyFileToContainer} (config + H2 driver JAR).

+ */ @RunWith(Parameterized.class) public class TestIcebergRESTCatalogGravitinoLlapLocalCliDriver { private static final CliAdapter CLI_ADAPTER = new CliConfigs.TestIcebergRESTCatalogGravitinoLlapLocalCliDriver().getCliAdapter(); - + private static final Logger LOG = LoggerFactory.getLogger(TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.class); - + private static final String CATALOG_NAME = "ice01"; private static final long GRAVITINO_STARTUP_TIMEOUT_MINUTES = 5L; private static final int GRAVITINO_HTTP_PORT = 9001; - private static final String GRAVITINO_CONF_FILE_TEMPLATE = "gravitino-h2-test-template.conf"; + + private static final String GRAVITINO_S3_CONF_TEMPLATE = "gravitino-s3-vended-oauth-template.conf"; private static final String GRAVITINO_ROOT_DIR = "/root/gravitino-iceberg-rest-server"; private static final String GRAVITINO_STARTUP_SCRIPT = GRAVITINO_ROOT_DIR + "/bin/start-iceberg-rest-server.sh"; private static final String GRAVITINO_H2_LIB = GRAVITINO_ROOT_DIR + "/libs/h2-driver.jar"; @@ -82,6 +90,8 @@ public class TestIcebergRESTCatalogGravitinoLlapLocalCliDriver { private static final DockerImageName GRAVITINO_IMAGE = DockerImageName.parse("apache/gravitino-iceberg-rest:1.0.0"); + private static final String S3_BUCKET = "iceberg-vend"; + private static final String OAUTH2_SERVER_ICEBERG_CLIENT_ID = "iceberg-client"; private static final String OAUTH2_SERVER_ICEBERG_CLIENT_SECRET = "iceberg-client-secret"; @@ -89,8 +99,8 @@ public class TestIcebergRESTCatalogGravitinoLlapLocalCliDriver { private final File qfile; private GenericContainer gravitinoContainer; + private OzoneS3GatewayContainers ozoneS3; private Path warehouseDir; - private final ScheduledExecutorService fileSyncExecutor = Executors.newSingleThreadScheduledExecutor(); private OAuth2AuthorizationServer oAuth2AuthorizationServer; @Parameters(name = "{0}") @@ -110,14 +120,15 @@ public TestIcebergRESTCatalogGravitinoLlapLocalCliDriver(String name, File qfile } @Before - public void setup() throws IOException { + public void setup() throws Exception { Network dockerNetwork = Network.newNetwork(); - + startOAuth2AuthorizationServer(dockerNetwork); createWarehouseDir(); + ozoneS3 = new OzoneS3GatewayContainers(); + ozoneS3.start(dockerNetwork, S3_BUCKET); prepareGravitinoConfig(); startGravitinoContainer(dockerNetwork); - fileSyncExecutor.scheduleAtFixedRate(this::syncWarehouseDir, 0, 5, TimeUnit.SECONDS); String host = gravitinoContainer.getHost(); Integer port = gravitinoContainer.getMappedPort(GRAVITINO_HTTP_PORT); @@ -137,63 +148,123 @@ public void setup() throws IOException { conf.set(restCatalogPrefix + "rest.auth.type", "oauth2"); conf.set(restCatalogPrefix + "oauth2-server-uri", oAuth2AuthorizationServer.getTokenEndpoint()); conf.set(restCatalogPrefix + "credential", oAuth2AuthorizationServer.getClientCredential()); + conf.set( + restCatalogPrefix + IcebergCatalogProperties.REST_ACCESS_DELEGATION_HEADER_PROPERTY, + RestAccessDelegationMode.VENDED_CREDENTIALS.modeName()); + + applyHostS3FilesystemSettings(conf); + applyIcebergS3ClientEndpointOverride(conf, restCatalogPrefix); } @After - public void teardown() throws IOException { + public void teardown() throws Exception { if (gravitinoContainer != null) { gravitinoContainer.stop(); } - + + if (ozoneS3 != null) { + ozoneS3.stop(); + } + if (oAuth2AuthorizationServer != null) { oAuth2AuthorizationServer.stop(); } - fileSyncExecutor.shutdownNow(); - FileUtils.deleteDirectory(warehouseDir.toFile()); + if (warehouseDir != null) { + FileUtils.deleteDirectory(warehouseDir.toFile()); + } + } + + /** + * Puts host-reachable Iceberg S3 client settings on the HS2 session. + * + *

Gravitino runs in Docker and vends storage credentials whose endpoint is + * {@link OzoneS3GatewayContainers#S3_DOCKER_ENDPOINT} (reachable inside the compose network only). + * Hive runs on the host, so this sets {@code iceberg.catalog.<catalog>.s3.endpoint} to the + * published Ozone S3G host/port, plus path-style and region, matching what operators configure in + * {@code hive-site.xml} in a real deployment. + * + *

Those keys are the source for {@link org.apache.iceberg.mr.hive.IcebergVendedCredentialUtil}: + * at plan time and on tasks it merges them over the vended endpoint in credentials and job conf. This + * method does not replace that util — it seeds session conf so the util has a host endpoint to apply. + */ + private void applyIcebergS3ClientEndpointOverride(Configuration conf, String restCatalogPrefix) { + String host = ozoneS3.getHost(); + int port = ozoneS3.getMappedPort(); + @SuppressWarnings("HttpUrlsUsage") + String icebergS3Endpoint = String.format("http://%s:%d", host, port); + conf.set(restCatalogPrefix + S3FileIOProperties.ENDPOINT, icebergS3Endpoint); + conf.set(restCatalogPrefix + S3FileIOProperties.PATH_STYLE_ACCESS, "true"); + conf.set(restCatalogPrefix + AwsClientProperties.CLIENT_REGION, "us-east-1"); + } + + /** + * Wires Hadoop to use S3A for {@code s3://} on the host-visible Ozone S3G endpoint. + * + *

What: sets {@code fs.s3}/{@code fs.s3a} implementation classes, per-bucket endpoint and path-style + * ({@code fs.s3a.bucket.<bucket>.*}), and disables SSL for this local Ozone S3G test. + * + *

Why: Hive and Tez use Hadoop {@code FileSystem} for many {@code s3://} paths, not only Iceberg + * {@code S3FileIO}. {@link org.apache.iceberg.mr.hive.IcebergVendedCredentialUtil} propagates vended keys and + * overlapping S3A settings onto job conf at plan time; it does not register the S3A filesystem or seed + * session defaults. Session wiring is still needed for HS2 and for paths that read session conf before job + * properties exist. + * + *

Access keys are intentionally omitted here; they are vended per query and copied into job secrets by the util. + */ + private void applyHostS3FilesystemSettings(Configuration conf) { + String s3Host = ozoneS3.getHost(); + int s3Port = ozoneS3.getMappedPort(); + @SuppressWarnings("HttpUrlsUsage") + String endpoint = String.format("http://%s:%d", s3Host, s3Port); + conf.set("fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem"); + conf.set("fs.AbstractFileSystem.s3.impl", "org.apache.hadoop.fs.s3a.S3A"); + conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem"); + String bucketPrefix = "fs.s3a.bucket." + S3_BUCKET + "."; + conf.set(bucketPrefix + "endpoint", endpoint); + conf.setBoolean(bucketPrefix + "path.style.access", true); + conf.setBoolean(bucketPrefix + "connection.ssl.enabled", false); } /** - * Starts a Gravitino container with the Iceberg REST server configured for testing. + * Starts a Gravitino container with the Iceberg REST server configured for this test. * *

This method configures the container to: *

* - *

Note: The {@code @SuppressWarnings("resource")} annotation is applied because - * IntelliJ and some compilers flag {@link org.testcontainers.containers.GenericContainer} - * as a resource that should be managed with try-with-resources. In this test setup, - * the container lifecycle is managed explicitly: it is started here and stopped in - * {@code @After} (via {@code gravitinoContainer.stop()}). Using try-with-resources - * would not work in this context, since the container must remain running across - * multiple test methods rather than being confined to a single block scope.

+ *

Note: the {@code @SuppressWarnings("resource")} annotation is applied because IntelliJ and some compilers + * flag {@link GenericContainer} as a resource that should be used with try-with-resources. Here the container + * lifecycle is explicit: it is started in this method and stopped in {@link #teardown()} via + * {@code gravitinoContainer.stop()}.

*/ @SuppressWarnings("resource") private void startGravitinoContainer(Network dockerNetwork) { gravitinoContainer = new GenericContainer<>(GRAVITINO_IMAGE) .withExposedPorts(GRAVITINO_HTTP_PORT) - // Update entrypoint to create the warehouse directory before starting the server + // Bootstrap dir for the server script; warehouse is s3:// on Ozone S3G (see template) .withCreateContainerCmdModifier(cmd -> cmd.withEntrypoint("bash", "-c", - String.format("mkdir -p %s && exec %s", warehouseDir.toString(), GRAVITINO_STARTUP_SCRIPT))) - // Mount gravitino configuration file + "mkdir -p /tmp/gravitino-bootstrap && exec " + GRAVITINO_STARTUP_SCRIPT)) + // Mount Gravitino configuration file (rendered under warehouseDir on the host) .withCopyFileToContainer( - MountableFile.forHostPath(Paths.get(warehouseDir.toString(), GRAVITINO_CONF_FILE_TEMPLATE)), - GRAVITINO_CONF_FILE - ) + MountableFile.forHostPath(Paths.get(warehouseDir.toString(), GRAVITINO_S3_CONF_TEMPLATE)), + GRAVITINO_CONF_FILE) // Mount the H2 driver JAR into the server's lib directory .withCopyFileToContainer( MountableFile.forHostPath( - Paths.get("target", "test-dependencies", "h2-driver.jar").toAbsolutePath() - ), - GRAVITINO_H2_LIB - ) - // Use the same Docker network as the OAuth2 server so they can communicate + Paths.get("target", "test-dependencies", "h2-driver.jar").toAbsolutePath()), + GRAVITINO_H2_LIB) + // Same Docker network as OAuth2 and Ozone S3G (Gravitino uses http://s3.ozone:9878 in config) .withNetwork(dockerNetwork) // Wait for the server to be fully started .waitingFor( @@ -201,123 +272,71 @@ private void startGravitinoContainer(Network dockerNetwork) { .withStrategy(Wait.forLogMessage(".*GravitinoIcebergRESTServer is running.*\\n", 1) .withStartupTimeout(Duration.ofMinutes(GRAVITINO_STARTUP_TIMEOUT_MINUTES))) .withStrategy(Wait.forListeningPort() - .withStartupTimeout(Duration.ofMinutes(GRAVITINO_STARTUP_TIMEOUT_MINUTES))) - ) - .withLogConsumer(new Slf4jLogConsumer(LoggerFactory - .getLogger(TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.class))); + .withStartupTimeout(Duration.ofMinutes(GRAVITINO_STARTUP_TIMEOUT_MINUTES)))) + .withLogConsumer(new Slf4jLogConsumer(LOG)); gravitinoContainer.start(); } - /** - * Starts a background daemon that continuously synchronizes the Iceberg warehouse - * directory from the running Gravitino container to the host file system. - * - *

In CI environments, Testcontainers' {@code .withFileSystemBind()} cannot reliably - * bind the same host path to the same path inside the container, especially when - * using remote Docker hosts or Docker-in-Docker setups. This causes the container's - * writes (e.g., Iceberg metadata files like {@code .metadata.json}) to be invisible - * on the host.

- * - *

This method works around that limitation by repeatedly copying new files from - * the container's warehouse directory to the corresponding host directory. Existing - * files on the host are preserved, and only files that do not yet exist are copied. - * The sync runs every 1 second while the container is running.

- * - *

Each archive copy from the container is extracted using a {@link TarArchiveInputStream}, - * and directories are created as needed. Files that already exist on the host are skipped - * to avoid overwriting container data.

- */ - private void syncWarehouseDir() { - if (gravitinoContainer.isRunning()) { - try (CopyArchiveFromContainerCmd copyArchiveFromContainerCmd = - gravitinoContainer - .getDockerClient() - .copyArchiveFromContainerCmd(gravitinoContainer.getContainerId(), warehouseDir.toString()); - InputStream tarStream = copyArchiveFromContainerCmd.exec(); - TarArchiveInputStream tis = new TarArchiveInputStream(tarStream)) { - - TarArchiveEntry entry; - while ((entry = tis.getNextEntry()) != null) { - // Skip directories because we only want to copy metadata files from the container. - if (entry.isDirectory()) { - continue; - } - - /* - * Tar entry names include a container-specific top-level folder, e.g.: - * iceberg-test-1759245909247/iceberg_warehouse/ice_rest/.../metadata.json - * - * Strip the first part so the relative path inside the warehouse is preserved - * when mapping to the host warehouseDir. - */ - - String[] parts = entry.getName().split("/", 2); - if (parts.length < 2) { - continue; // defensive guard - } - - Path relativePath = Paths.get(parts[1]); - Path outputPath = warehouseDir.resolve(relativePath); - - // Skip if already present on host to avoid overwriting - if (Files.exists(outputPath)) { - continue; - } - - Files.createDirectories(outputPath.getParent()); - Files.copy(tis, outputPath); - } - - } catch (Exception e) { - LOG.error("Warehouse folder sync failed: {}", e.getMessage()); - } - } - } - + /** Keycloak-backed OAuth2 used by Gravitino REST authentication and by the Hive REST client. */ private void startOAuth2AuthorizationServer(Network dockerNetwork) { oAuth2AuthorizationServer = new OAuth2AuthorizationServer(dockerNetwork, false); oAuth2AuthorizationServer.start(); } + /** + * Host directory used to write the rendered Gravitino config (see {@link #prepareGravitinoConfig}) and as the + * source path for {@code .withCopyFileToContainer} in {@link #startGravitinoContainer}. This is not the Iceberg + * warehouse root; the warehouse is {@code s3://}{@link #S3_BUCKET}{@code /...} on Ozone S3G. + */ private void createWarehouseDir() { try { warehouseDir = Paths.get("/tmp", "iceberg-test-" + System.currentTimeMillis()).toAbsolutePath(); Files.createDirectories(warehouseDir); } catch (Exception e) { - throw new RuntimeException("Failed to create the Iceberg warehouse directory", e); + throw new RuntimeException("Failed to create temp directory for Gravitino config staging", e); } } + /** + * Reads {@link #GRAVITINO_S3_CONF_TEMPLATE} from the classpath, substitutes bucket / Ozone S3 / OAuth placeholders, + * and writes the result under {@link #warehouseDir} for copying into the Gravitino container. + */ private void prepareGravitinoConfig() throws IOException { String content; try (InputStream in = TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.class.getClassLoader() - .getResourceAsStream(GRAVITINO_CONF_FILE_TEMPLATE)) { + .getResourceAsStream(GRAVITINO_S3_CONF_TEMPLATE)) { if (in == null) { - throw new IOException("Resource not found: " + GRAVITINO_CONF_FILE_TEMPLATE); + throw new IOException("Resource not found: " + GRAVITINO_S3_CONF_TEMPLATE); } content = new String(in.readAllBytes(), StandardCharsets.UTF_8); } String updatedContent = content - .replace("/WAREHOUSE_DIR", warehouseDir.toString()) + .replace("S3_BUCKET", S3_BUCKET) + .replace("S3_ACCESS_KEY", OzoneS3GatewayContainers.ACCESS_KEY) + .replace("S3_SECRET_KEY", OzoneS3GatewayContainers.SECRET_KEY) .replace("OAUTH2_SERVER_URI", oAuth2AuthorizationServer.getIssuer()) .replace("OAUTH2_JWKS_URI", getJwksUri()) .replace("OAUTH2_CLIENT_ID", OAUTH2_SERVER_ICEBERG_CLIENT_ID) .replace("OAUTH2_CLIENT_SECRET", OAUTH2_SERVER_ICEBERG_CLIENT_SECRET) .replace("HTTP_PORT", String.valueOf(GRAVITINO_HTTP_PORT)); - Path configFile = warehouseDir.resolve(GRAVITINO_CONF_FILE_TEMPLATE); + Path configFile = warehouseDir.resolve(GRAVITINO_S3_CONF_TEMPLATE); Files.writeString(configFile, updatedContent); } + /** + * JWKS URL reachable from inside the Gravitino container: host/port in the issuer are rewritten to the + * Keycloak container hostname and its internal HTTP port. + */ private String getJwksUri() { String reachableHost = oAuth2AuthorizationServer.getKeycloackContainerDockerInternalHostName(); int internalPort = 8080; // Keycloak container's internal port return oAuth2AuthorizationServer.getIssuer() .replace("localhost", reachableHost) .replace("127.0.0.1", reachableHost) - // replace issuer's mapped port with keyclock container's internal port + // Replace issuer's mapped host port with Keycloak's internal port on the Docker network .replaceFirst(":[0-9]+", ":" + internalPort); } diff --git a/itests/qtest-iceberg/src/test/resources/gravitino-s3-vended-oauth-template.conf b/itests/qtest-iceberg/src/test/resources/gravitino-s3-vended-oauth-template.conf new file mode 100644 index 000000000000..9f78ec628daf --- /dev/null +++ b/itests/qtest-iceberg/src/test/resources/gravitino-s3-vended-oauth-template.conf @@ -0,0 +1,37 @@ +gravitino.iceberg-rest.httpPort = HTTP_PORT + +# --- Iceberg REST Catalog Backend (JDBC for catalog metadata) --- +gravitino.iceberg-rest.catalog-backend = jdbc +gravitino.iceberg-rest.uri = jdbc:h2:file:/tmp/gravitino_h2_db;AUTO_SERVER=TRUE +gravitino.iceberg-rest.jdbc-driver = org.h2.Driver +gravitino.iceberg-rest.jdbc-user = sa +gravitino.iceberg-rest.jdbc-password = "" +gravitino.iceberg-rest.jdbc-initialize = true + +# --- Warehouse on Ozone S3G + credential vending (s3-secret-key) --- +gravitino.iceberg-rest.warehouse = s3://S3_BUCKET/warehouse +gravitino.iceberg-rest.io-impl = org.apache.iceberg.aws.s3.S3FileIO +gravitino.iceberg-rest.credential-providers = s3-secret-key +gravitino.iceberg-rest.s3-access-key-id = S3_ACCESS_KEY +gravitino.iceberg-rest.s3-secret-access-key = S3_SECRET_KEY +# Docker network only (Gravitino container -> Ozone S3G). Host-side Hive/Tez/Iceberg use +# iceberg.catalog.ice01.s3.endpoint (no static keys); credentials are vended per table load. +gravitino.iceberg-rest.s3-endpoint = http://s3.ozone:9878 +gravitino.iceberg-rest.s3-path-style-access = true +gravitino.iceberg-rest.s3-region = us-east-1 + +# --- OAuth2 Authentication (catalog HTTP API) --- +gravitino.authenticators = oauth + +gravitino.authenticator.oauth.serverUri = OAUTH2_SERVER_URI +gravitino.authenticator.oauth.tokenPath = /protocol/openid-connect/token +gravitino.authenticator.oauth.clientId = OAUTH2_CLIENT_ID +gravitino.authenticator.oauth.scope = openid catalog +gravitino.authenticator.oauth.clientSecret = OAUTH2_CLIENT_SECRET + +gravitino.authenticator.oauth.tokenValidatorClass = org.apache.gravitino.server.authentication.JwksTokenValidator +gravitino.authenticator.oauth.jwksUri = OAUTH2_JWKS_URI/protocol/openid-connect/certs +gravitino.authenticator.oauth.provider = default +gravitino.authenticator.oauth.principalFields = sub +gravitino.authenticator.oauth.allowSkewSecs = 60 +gravitino.authenticator.oauth.serviceAudience = hive-metastore diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java index 5d4bfd3dffb9..8b02ec22ea5c 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java @@ -375,6 +375,13 @@ private List getNextSplits() throws Exception { Class formatter = currDesc.getInputFileFormatClass(); Utilities.copyTableJobPropertiesToConf(currDesc.getTableDesc(), job); + // A fetch task runs in-process without a DAG, so storage-handler secrets (HIVE-20651) are + // never restored from the job Credentials as on a task. Surface them into this process-local + // conf (never serialized or shown in EXPLAIN) so in-process readers can access storage. + Map jobSecrets = currDesc.getTableDesc().getJobSecrets(); + if (jobSecrets != null) { + jobSecrets.forEach(job::set); + } InputFormat inputFormat = getInputFormatFromCache(formatter, job); if(inputFormat instanceof HiveSequenceFileInputFormat) { // input format could be cached, in which case we need to reset the list of files to fetch diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java index 5d3c2dccf5cf..a3d48f36a314 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java @@ -942,6 +942,11 @@ protected void createBucketForFileIdx(FSPaths fsp, int filesIdx) updateDPCounters(fsp, filesIdx); + try { + Utilities.copyJobSecretToTableProperties(conf.getTableInfo()); + } catch (IOException e) { + throw new HiveException(e); + } Utilities.copyTableJobPropertiesToConf(conf.getTableInfo(), jc); // only create bucket files only if no dynamic partitions, // buckets of dynamic partitions will be created for each newly created partition @@ -1669,6 +1674,11 @@ public void checkOutputSpecs(FileSystem ignored, JobConf job) throws IOException private void createHiveOutputFormat(JobConf job) throws HiveException { if (hiveOutputFormat == null) { + try { + Utilities.copyJobSecretToTableProperties(conf.getTableInfo()); + } catch (IOException e) { + throw new HiveException(e); + } Utilities.copyTableJobPropertiesToConf(conf.getTableInfo(), job); } try { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java index f58fafc85566..8a72a22b97f5 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java @@ -2517,7 +2517,7 @@ public static void copyJobSecretToTableProperties(TableDesc tbl) throws IOExcept String[] comps = keyString.split(TableDesc.SECRET_DELIMIT); String tblName = comps[1]; String keyName = comps[2]; - if (tbl.getTableName().equalsIgnoreCase(tblName)) { + if (tblName.equalsIgnoreCase(tbl.getTableName())) { tbl.getProperties().put(keyName, new String(credentials.getSecretKey(key))); } } @@ -3899,8 +3899,10 @@ private static void createTmpDirs(Configuration conf, if (op instanceof FileSinkOperator) { FileSinkDesc fdesc = ((FileSinkOperator) op).getConf(); - if (fdesc.isMmTable() || fdesc.isDirectInsert()) { - // No need to create for MM tables, or ACID insert + if (fdesc.isMmTable() || fdesc.isDirectInsert() || + (fdesc.getTableInfo() != null && fdesc.getTableInfo().isNonNative())) { + // No need to create for MM tables or ACID insert; storage-handler tables manage + // their own output locations and commit continue; } Path tempDir = fdesc.getDirName();