From 254c75717bd483969f3a02e2c9f8325db2585a02 Mon Sep 17 00:00:00 2001 From: Dmitriy Fingerman Date: Tue, 28 Jul 2026 15:01:03 -0400 Subject: [PATCH] HIVE-29777: Iceberg: Add GCS, ADLS, and OSS Hadoop mapping for REST vended credentials --- .../org/apache/hadoop/hive/conf/HiveConf.java | 8 + .../mr/hive/IcebergVendedCredentialUtil.java | 124 ++----- .../AdlsVendedCredentialHadoopMapper.java | 104 ++++++ .../GcsVendedCredentialHadoopMapper.java | 60 +++ .../OssVendedCredentialHadoopMapper.java | 56 +++ .../S3VendedCredentialHadoopMapper.java | 60 +++ .../vended/VendedCredentialHadoopMapper.java | 59 +++ .../vended/VendedCredentialHadoopMappers.java | 57 +++ .../vended/VendedCredentialPrefixUtil.java | 72 ++++ .../hive/vended/VendedCredentialSupport.java | 149 ++++++++ .../hive/TestIcebergVendedCredentialUtil.java | 344 +++++++++++++++++- 11 files changed, 1002 insertions(+), 91 deletions(-) create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/AdlsVendedCredentialHadoopMapper.java create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/GcsVendedCredentialHadoopMapper.java create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/OssVendedCredentialHadoopMapper.java create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/S3VendedCredentialHadoopMapper.java create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialHadoopMapper.java create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialHadoopMappers.java create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialPrefixUtil.java create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialSupport.java 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 54516007d32a..7032166a26cd 100644 --- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java +++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java @@ -5536,6 +5536,14 @@ public static enum ConfVars { + ",s3.access-key-id" + ",s3.secret-access-key" + ",s3.session-token" + // Iceberg FileIO vended credential keys (GCS, ADLS, OSS) + + ",gcs.oauth2.token" + + ",adls.auth.shared-key.account.key" + + ",adls.sas-token." + + ",adls.token" + + ",client.access-key-id" + + ",client.access-key-secret" + + ",client.security-token" + ",dfs.adls.oauth2.credential" + ",fs.adl.oauth2.credential" + ",fs.azure.account.oauth2.client.secret" diff --git a/iceberg/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 index 95bfd2f9623d..c9bec0487233 100644 --- 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 @@ -38,8 +38,6 @@ import org.apache.iceberg.StaticTableOperations; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; -import org.apache.iceberg.aws.AwsClientProperties; -import org.apache.iceberg.aws.s3.S3FileIOProperties; import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.hive.IcebergCatalogProperties; import org.apache.iceberg.hive.rest.catalog.RestCatalogAccessDelegation; @@ -47,6 +45,8 @@ import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsStorageCredentials; import org.apache.iceberg.mr.InputFormatConfig; +import org.apache.iceberg.mr.hive.vended.VendedCredentialHadoopMapper; +import org.apache.iceberg.mr.hive.vended.VendedCredentialSupport; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.util.SerializationUtil; @@ -102,17 +102,23 @@ public static void propagateToJob(Table table, String catalogName, MapDerives the S3 bucket from {@link StorageCredential#prefix()} and delegates to + *

Derives storage scope from {@link StorageCredential#prefix()} and delegates to * {@link #addCredentialEntry} for every entry in {@link StorageCredential#config()}, which maps - * Iceberg keys to catalog-level and per-bucket S3A job properties or secrets. + * Iceberg keys to catalog-level and provider-specific Hadoop job properties or secrets. */ private static void addCredentialEntries(String catalogName, StorageCredential credential, Map jobProperties, Map jobSecrets, Configuration conf) { - String bucket = bucketFromPrefix(credential.prefix()); + VendedCredentialHadoopMapper mapper = VendedCredentialSupport.mapperFor(credential); + String scope = mapper != null ? mapper.scopeFromPrefix(credential.prefix()) : + VendedCredentialSupport.scopeFromPrefix(credential.prefix()); for (Map.Entry entry : credential.config().entrySet()) { addCredentialEntry( - catalogName, bucket, entry.getKey(), entry.getValue(), jobProperties, jobSecrets, conf); + catalogName, scope, mapper, entry.getKey(), entry.getValue(), jobProperties, jobSecrets, conf); + } + if (jobProperties != null) { + VendedCredentialSupport.additionalNonSecretHadoopProperties(mapper, scope, credential.config()) + .forEach(jobProperties::putIfAbsent); } } @@ -124,7 +130,8 @@ private static void addCredentialEntries(String catalogName, StorageCredential c * secret keys (access key, secret key, session token) to {@code jobSecrets}. Either map may be * {@code null} when {@link #propagateToJob} is called for only properties or only secrets. */ - private static void addCredentialEntry(String catalogName, String bucket, String icebergKey, String value, + private static void addCredentialEntry(String catalogName, String scope, + VendedCredentialHadoopMapper mapper, String icebergKey, String value, Map jobProperties, Map jobSecrets, Configuration conf) { if (StringUtils.isBlank(value)) { @@ -133,22 +140,22 @@ private static void addCredentialEntry(String catalogName, String bucket, String String resolvedValue = resolveCredentialValue(catalogName, icebergKey, value, conf); if (jobProperties != null && !isSecretKey(icebergKey, conf)) { - addNonSecretCredentialEntry(catalogName, bucket, icebergKey, resolvedValue, jobProperties); + addNonSecretCredentialEntry(catalogName, scope, mapper, icebergKey, resolvedValue, jobProperties); } if (jobSecrets != null && isSecretKey(icebergKey, conf)) { - addSecretCredentialEntry(bucket, icebergKey, resolvedValue, jobSecrets); + addSecretCredentialEntry(scope, mapper, icebergKey, resolvedValue, jobSecrets); } } /** - * Adds one non-secret vended value to {@code jobProperties} for Iceberg and Hadoop S3A. + * Adds one non-secret vended value to {@code jobProperties} for Iceberg and Hadoop. * *

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. + * When a Hadoop mapper is present, also writes the matching provider-specific Hadoop key. */ - private static void addNonSecretCredentialEntry(String catalogName, String bucket, String icebergKey, String value, + private static void addNonSecretCredentialEntry(String catalogName, String scope, + VendedCredentialHadoopMapper mapper, String icebergKey, String value, Map jobProperties) { if (catalogName != null) { @@ -157,23 +164,21 @@ private static void addNonSecretCredentialEntry(String catalogName, String bucke jobProperties.putIfAbsent(catalogConfigKey, value); } - if (bucket != null) { - String s3aKey = toS3aBucketProperty(bucket, icebergKey); - if (s3aKey != null) { - jobProperties.putIfAbsent(s3aKey, value); + if (mapper != null && scope != null) { + String hadoopKey = VendedCredentialSupport.toHadoopProperty(mapper, scope, icebergKey); + if (hadoopKey != null) { + jobProperties.putIfAbsent(hadoopKey, value); } } } - /** Writes Hadoop S3A per-bucket keys only; Iceberg secrets are carried in the serialized blob. - * First table wins on a shared bucket, matching the non-secret entries, so an endpoint and its - * key always come from the same credential. */ - private static void addSecretCredentialEntry(String bucket, String icebergKey, String value, - Map jobSecrets) { - if (bucket != null) { - String s3aSecretKey = toS3aBucketProperty(bucket, icebergKey); - if (s3aSecretKey != null) { - jobSecrets.putIfAbsent(s3aSecretKey, value); + /** Writes Hadoop keys only; Iceberg secrets are carried in the serialized blob. */ + private static void addSecretCredentialEntry(String scope, VendedCredentialHadoopMapper mapper, + String icebergKey, String value, Map jobSecrets) { + if (mapper != null && scope != null) { + String hadoopSecretKey = VendedCredentialSupport.toHadoopProperty(mapper, scope, icebergKey); + if (hadoopSecretKey != null) { + jobSecrets.putIfAbsent(hadoopSecretKey, value); } } } @@ -386,50 +391,16 @@ static List extractCredentials(Table table) { 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 + "/"; + return VendedCredentialSupport.credentialsFromFileIoProperties(table, io); } /** - * REST catalogs vend credentials together with S3 connectivity settings such as endpoint + * REST catalogs vend credentials together with storage connectivity settings such as endpoint * and path-style access. These settings reflect the catalog's network view and may reference * hosts that are not reachable from Hive (for example, an internal {@code s3.ozone:9878} * hostname). * - * Catalog S3 properties configured in the Hive session (for example, + * Catalog properties configured in the Hive session (for example, * {@code iceberg.catalog.ice01.s3.endpoint}) override the corresponding vended connectivity * settings so the driver and executors use reachable endpoints. Vended credentials are * preserved; only non-secret connectivity properties are overridden. @@ -470,31 +441,4 @@ private static String resolveCredentialValue( conf.get(IcebergCatalogProperties.catalogPropertyConfigKey(catalogName, icebergKey)); return StringUtils.isNotBlank(override) ? override : vendedValue; } - - /** Authority (bucket) of a storage prefix such as {@code s3://bucket/path}, any scheme. - * Plain string parse: {@code URI.getHost()} rejects legal bucket names with underscores. */ - @SuppressWarnings("java:S1075") // storage URI path separator, not a filesystem path - private static String bucketFromPrefix(String prefix) { - int schemeEnd = prefix == null ? -1 : prefix.indexOf("://"); - if (schemeEnd < 0) { - return null; - } - String withoutScheme = prefix.substring(schemeEnd + 3); - int slash = withoutScheme.indexOf('/'); - String bucket = slash >= 0 ? withoutScheme.substring(0, slash) : withoutScheme; - return StringUtils.defaultIfBlank(bucket, null); - } - - private static String toS3aBucketProperty(String bucket, String icebergKey) { - String bucketPrefix = "fs.s3a.bucket." + bucket + "."; - return switch (icebergKey) { - case S3FileIOProperties.ACCESS_KEY_ID -> bucketPrefix + "access.key"; - case S3FileIOProperties.SECRET_ACCESS_KEY -> bucketPrefix + "secret.key"; - case S3FileIOProperties.SESSION_TOKEN -> bucketPrefix + "session.token"; - case S3FileIOProperties.ENDPOINT -> bucketPrefix + "endpoint"; - case S3FileIOProperties.PATH_STYLE_ACCESS -> bucketPrefix + "path.style.access"; - case AwsClientProperties.CLIENT_REGION -> bucketPrefix + "endpoint.region"; - default -> null; - }; - } } diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/AdlsVendedCredentialHadoopMapper.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/AdlsVendedCredentialHadoopMapper.java new file mode 100644 index 000000000000..32fbd09d5150 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/AdlsVendedCredentialHadoopMapper.java @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive.vended; + +import java.util.LinkedHashMap; +import java.util.Map; +import org.apache.commons.lang3.StringUtils; + +/** Maps Iceberg ADLS FileIO properties to Hadoop ABFS connector keys. */ +enum AdlsVendedCredentialHadoopMapper implements VendedCredentialHadoopMapper { + INSTANCE; + + static final String ADLS_SAS_TOKEN_PREFIX = "adls.sas-token."; + static final String ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX = "adls.sas-token-expires-at-ms."; + static final String ADLS_SHARED_KEY_ACCOUNT_KEY = "adls.auth.shared-key.account.key"; + + private static final String DFS_CORE_WINDOWS_NET_SUFFIX = ".dfs.core.windows.net"; + + @Override + public boolean supportsPrefix(String prefix) { + String scheme = VendedCredentialPrefixUtil.schemeFromPrefix(prefix); + return "abfs".equals(scheme) || "abfss".equals(scheme) || "wasb".equals(scheme) || "wasbs".equals(scheme); + } + + @Override + public boolean supportsConfigKey(String icebergKey) { + return icebergKey.startsWith("adls."); + } + + @Override + public String scopeFromPrefix(String prefix) { + return VendedCredentialPrefixUtil.scopeFromPrefix(prefix); + } + + @Override + public String toHadoopProperty(String scope, String icebergKey) { + // SAS token: the Iceberg key carries the account (adls.sas-token.), so the + // account-scoped ABFS fixed-token key is derived directly from it. ABFS also requires + // fs.azure.account.auth.type....=SAS, emitted by additionalNonSecretHadoopProperties. + if (icebergKey.startsWith(ADLS_SAS_TOKEN_PREFIX) && + !icebergKey.startsWith(ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX)) { + String account = icebergKey.substring(ADLS_SAS_TOKEN_PREFIX.length()); + return "fs.azure.sas.fixed.token." + account + DFS_CORE_WINDOWS_NET_SUFFIX; + } + // Shared-key: the account name is not in this entry, so it is taken from the storage prefix. + // SharedKey is the ABFS default auth type, so no auth-type companion property is needed. + if (icebergKey.equals(ADLS_SHARED_KEY_ACCOUNT_KEY)) { + String account = accountFromScope(scope); + if (account != null) { + return "fs.azure.account.key." + account + DFS_CORE_WINDOWS_NET_SUFFIX; + } + return null; + } + // adls.token (OAuth bearer) has no plain ABFS config key — ABFS OAuth needs a provider type + // (client id/secret/endpoint or refresh token). It rides the Iceberg StorageCredential blob. + return null; + } + + @Override + public Map additionalNonSecretHadoopProperties(String scope, Map config) { + Map extra = new LinkedHashMap<>(); + for (String key : config.keySet()) { + if (key.startsWith(ADLS_SAS_TOKEN_PREFIX) && !key.startsWith(ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX)) { + String account = key.substring(ADLS_SAS_TOKEN_PREFIX.length()); + extra.put("fs.azure.account.auth.type." + account + DFS_CORE_WINDOWS_NET_SUFFIX, "SAS"); + } + } + return extra; + } + + /** + * Storage account from a credential prefix scope such as + * {@code container@account.dfs.core.windows.net} (or {@code account.dfs.core.windows.net}), + * with the {@code .dfs.core.windows.net} suffix stripped. + */ + private static String accountFromScope(String scope) { + if (StringUtils.isBlank(scope)) { + return null; + } + int at = scope.indexOf('@'); + String host = at < 0 ? scope : scope.substring(at + 1); + if (host.endsWith(DFS_CORE_WINDOWS_NET_SUFFIX)) { + host = host.substring(0, host.length() - DFS_CORE_WINDOWS_NET_SUFFIX.length()); + } + return StringUtils.defaultIfBlank(host, null); + } +} diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/GcsVendedCredentialHadoopMapper.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/GcsVendedCredentialHadoopMapper.java new file mode 100644 index 000000000000..97028d1fe5a9 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/GcsVendedCredentialHadoopMapper.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive.vended; + +/** Maps Iceberg GCS FileIO properties to Hadoop Google Cloud Storage connector keys. */ +enum GcsVendedCredentialHadoopMapper implements VendedCredentialHadoopMapper { + INSTANCE; + + static final String GCS_OAUTH2_TOKEN = "gcs.oauth2.token"; + static final String GCS_OAUTH2_TOKEN_EXPIRES_AT = "gcs.oauth2.token-expires-at"; + static final String GCS_PROJECT_ID = "gcs.project-id"; + static final String GCS_SERVICE_HOST = "gcs.service.host"; + + @Override + public boolean supportsPrefix(String prefix) { + String scheme = VendedCredentialPrefixUtil.schemeFromPrefix(prefix); + return "gs".equals(scheme) || "gcs".equals(scheme); + } + + @Override + public boolean supportsConfigKey(String icebergKey) { + return icebergKey.startsWith("gcs."); + } + + @Override + public String scopeFromPrefix(String prefix) { + return VendedCredentialPrefixUtil.scopeFromPrefix(prefix); + } + + @Override + public String toHadoopProperty(String bucket, String icebergKey) { + // The Google Cloud Storage Hadoop connector has no plain config key to inject a raw OAuth + // token; a vended token requires fs.gs.auth.type=ACCESS_TOKEN_PROVIDER plus an + // AccessTokenProvider implementation class. Until that is wired, gcs.oauth2.token and its + // expiry travel only in the serialized Iceberg StorageCredential blob and are consumed by + // Iceberg GCSFileIO. Only connectivity/config that maps to real connector keys is emitted. + return switch (icebergKey) { + case GCS_PROJECT_ID -> "fs.gs.project.id"; + case GCS_SERVICE_HOST -> "fs.gs.storage.root.url"; + default -> null; + }; + } +} diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/OssVendedCredentialHadoopMapper.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/OssVendedCredentialHadoopMapper.java new file mode 100644 index 000000000000..80dad676df4e --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/OssVendedCredentialHadoopMapper.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive.vended; + +/** Maps Iceberg OSS FileIO properties to Hadoop Aliyun OSS filesystem keys. */ +enum OssVendedCredentialHadoopMapper implements VendedCredentialHadoopMapper { + INSTANCE; + + static final String CLIENT_ACCESS_KEY_ID = "client.access-key-id"; + static final String CLIENT_ACCESS_KEY_SECRET = "client.access-key-secret"; + static final String CLIENT_SECURITY_TOKEN = "client.security-token"; + static final String OSS_ENDPOINT = "oss.endpoint"; + + @Override + public boolean supportsPrefix(String prefix) { + return "oss".equals(VendedCredentialPrefixUtil.schemeFromPrefix(prefix)); + } + + @Override + public boolean supportsConfigKey(String icebergKey) { + return icebergKey.startsWith("client.") || icebergKey.startsWith("oss."); + } + + @Override + public String scopeFromPrefix(String prefix) { + return VendedCredentialPrefixUtil.scopeFromPrefix(prefix); + } + + @Override + public String toHadoopProperty(String bucket, String icebergKey) { + return switch (icebergKey) { + case CLIENT_ACCESS_KEY_ID -> "fs.oss.accessKeyId"; + case CLIENT_ACCESS_KEY_SECRET -> "fs.oss.accessKeySecret"; + case CLIENT_SECURITY_TOKEN -> "fs.oss.securityToken"; + case OSS_ENDPOINT -> "fs.oss.endpoint"; + default -> null; + }; + } +} diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/S3VendedCredentialHadoopMapper.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/S3VendedCredentialHadoopMapper.java new file mode 100644 index 000000000000..6ed1420ca784 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/S3VendedCredentialHadoopMapper.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive.vended; + +import org.apache.iceberg.aws.AwsClientProperties; +import org.apache.iceberg.aws.s3.S3FileIOProperties; + +enum S3VendedCredentialHadoopMapper implements VendedCredentialHadoopMapper { + INSTANCE; + + @Override + public boolean supportsPrefix(String prefix) { + String scheme = VendedCredentialPrefixUtil.schemeFromPrefix(prefix); + return "s3".equals(scheme) || "s3a".equals(scheme) || "s3n".equals(scheme); + } + + @Override + public boolean supportsConfigKey(String icebergKey) { + return icebergKey.startsWith("s3.") || AwsClientProperties.CLIENT_REGION.equals(icebergKey); + } + + @Override + public String scopeFromPrefix(String prefix) { + return VendedCredentialPrefixUtil.scopeFromPrefix(prefix); + } + + @Override + public String toHadoopProperty(String bucket, String icebergKey) { + if (bucket == null) { + return null; + } + 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/vended/VendedCredentialHadoopMapper.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialHadoopMapper.java new file mode 100644 index 000000000000..11c05e184a09 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialHadoopMapper.java @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive.vended; + +import java.util.Collections; +import java.util.Map; +import org.apache.iceberg.io.StorageCredential; + +/** + * Maps Iceberg {@link StorageCredential} config entries to Hadoop + * FileSystem properties for Tez/LLAP paths that use {@code FileSystem.get()} instead of Iceberg + * FileIO alone. + */ +public interface VendedCredentialHadoopMapper { + + /** Returns true when {@code prefix} uses this provider's URI scheme (for example {@code s3://}). */ + boolean supportsPrefix(String prefix); + + /** Returns true when {@code icebergKey} belongs to this provider's Iceberg property namespace. */ + boolean supportsConfigKey(String icebergKey); + + /** + * Storage scope extracted from {@code prefix}: bucket name for object stores, or ADLS + * {@code container@account.dfs.core.windows.net} authority. + */ + String scopeFromPrefix(String prefix); + + /** + * Hadoop configuration key for {@code icebergKey}, or {@code null} when there is no Hadoop + * equivalent (Iceberg FileIO still receives the value via the serialized credentials blob). + */ + String toHadoopProperty(String scope, String icebergKey); + + /** + * Fixed, non-secret Hadoop properties (not tied to a single vended value) required to activate + * the mapped credentials — e.g. ABFS {@code fs.azure.account.auth.type....=SAS} so a + * fixed SAS token is actually used. Defaults to none. + */ + default Map additionalNonSecretHadoopProperties(String scope, Map config) { + return Collections.emptyMap(); + } +} diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialHadoopMappers.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialHadoopMappers.java new file mode 100644 index 000000000000..d113b82942c8 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialHadoopMappers.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive.vended; + +import java.util.List; +import org.apache.iceberg.io.StorageCredential; + +/** Selects the Hadoop mapper for a vended {@link StorageCredential}. */ +final class VendedCredentialHadoopMappers { + + private static final List MAPPERS = + List.of( + S3VendedCredentialHadoopMapper.INSTANCE, + GcsVendedCredentialHadoopMapper.INSTANCE, + AdlsVendedCredentialHadoopMapper.INSTANCE, + OssVendedCredentialHadoopMapper.INSTANCE); + + private VendedCredentialHadoopMappers() { + } + + static VendedCredentialHadoopMapper forCredential(StorageCredential credential) { + if (credential == null) { + return null; + } + String prefix = credential.prefix(); + for (VendedCredentialHadoopMapper mapper : MAPPERS) { + if (mapper.supportsPrefix(prefix)) { + return mapper; + } + } + for (String icebergKey : credential.config().keySet()) { + for (VendedCredentialHadoopMapper mapper : MAPPERS) { + if (mapper.supportsConfigKey(icebergKey)) { + return mapper; + } + } + } + return null; + } +} diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialPrefixUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialPrefixUtil.java new file mode 100644 index 000000000000..d6f7c7188432 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialPrefixUtil.java @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive.vended; + +import org.apache.commons.lang3.StringUtils; + +/** Helpers for parsing Iceberg {@link org.apache.iceberg.io.StorageCredential#prefix()} values. */ +final class VendedCredentialPrefixUtil { + + private VendedCredentialPrefixUtil() { + } + + static String schemeFromPrefix(String prefix) { + if (prefix == null) { + return null; + } + int schemeEnd = prefix.indexOf("://"); + if (schemeEnd <= 0) { + return null; + } + return prefix.substring(0, schemeEnd).toLowerCase(); + } + + /** + * Authority segment of a storage prefix ({@code s3://bucket/path} → {@code bucket}, + * {@code abfss://container@account.dfs.core.windows.net/path} → + * {@code container@account.dfs.core.windows.net}). + */ + @SuppressWarnings("java:S1075") + static String scopeFromPrefix(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 scope = slash >= 0 ? withoutScheme.substring(0, slash) : withoutScheme; + return StringUtils.defaultIfBlank(scope, null); + } + + /** Normalizes a table location to a credential prefix with trailing slash. */ + @SuppressWarnings("java:S1075") + static String storagePrefixFromLocation(String location) { + if (StringUtils.isBlank(location)) { + return ""; + } + int schemeEnd = location.indexOf("://"); + if (schemeEnd < 0) { + return location.endsWith("/") ? location : location + "/"; + } + int pathStart = location.indexOf('/', schemeEnd + 3); + String base = pathStart >= 0 ? location.substring(0, pathStart) : location; + return base.endsWith("/") ? base : base + "/"; + } +} diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialSupport.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialSupport.java new file mode 100644 index 000000000000..725ce1160bf4 --- /dev/null +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vended/VendedCredentialSupport.java @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive.vended; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.apache.commons.lang3.StringUtils; +import org.apache.iceberg.Table; +import org.apache.iceberg.aws.AwsClientProperties; +import org.apache.iceberg.aws.s3.S3FileIOProperties; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.StorageCredential; + +/** Public helpers for {@link org.apache.iceberg.mr.hive.IcebergVendedCredentialUtil}. */ +public final class VendedCredentialSupport { + + private VendedCredentialSupport() { + } + + public static String storagePrefixFromLocation(String location) { + return VendedCredentialPrefixUtil.storagePrefixFromLocation(location); + } + + public static String scopeFromPrefix(String prefix) { + return VendedCredentialPrefixUtil.scopeFromPrefix(prefix); + } + + public static VendedCredentialHadoopMapper mapperFor(StorageCredential credential) { + return VendedCredentialHadoopMappers.forCredential(credential); + } + + public static String toHadoopProperty( + VendedCredentialHadoopMapper mapper, String scope, String icebergKey) { + if (mapper == null || scope == null) { + return null; + } + return mapper.toHadoopProperty(scope, icebergKey); + } + + public static Map additionalNonSecretHadoopProperties( + VendedCredentialHadoopMapper mapper, String scope, Map config) { + if (mapper == null) { + return Map.of(); + } + return mapper.additionalNonSecretHadoopProperties(scope, config); + } + + public static List credentialsFromFileIoProperties(Table table, FileIO io) { + Map props = io.properties(); + if (props == null || props.isEmpty()) { + return List.of(); + } + String prefix = storagePrefixFromLocation(table.location()); + List s3 = credentialsFromS3Properties(prefix, props); + if (!s3.isEmpty()) { + return s3; + } + List gcs = credentialsFromGcsProperties(prefix, props); + if (!gcs.isEmpty()) { + return gcs; + } + List adls = credentialsFromAdlsProperties(prefix, props); + if (!adls.isEmpty()) { + return adls; + } + return credentialsFromOssProperties(prefix, props); + } + + private static List credentialsFromS3Properties( + String prefix, Map props) { + if (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(prefix, config)); + } + + private static List credentialsFromGcsProperties( + String prefix, Map props) { + if (StringUtils.isBlank(props.get(GcsVendedCredentialHadoopMapper.GCS_OAUTH2_TOKEN))) { + return List.of(); + } + Map config = new LinkedHashMap<>(); + putIfPresent(config, props, GcsVendedCredentialHadoopMapper.GCS_OAUTH2_TOKEN); + putIfPresent(config, props, GcsVendedCredentialHadoopMapper.GCS_OAUTH2_TOKEN_EXPIRES_AT); + putIfPresent(config, props, GcsVendedCredentialHadoopMapper.GCS_PROJECT_ID); + putIfPresent(config, props, GcsVendedCredentialHadoopMapper.GCS_SERVICE_HOST); + return List.of(StorageCredential.create(prefix, config)); + } + + private static List credentialsFromAdlsProperties( + String prefix, Map props) { + Map config = new LinkedHashMap<>(); + props.forEach((key, value) -> { + if (key.startsWith("adls.") && StringUtils.isNotBlank(value)) { + config.put(key, value); + } + }); + if (config.isEmpty()) { + return List.of(); + } + return List.of(StorageCredential.create(prefix, config)); + } + + private static List credentialsFromOssProperties( + String prefix, Map props) { + if (StringUtils.isBlank(props.get(OssVendedCredentialHadoopMapper.CLIENT_ACCESS_KEY_ID)) || + StringUtils.isBlank(props.get(OssVendedCredentialHadoopMapper.CLIENT_ACCESS_KEY_SECRET))) { + return List.of(); + } + Map config = new LinkedHashMap<>(); + putIfPresent(config, props, OssVendedCredentialHadoopMapper.CLIENT_ACCESS_KEY_ID); + putIfPresent(config, props, OssVendedCredentialHadoopMapper.CLIENT_ACCESS_KEY_SECRET); + putIfPresent(config, props, OssVendedCredentialHadoopMapper.CLIENT_SECURITY_TOKEN); + putIfPresent(config, props, OssVendedCredentialHadoopMapper.OSS_ENDPOINT); + return List.of(StorageCredential.create(prefix, 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)); + } + } +} 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 index 1d80244e70d9..f74b673bdf44 100644 --- 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 @@ -273,7 +273,7 @@ public void propagateSecretsOnlyWhenJobPropertiesNull() { new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO), "db.t"); Map jobSecrets = Maps.newHashMap(); - IcebergVendedCredentialUtil.propagateToJob(table, "ice01", null, jobSecrets, new Configuration()); + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", null, jobSecrets, new HiveConf()); assertThat(jobSecrets) .containsEntry("fs.s3a.bucket.my-bucket.access.key", "access") @@ -297,6 +297,348 @@ public void propagateSecretsOnlyWhenJobPropertiesNull() { .containsEntry(S3FileIOProperties.SECRET_ACCESS_KEY, "secret"); } + @Test + public void propagateToJobMapsGcsProperties() { + Configuration conf = new HiveConf(); + conf.set("iceberg.catalog.ice01.gcs.service.host", "http://host-gcs:4443"); + + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "gs://my-bucket/", + Map.of( + "gcs.oauth2.token", "gcs-token", + "gcs.project-id", "my-project", + "gcs.service.host", "http://internal-gcs:4443")))); + + Table table = + new BaseTable(new StaticTableOperations("gs://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.gcs.project-id", "my-project") + .containsEntry("iceberg.catalog.ice01.gcs.service.host", "http://host-gcs:4443") + .containsEntry("fs.gs.project.id", "my-project") + .containsEntry("fs.gs.storage.root.url", "http://host-gcs:4443") + .doesNotContainKey("iceberg.catalog.ice01.gcs.oauth2.token") + .doesNotContainKey("fs.gs.auth.access.token"); + + // The GCS OAuth token has no plain Hadoop connector key; it must not be emitted to job conf + // and only rides the serialized Iceberg credentials blob. + assertThat(jobSecrets) + .doesNotContainKey("fs.gs.auth.access.token") + .doesNotContainKey("iceberg.catalog.ice01.gcs.oauth2.token") + .satisfies(map -> + assertThat(map.get(InputFormatConfig.vendedCredentialsKey("db.t"))) + .isNotBlank()); + + List serialized = + SerializationUtil.deserializeFromBase64( + jobSecrets.get(InputFormatConfig.vendedCredentialsKey("db.t"))); + assertThat(serialized.getFirst().config()).containsEntry("gcs.oauth2.token", "gcs-token"); + } + + @Test + public void propagateToJobMapsAdlsSasToken() { + String prefix = "abfss://mycontainer@mystorageaccount.dfs.core.windows.net/"; + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + prefix, + Map.of( + "adls.sas-token.mystorageaccount", "sas-token-value", + "adls.sas-token-expires-at-ms.mystorageaccount", "1730234407000")))); + + Table table = + new BaseTable(new StaticTableOperations(prefix + "t", fileIO), "db.t"); + Map jobProps = Maps.newHashMap(); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob( + table, "ice01", jobProps, jobSecrets, new HiveConf()); + + assertThat(jobProps) + .containsEntry( + "iceberg.catalog.ice01.adls.sas-token-expires-at-ms.mystorageaccount", + "1730234407000") + // ABFS only uses a fixed SAS token when the account auth type is set to SAS. + .containsEntry( + "fs.azure.account.auth.type.mystorageaccount.dfs.core.windows.net", + "SAS") + .doesNotContainKey("adls.sas-token.mystorageaccount") + .doesNotContainKey("fs.azure.sas.fixed.token.mystorageaccount.dfs.core.windows.net"); + + assertThat(jobSecrets) + .containsEntry( + "fs.azure.sas.fixed.token.mystorageaccount.dfs.core.windows.net", + "sas-token-value"); + } + + @Test + public void propagateToJobMapsAdlsAccountKey() { + String prefix = "abfss://mycontainer@mystorageaccount.dfs.core.windows.net/"; + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + prefix, + Map.of( + "adls.auth.shared-key.account.name", "mystorageaccount", + "adls.auth.shared-key.account.key", "account-key")))); + + Table table = + new BaseTable(new StaticTableOperations(prefix + "t", fileIO), "db.t"); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", null, jobSecrets, new HiveConf()); + + assertThat(jobSecrets) + .containsEntry( + "fs.azure.account.key.mystorageaccount.dfs.core.windows.net", + "account-key") + .doesNotContainKey("iceberg.catalog.ice01.adls.auth.shared-key.account.key"); + } + + @Test + public void propagateToJobMapsOssProperties() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "oss://my-bucket/", + Map.of( + "client.access-key-id", "oss-access", + "client.access-key-secret", "oss-secret", + "client.security-token", "oss-token", + "oss.endpoint", "oss-cn-hangzhou.aliyuncs.com")))); + + Table table = + new BaseTable(new StaticTableOperations("oss://my-bucket/t", fileIO), "db.t"); + Map jobProps = Maps.newHashMap(); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob( + table, "ice01", jobProps, jobSecrets, new HiveConf()); + + assertThat(jobProps) + .containsEntry("iceberg.catalog.ice01.oss.endpoint", "oss-cn-hangzhou.aliyuncs.com") + .containsEntry("fs.oss.endpoint", "oss-cn-hangzhou.aliyuncs.com") + .doesNotContainKey("client.access-key-id"); + + assertThat(jobSecrets) + .containsEntry("fs.oss.accessKeyId", "oss-access") + .containsEntry("fs.oss.accessKeySecret", "oss-secret") + .containsEntry("fs.oss.securityToken", "oss-token"); + } + + @Test + public void propagateToJobMapsS3SessionToken() { + 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.SESSION_TOKEN, "session-token")))); + + Table table = + new BaseTable(new StaticTableOperations("s3://my-bucket/t", fileIO), "db.t"); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", null, jobSecrets, new HiveConf()); + + assertThat(jobSecrets) + .containsEntry("fs.s3a.bucket.my-bucket.session.token", "session-token"); + } + + @Test + public void extractCredentialsFromGcsFileIoPropertiesWhenCredentialListEmpty() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.initialize( + Map.of( + "gcs.oauth2.token", "gcs-token", + "gcs.project-id", "my-project")); + Schema schema = new Schema(Types.NestedField.required(1, "x", Types.IntegerType.get())); + TableMetadata metadata = + TableMetadata.newTableMetadata( + schema, PartitionSpec.unpartitioned(), "gs://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("gs://my-bucket/"); + assertThat(extracted.config()) + .containsEntry("gcs.oauth2.token", "gcs-token") + .containsEntry("gcs.project-id", "my-project"); + } + + @Test + public void propagateToJobSerializesBlobWithoutHadoopMappingForUnknownScheme() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "hdfs://namenode:8020/warehouse/", + Map.of("custom.key", "custom-value")))); + + Table table = + new BaseTable(new StaticTableOperations("hdfs://namenode:8020/warehouse/t", fileIO), "db.t"); + Map jobProps = Maps.newHashMap(); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob( + table, "ice01", jobProps, jobSecrets, new HiveConf()); + + assertThat(jobProps) + .containsEntry("iceberg.catalog.ice01.custom.key", "custom-value"); + assertThat(jobSecrets) + .satisfies(map -> + assertThat(map.get(InputFormatConfig.vendedCredentialsKey("db.t"))) + .isNotBlank()); + } + + /** + * Tests the full HIVE-20651 Credentials-channel round trip for a non-S3 (GCS) provider, proving + * the executor restore path is provider-agnostic: only S3 was previously covered end-to-end. + */ + @Test + public void applyFromJobConfRestoresGcsCredentialsViaCredentialsChannel() throws Exception { + CredentialFileIO hs2Io = new CredentialFileIO(); + hs2Io.setCredentials( + List.of( + StorageCredential.create( + "gs://my-bucket/", + Map.of("gcs.oauth2.token", "gcs-token", "gcs.project-id", "my-project")))); + Table hs2Table = + new BaseTable(new StaticTableOperations("gs://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 HiveConf()); + tableDesc.setJobSecrets(jobSecrets); + + JobConf jobConf = new JobConf(); + PlanUtils.configureJobConf(tableDesc, jobConf); + // The GCS OAuth token is a secret — it must travel via the Credentials channel, never 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("gs://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("gs://my-bucket/"); + assertThat(applied.config()) + .containsEntry("gcs.oauth2.token", "gcs-token") + .containsEntry("gcs.project-id", "my-project"); + } + + /** + * Tests provider selection via the config-key fallback in {@code VendedCredentialHadoopMappers}: + * when the prefix scheme is not recognized, the mapper is chosen from the Iceberg config keys + * (here {@code gcs.*}), so GCS Hadoop keys are still emitted. + */ + @Test + public void propagateToJobSelectsMapperByConfigKeyWhenSchemeUnknown() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "unknownscheme://my-bucket/", + Map.of("gcs.project-id", "my-project")))); + + Table table = + new BaseTable(new StaticTableOperations("unknownscheme://my-bucket/t", fileIO), "db.t"); + Map jobProps = Maps.newHashMap(); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", jobProps, jobSecrets, new HiveConf()); + + assertThat(jobProps).containsEntry("fs.gs.project.id", "my-project"); + } + + /** + * Tests {@code accountFromScope} for an account-only ADLS prefix (no {@code container@} + * authority), which must still yield the correct account-scoped Hadoop account-key. + */ + @Test + public void propagateToJobMapsAdlsAccountKeyForAccountOnlyPrefix() { + String prefix = "abfss://mystorageaccount.dfs.core.windows.net/"; + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + prefix, + Map.of( + "adls.auth.shared-key.account.name", "mystorageaccount", + "adls.auth.shared-key.account.key", "account-key")))); + + Table table = + new BaseTable(new StaticTableOperations(prefix + "t", fileIO), "db.t"); + Map jobSecrets = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", null, jobSecrets, new HiveConf()); + + assertThat(jobSecrets) + .containsEntry("fs.azure.account.key.mystorageaccount.dfs.core.windows.net", "account-key"); + } + + /** + * Tests that a provider access-key id is treated as a secret: with only {@code jobProperties} + * passed, an OSS access-key id must not leak into job properties (neither the catalog-prefixed + * key nor the Hadoop key) — it belongs solely in the secrets channel. + */ + @Test + public void propagateToJobKeepsOssAccessKeyIdOutOfJobProperties() { + CredentialFileIO fileIO = new CredentialFileIO(); + fileIO.setCredentials( + List.of( + StorageCredential.create( + "oss://my-bucket/", + Map.of( + "client.access-key-id", "oss-access", + "client.access-key-secret", "oss-secret")))); + + Table table = + new BaseTable(new StaticTableOperations("oss://my-bucket/t", fileIO), "db.t"); + Map jobProps = Maps.newHashMap(); + + IcebergVendedCredentialUtil.propagateToJob(table, "ice01", jobProps, null, new HiveConf()); + + assertThat(jobProps) + .doesNotContainKey("fs.oss.accessKeyId") + .doesNotContainKey("iceberg.catalog.ice01.client.access-key-id") + .doesNotContainKey("fs.oss.accessKeySecret"); + } + /** * Tests {@link IcebergVendedCredentialUtil#applyFromJobConf} for a non-vending catalog. *