From 0e66ef220b4f0d2f7b89ebffb7190e54ec88a730 Mon Sep 17 00:00:00 2001 From: Guangyu Yang Date: Wed, 19 Aug 2026 11:54:14 -0400 Subject: [PATCH 1/2] [GLUTEN-12701][CORE][VL] Support Iceberg REST-catalog vended credentials in native scans A REST catalog can be configured to vend credentials instead of letting the compute use its own identity ('X-Iceberg-Access-Delegation: vended-credentials', e.g. Apache Polaris). loadTable then returns credentials scoped to one table, and only the JVM FileIO ever sees them: the native scan path receives file paths only, so Velox's S3 client falls back to the process credential chain, which by design has no access to the warehouse. Every native TableScan of such a table fails with S3 403 while the same query reads fine on vanilla Spark. Read the vended credentials from the scan table's FileIO at split planning time and carry them, with the table location, in a new table-scoped ReadRel.LocalFiles.read_properties map. A LocalFiles is single-table by construction, so table granularity is split granularity. The Velox backend turns the union of the task's scans into a file system token provider that resolves credentials by longest-prefix match of the file path against the table locations, so a query joining two vended-credential tables that live in one bucket cannot mix them up. FileIO.properties() is implemented by S3FileIO and ResolvingFileIO and defaults to throwing, which is treated as "no credentials". Tables whose files the process credentials can read attach nothing and are untouched. New conf spark.gluten.sql.columnar.iceberg.enableVendedCredentials (default true): when disabled, scans of tables read with vended credentials fail native validation and fall back to vanilla Spark, which reads them correctly through the JVM FileIO. Validation also requires the new backend capability BackendSettingsApi.supportIcebergVendedCredentialsRead(), so a backend that does not consume read_properties falls back instead of failing with access-denied errors. Credentials are snapshotted on the driver, so a scan has to start within the lifetime of the credentials the catalog vended; executors cannot re-vend. A 403 on expiry stays a retriable task failure. Fixes #12701 --- .../backendsapi/velox/VeloxBackend.scala | 2 + cpp/velox/CMakeLists.txt | 3 +- cpp/velox/compute/VeloxPlanConverter.cc | 3 + cpp/velox/compute/WholeStageResultIterator.cc | 19 ++- cpp/velox/compute/WholeStageResultIterator.h | 6 + cpp/velox/substrait/SubstraitToVeloxPlan.h | 5 + cpp/velox/tests/CMakeLists.txt | 2 + cpp/velox/tests/GlutenS3TokenProviderTest.cc | 130 ++++++++++++++++++ cpp/velox/utils/GlutenS3TokenProvider.cc | 116 ++++++++++++++++ cpp/velox/utils/GlutenS3TokenProvider.h | 84 +++++++++++ docs/get-started/VeloxIceberg.md | 23 +++- .../gluten/config/GlutenIcebergConfig.scala | 13 ++ .../execution/IcebergScanTransformer.scala | 22 ++- .../source/GlutenIcebergSourceUtil.scala | 78 ++++++++++- ...ebergLocalFilesNodeReadPropertiesTest.java | 66 +++++++++ .../source/GlutenIcebergSourceUtilSuite.scala | 78 +++++++++++ .../gluten/substrait/rel/LocalFilesNode.java | 12 ++ .../substrait/proto/substrait/algebra.proto | 6 + .../backendsapi/BackendSettingsApi.scala | 6 + 19 files changed, 667 insertions(+), 7 deletions(-) create mode 100644 cpp/velox/tests/GlutenS3TokenProviderTest.cc create mode 100644 cpp/velox/utils/GlutenS3TokenProvider.cc create mode 100644 cpp/velox/utils/GlutenS3TokenProvider.h create mode 100644 gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeReadPropertiesTest.java create mode 100644 gluten-iceberg/src/test/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtilSuite.scala diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala index 27d1dc0a2d2..7368c0ee237 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala @@ -573,6 +573,8 @@ object VeloxBackendSettings extends BackendSettingsApi { override def supportIcebergInitialDefaultRead(): Boolean = true + override def supportIcebergVendedCredentialsRead(): Boolean = true + override def reorderColumnsForPartitionWrite(): Boolean = true override def enableEnhancedFeatures(): Boolean = VeloxConfig.get.enableEnhancedFeatures() diff --git a/cpp/velox/CMakeLists.txt b/cpp/velox/CMakeLists.txt index e347381887e..8e6f5aabefa 100644 --- a/cpp/velox/CMakeLists.txt +++ b/cpp/velox/CMakeLists.txt @@ -214,7 +214,8 @@ set(VELOX_SRCS utils/VeloxWriterUtils.cc) if(ENABLE_S3) - list(APPEND VELOX_SRCS filesystem/GlutenS3FileSystem.cc) + list(APPEND VELOX_SRCS filesystem/GlutenS3FileSystem.cc + utils/GlutenS3TokenProvider.cc) find_package(ZLIB) endif() diff --git a/cpp/velox/compute/VeloxPlanConverter.cc b/cpp/velox/compute/VeloxPlanConverter.cc index 82c1f290368..43fe1aaff18 100644 --- a/cpp/velox/compute/VeloxPlanConverter.cc +++ b/cpp/velox/compute/VeloxPlanConverter.cc @@ -149,6 +149,9 @@ std::shared_ptr parseScanSplitInfo( splitInfo->partitionColumns.reserve(fileList.size()); splitInfo->properties.reserve(fileList.size()); splitInfo->metadataColumns.reserve(fileList.size()); + for (const auto& readProperty : localFiles.read_properties()) { + splitInfo->readProperties[readProperty.first] = readProperty.second; + } for (const auto& file : fileList) { // Expect all Partitions share the same index. splitInfo->partitionIndex = file.partition_index(); diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 4e680fb25ba..e5193ea8cd4 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -28,6 +28,9 @@ #include "compute/delta/DeltaSplitInfo.h" #include "config/VeloxConfig.h" #include "utils/ConfigExtractor.h" +#ifdef ENABLE_S3 +#include "utils/GlutenS3TokenProvider.h" +#endif #include "velox/connectors/hive/HiveConfig.h" #include "velox/connectors/hive/HiveConnectorSplit.h" #include "velox/exec/PlanNodeStats.h" @@ -277,10 +280,24 @@ std::shared_ptr WholeStageResultIterator::createNewVeloxQ "Gluten_Stage_{}_TID_{}_VTID_{}", std::to_string(taskInfo_.stageId), std::to_string(taskInfo_.taskId), - std::to_string(taskInfo_.vId))); + std::to_string(taskInfo_.vId)), + createFsTokenProvider()); return ctx; } +std::shared_ptr WholeStageResultIterator::createFsTokenProvider() const { +#ifdef ENABLE_S3 + std::vector> readProperties; + readProperties.reserve(scanInfos_.size()); + for (const auto& scanInfo : scanInfos_) { + readProperties.push_back(scanInfo->readProperties); + } + return GlutenS3TokenProvider::create(readProperties); +#else + return nullptr; +#endif +} + std::shared_ptr WholeStageResultIterator::next() { while (true) { if (!cursor_->moveNext()) { diff --git a/cpp/velox/compute/WholeStageResultIterator.h b/cpp/velox/compute/WholeStageResultIterator.h index fb2c9229ce3..650abd6f0ca 100644 --- a/cpp/velox/compute/WholeStageResultIterator.h +++ b/cpp/velox/compute/WholeStageResultIterator.h @@ -26,6 +26,7 @@ #include "substrait/plan.pb.h" #include "utils/Metrics.h" #include "velox/common/config/Config.h" +#include "velox/common/file/TokenProvider.h" #include "velox/connectors/hive/iceberg/IcebergSplit.h" #include "velox/core/PlanNode.h" #include "velox/exec/Cursor.h" @@ -110,6 +111,11 @@ class WholeStageResultIterator : public SplitAwareColumnarBatchIterator { /// Create QueryCtx. std::shared_ptr createNewVeloxQueryCtx(); + /// The file system token provider built from the scans' table-scoped read + /// properties, e.g. the per-table S3 credentials an Iceberg REST catalog + /// vended. Null when no scan carries credentials, or when built without S3. + std::shared_ptr createFsTokenProvider() const; + /// Get all the children plan node ids with postorder traversal. void getOrderedNodeIds( const std::shared_ptr&, diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.h b/cpp/velox/substrait/SubstraitToVeloxPlan.h index b0cf76fff3e..9fc4ea0c0a5 100644 --- a/cpp/velox/substrait/SubstraitToVeloxPlan.h +++ b/cpp/velox/substrait/SubstraitToVeloxPlan.h @@ -65,6 +65,11 @@ struct SplitInfo { /// The file sizes and modification times of the files to be scanned. std::vector> properties; + /// Table-scoped storage properties from LocalFiles.read_properties, e.g. the + /// S3 credentials an Iceberg REST catalog vended for this table plus the table + /// location. Empty for tables whose files the process credentials can read. + std::unordered_map readProperties; + /// The schema of the table being scanned. RowTypePtr tableSchema; diff --git a/cpp/velox/tests/CMakeLists.txt b/cpp/velox/tests/CMakeLists.txt index 87331de818c..09d90c9fd35 100644 --- a/cpp/velox/tests/CMakeLists.txt +++ b/cpp/velox/tests/CMakeLists.txt @@ -139,6 +139,8 @@ add_velox_test(velox_memory_test SOURCES MemoryManagerTest.cc) add_velox_test(buffer_outputstream_test SOURCES BufferOutputStreamTest.cc) if(ENABLE_S3) add_velox_test(gluten_s3_file_system_test SOURCES GlutenS3FileSystemTest.cc) + add_velox_test(gluten_s3_token_provider_test SOURCES + GlutenS3TokenProviderTest.cc) endif() add_velox_test(scoped_timer_test SOURCES ScopedTimerTest.cc) add_velox_test(row_based_checksum_test SOURCES RowBasedChecksumTest.cc) diff --git a/cpp/velox/tests/GlutenS3TokenProviderTest.cc b/cpp/velox/tests/GlutenS3TokenProviderTest.cc new file mode 100644 index 00000000000..f96482500ce --- /dev/null +++ b/cpp/velox/tests/GlutenS3TokenProviderTest.cc @@ -0,0 +1,130 @@ +/* + * 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. + */ + +#include "utils/GlutenS3TokenProvider.h" + +#include + +#include "velox/common/file/PlainUserNameTokenProvider.h" + +namespace gluten { +namespace { + +using namespace facebook::velox::filesystems; + +std::unordered_map readProperties( + const std::string& location, + const std::string& accessKeyId, + const std::string& secretAccessKey, + const std::string& sessionToken = "") { + std::unordered_map properties{ + {kReadPropertiesLocation, location}, + {kReadPropertiesAccessKeyId, accessKeyId}, + {kReadPropertiesSecretAccessKey, secretAccessKey}}; + if (!sessionToken.empty()) { + properties[kReadPropertiesSessionToken] = sessionToken; + } + return properties; +} + +std::shared_ptr tokenFor(const TokenProvider& provider, const std::string& path) { + return std::dynamic_pointer_cast(provider.getToken(S3AccessTokenKey{path})); +} + +// A key belonging to some other file system. +class OtherAccessTokenKey : public AccessTokenKey {}; + +} // namespace + +TEST(GlutenS3TokenProviderTest, noProviderWithoutCredentials) { + ASSERT_EQ(GlutenS3TokenProvider::create({}), nullptr); + // A scan of a table the process credentials can read carries nothing. + ASSERT_EQ(GlutenS3TokenProvider::create({{}}), nullptr); + // An incomplete credential set is not usable and must not be installed. + ASSERT_EQ( + GlutenS3TokenProvider::create( + {{{kReadPropertiesLocation, "s3://bucket/db/t"}, {kReadPropertiesAccessKeyId, "ASIA"}}}), + nullptr); +} + +TEST(GlutenS3TokenProviderTest, resolvesCredentialsOfTheTableOwningThePath) { + const auto provider = GlutenS3TokenProvider::create( + {readProperties("s3://bucket/db/first", "ASIAFIRST", "first-secret", "first-token"), + readProperties("s3a://bucket/db/second", "ASIASECOND", "second-secret")}); + ASSERT_NE(provider, nullptr); + + const auto first = tokenFor(*provider, "bucket/db/first/data/00000-0-a.parquet"); + ASSERT_NE(first, nullptr); + EXPECT_EQ(first->accessKeyId(), "ASIAFIRST"); + EXPECT_EQ(first->secretAccessKey(), "first-secret"); + EXPECT_EQ(first->sessionToken(), "first-token"); + + // Same bucket, different table: the other credential set, and no session + // token because the catalog vended none. + const auto second = tokenFor(*provider, "bucket/db/second/data/00000-0-b.parquet"); + ASSERT_NE(second, nullptr); + EXPECT_EQ(second->accessKeyId(), "ASIASECOND"); + EXPECT_EQ(second->secretAccessKey(), "second-secret"); + EXPECT_EQ(second->sessionToken(), ""); + + // A path no scan covers gets no token, which leaves the file system on its + // configured credentials. + EXPECT_EQ(tokenFor(*provider, "bucket/db/third/data/00000-0-c.parquet"), nullptr); + // A table name the prefix is a string prefix of is a different table. + EXPECT_EQ(tokenFor(*provider, "bucket/db/firstborn/data/00000-0-d.parquet"), nullptr); +} + +TEST(GlutenS3TokenProviderTest, theLongestMatchingPrefixWins) { + // A table whose location is nested inside another table's location must get + // its own credentials, not the enclosing one's. + const auto provider = GlutenS3TokenProvider::create( + {readProperties("s3://bucket/db", "ASIAOUTER", "outer-secret"), + readProperties("s3://bucket/db/nested", "ASIAINNER", "inner-secret")}); + ASSERT_NE(provider, nullptr); + + EXPECT_EQ(tokenFor(*provider, "bucket/db/nested/data/f.parquet")->accessKeyId(), "ASIAINNER"); + EXPECT_EQ(tokenFor(*provider, "bucket/db/other/data/f.parquet")->accessKeyId(), "ASIAOUTER"); +} + +TEST(GlutenS3TokenProviderTest, identityCoversAllCredentials) { + const auto provider = GlutenS3TokenProvider::create({readProperties("s3://bucket/db/t", "ASIA", "secret", "token")}); + const auto same = GlutenS3TokenProvider::create({readProperties("s3://bucket/db/t", "ASIA", "secret", "token")}); + // A re-vended credential set for the same table is a different identity, so + // velox's file handle cache cannot serve handles opened with the old one. + const auto rotated = + GlutenS3TokenProvider::create({readProperties("s3://bucket/db/t", "ASIA2", "secret2", "token2")}); + + EXPECT_TRUE(provider->equals(*same)); + EXPECT_EQ(provider->hash(), same->hash()); + EXPECT_FALSE(provider->equals(*rotated)); + EXPECT_NE(provider->hash(), rotated->hash()); + + // Providers of another kind are never equal. + PlainUserNameTokenProvider other{"user"}; + EXPECT_FALSE(provider->equals(other)); + // A key belonging to another file system resolves nothing. + EXPECT_EQ(provider->getToken(OtherAccessTokenKey{}), nullptr); +} + +TEST(GlutenS3TokenProviderTest, normalizesS3Schemes) { + EXPECT_EQ(GlutenS3TokenProvider::normalizeS3Path("s3://bucket/db/t"), "bucket/db/t"); + EXPECT_EQ(GlutenS3TokenProvider::normalizeS3Path("s3a://bucket/db/t"), "bucket/db/t"); + EXPECT_EQ(GlutenS3TokenProvider::normalizeS3Path("s3n://bucket/db/t"), "bucket/db/t"); + EXPECT_EQ(GlutenS3TokenProvider::normalizeS3Path("bucket/db/t"), "bucket/db/t"); +} + +} // namespace gluten diff --git a/cpp/velox/utils/GlutenS3TokenProvider.cc b/cpp/velox/utils/GlutenS3TokenProvider.cc new file mode 100644 index 00000000000..808110dee7a --- /dev/null +++ b/cpp/velox/utils/GlutenS3TokenProvider.cc @@ -0,0 +1,116 @@ +/* + * 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. + */ +#include "utils/GlutenS3TokenProvider.h" + +#include +#include + +#include "velox/common/base/BitUtil.h" + +namespace gluten { +namespace { + +// Segment-boundary-safe prefix match: "bucket/tableA" must not claim +// "bucket/tableAB/part.parquet". +bool prefixMatches(const std::string& path, const std::string& prefix) { + if (prefix.empty() || path.size() < prefix.size() || path.compare(0, prefix.size(), prefix) != 0) { + return false; + } + return path.size() == prefix.size() || prefix.back() == '/' || path[prefix.size()] == '/'; +} + +std::string findOrEmpty(const std::unordered_map& properties, const char* key) { + const auto it = properties.find(key); + return it == properties.end() ? "" : it->second; +} + +} // namespace + +GlutenS3TokenProvider::GlutenS3TokenProvider(std::map credentialsByPrefix) + : credentialsByPrefix_(std::move(credentialsByPrefix)) { + const std::hash hasher; + size_t hash = 0; + // std::map iteration order is deterministic, so equal contents hash equally. + for (const auto& [prefix, credentials] : credentialsByPrefix_) { + hash = facebook::velox::bits::hashMix(hash, hasher(prefix)); + hash = facebook::velox::bits::hashMix(hash, hasher(credentials.accessKeyId)); + hash = facebook::velox::bits::hashMix(hash, hasher(credentials.secretAccessKey)); + hash = facebook::velox::bits::hashMix(hash, hasher(credentials.sessionToken)); + } + hash_ = hash; +} + +std::shared_ptr GlutenS3TokenProvider::create( + const std::vector>& readProperties) { + std::map credentialsByPrefix; + for (const auto& properties : readProperties) { + const auto location = findOrEmpty(properties, kReadPropertiesLocation); + const auto accessKeyId = findOrEmpty(properties, kReadPropertiesAccessKeyId); + const auto secretAccessKey = findOrEmpty(properties, kReadPropertiesSecretAccessKey); + if (location.empty() || accessKeyId.empty() || secretAccessKey.empty()) { + continue; + } + credentialsByPrefix[normalizeS3Path(location)] = + S3TableCredentials{accessKeyId, secretAccessKey, findOrEmpty(properties, kReadPropertiesSessionToken)}; + } + if (credentialsByPrefix.empty()) { + return nullptr; + } + return std::make_shared(std::move(credentialsByPrefix)); +} + +bool GlutenS3TokenProvider::equals(const facebook::velox::filesystems::TokenProvider& other) const { + const auto* typedOther = dynamic_cast(&other); + return typedOther != nullptr && credentialsByPrefix_ == typedOther->credentialsByPrefix_; +} + +size_t GlutenS3TokenProvider::hash() const { + return hash_; +} + +std::shared_ptr GlutenS3TokenProvider::getToken( + const facebook::velox::filesystems::AccessTokenKey& key) const { + const auto* s3Key = dynamic_cast(&key); + if (s3Key == nullptr) { + return nullptr; + } + const auto& path = s3Key->path(); + const S3TableCredentials* longestMatch = nullptr; + size_t longestMatchSize = 0; + for (const auto& [prefix, credentials] : credentialsByPrefix_) { + if (prefix.size() >= longestMatchSize && prefixMatches(path, prefix)) { + longestMatch = &credentials; + longestMatchSize = prefix.size(); + } + } + if (longestMatch == nullptr) { + return nullptr; + } + return std::make_shared( + longestMatch->accessKeyId, longestMatch->secretAccessKey, longestMatch->sessionToken); +} + +std::string GlutenS3TokenProvider::normalizeS3Path(const std::string& path) { + for (const char* scheme : {"s3://", "s3a://", "s3n://"}) { + if (path.rfind(scheme, 0) == 0) { + return path.substr(std::strlen(scheme)); + } + } + return path; +} + +} // namespace gluten diff --git a/cpp/velox/utils/GlutenS3TokenProvider.h b/cpp/velox/utils/GlutenS3TokenProvider.h new file mode 100644 index 00000000000..10c5261730d --- /dev/null +++ b/cpp/velox/utils/GlutenS3TokenProvider.h @@ -0,0 +1,84 @@ +/* + * 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. + */ +#pragma once + +#include +#include +#include +#include +#include + +#include "velox/common/file/TokenProvider.h" +#include "velox/connectors/hive/storage_adapters/s3fs/S3AccessToken.h" + +namespace gluten { + +// Keys of the LocalFiles.read_properties map. Contract with the JVM side +// (GlutenIcebergSourceUtil.vendedReadProperties): the Iceberg FileIO property +// names of the credentials, plus the table location under kReadPropertiesLocation. +constexpr const char* kReadPropertiesLocation = "location"; +constexpr const char* kReadPropertiesAccessKeyId = "s3.access-key-id"; +constexpr const char* kReadPropertiesSecretAccessKey = "s3.secret-access-key"; +constexpr const char* kReadPropertiesSessionToken = "s3.session-token"; + +struct S3TableCredentials { + std::string accessKeyId; + std::string secretAccessKey; + // Empty when the credentials are not temporary. + std::string sessionToken; + + bool operator==(const S3TableCredentials& other) const { + return accessKeyId == other.accessKeyId && secretAccessKey == other.secretAccessKey && + sessionToken == other.sessionToken; + } +}; + +/// Resolves the per-table S3 credentials for native reads of tables whose +/// credentials an Iceberg REST catalog vends. Holds the credentials of all the +/// task's scans keyed by normalized table-location prefix; getToken() +/// longest-prefix-matches the file path, so a query joining two tables that +/// live in one bucket but carry different credentials cannot mix them up. +/// +/// equals()/hash() cover the full contents because they key velox's file handle +/// cache: a rotated credential set can never be served a handle that was opened +/// with the credentials it replaced. +class GlutenS3TokenProvider final : public facebook::velox::filesystems::TokenProvider { + public: + explicit GlutenS3TokenProvider(std::map credentialsByPrefix); + + /// Builds a provider from the read properties of a task's scans, or nullptr + /// when none of them carries credentials. + static std::shared_ptr create( + const std::vector>& readProperties); + + bool equals(const facebook::velox::filesystems::TokenProvider& other) const override; + + size_t hash() const override; + + std::shared_ptr getToken( + const facebook::velox::filesystems::AccessTokenKey& key) const override; + + /// "s3://bucket/path" (also s3a/s3n) -> "bucket/path", matching the + /// scheme-stripped paths the S3 file system puts in S3AccessTokenKey. + static std::string normalizeS3Path(const std::string& path); + + private: + const std::map credentialsByPrefix_; + size_t hash_; +}; + +} // namespace gluten diff --git a/docs/get-started/VeloxIceberg.md b/docs/get-started/VeloxIceberg.md index 42901a26cff..be7ccaee82d 100644 --- a/docs/get-started/VeloxIceberg.md +++ b/docs/get-started/VeloxIceberg.md @@ -103,8 +103,29 @@ the added column name is same to the deleted column, the scan will fall back. | --- | --- | --- | | spark.gluten.sql.columnar.iceberg.enableNativeRead | true | Enable offloading Iceberg scans to the native backend. When disabled, Iceberg scans fall back to vanilla Spark while scans of other formats stay offloaded. | | spark.gluten.sql.columnar.iceberg.enableNativeWrite | true | Enable offloading Iceberg writes to the native backend. When disabled, Iceberg writes fall back to vanilla Spark. Note the Velox backend additionally requires `spark.gluten.sql.enable.enhancedFeatures` to be enabled. | +| spark.gluten.sql.columnar.iceberg.enableVendedCredentials | true | Pass the per-table S3 credentials an Iceberg REST catalog vends from the table's FileIO to the native backend, so that such tables can be scanned natively. When disabled, scans of tables read with vended credentials fall back to vanilla Spark. Tables whose files are readable with the process credentials are unaffected either way. | -Both options are runtime modifiable, so they can be flipped per session with `SET`. +All three options are runtime modifiable, so they can be flipped per session with `SET`. + +### Credential vending + +REST catalogs can be configured to vend credentials rather than let the compute +use its own identity, for example Apache Polaris with + +``` +spark.sql.catalog..header.X-Iceberg-Access-Delegation=vended-credentials +``` + +In that case `loadTable` returns credentials scoped to one table, and only the +JVM `FileIO` sees them. Gluten reads them from the scan table's `FileIO` at split +planning time and passes them to the native reader with the split, which resolves +them per table by matching the file path against the table locations. A query may +therefore join tables that live in the same bucket but carry different +credentials. + +The credentials are read once on the driver, so a scan has to start within the +lifetime of the credentials the catalog vended; executors cannot ask the catalog +for new ones. ### Catalogs All the catalog configurations are transparent to Gluten diff --git a/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala b/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala index c148ff135bd..7d1af11f9b0 100644 --- a/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala +++ b/gluten-iceberg/src/main/scala/org/apache/gluten/config/GlutenIcebergConfig.scala @@ -24,6 +24,8 @@ class GlutenIcebergConfig(conf: SQLConf) extends GlutenCoreConfig(conf) { def enableNativeRead: Boolean = getConf(ENABLE_NATIVE_READ) def enableNativeWrite: Boolean = getConf(ENABLE_NATIVE_WRITE) + + def enableVendedCredentials: Boolean = getConf(ENABLE_VENDED_CREDENTIALS) } object GlutenIcebergConfig extends ConfigRegistry { @@ -39,6 +41,17 @@ object GlutenIcebergConfig extends ConfigRegistry { .booleanConf .createWithDefault(true) + val ENABLE_VENDED_CREDENTIALS: ConfigEntry[Boolean] = + buildConf("spark.gluten.sql.columnar.iceberg.enableVendedCredentials") + .doc("Pass the per-table S3 credentials an Iceberg REST catalog vends (e.g. Apache" + + " Polaris with 'X-Iceberg-Access-Delegation: vended-credentials') from the table's" + + " FileIO to the native backend, so that such tables can be scanned natively. When" + + " disabled, scans of tables read with vended credentials fall back to vanilla Spark," + + " which reads them through the JVM FileIO. Tables whose files are readable with the" + + " process credentials are unaffected either way.") + .booleanConf + .createWithDefault(true) + val ENABLE_NATIVE_WRITE: ConfigEntry[Boolean] = buildConf("spark.gluten.sql.columnar.iceberg.enableNativeWrite") .doc("Enable offloading Iceberg writes to the native backend. When disabled, Iceberg" + diff --git a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala index 16018e086df..7f9ab2d835f 100644 --- a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala +++ b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala @@ -17,6 +17,7 @@ package org.apache.gluten.execution import org.apache.gluten.backendsapi.BackendsApiManager +import org.apache.gluten.config.GlutenIcebergConfig import org.apache.gluten.exception.GlutenNotSupportException import org.apache.gluten.execution.IcebergScanTransformer.{containsMetadataColumn, containsUuidOrFixedType} import org.apache.gluten.sql.shims.SparkShimLoader @@ -76,6 +77,9 @@ case class IcebergScanTransformer( GlutenIcebergSourceUtil.getFieldIds(scan) } + private lazy val icebergVendedReadProperties = + GlutenIcebergSourceUtil.vendedReadProperties(scan) + override def withNewPushdownFilters(filters: Seq[Expression]): BatchScanExecTransformerBase = { this.copy(pushDownFilters = Some(filters)) } @@ -90,6 +94,21 @@ case class IcebergScanTransformer( return validationResult } + // Files of a table read with catalog-vended credentials are not readable with + // the process credentials, so offloading such a scan without passing the vended + // credentials down would only produce access-denied errors. + if (!icebergVendedReadProperties.isEmpty) { + if (!GlutenIcebergConfig.get.enableVendedCredentials) { + return ValidationResult.failed( + "Table is read with catalog-vended credentials and " + + s"${GlutenIcebergConfig.ENABLE_VENDED_CREDENTIALS.key} is disabled") + } + if (!BackendsApiManager.getSettings.supportIcebergVendedCredentialsRead()) { + return ValidationResult.failed( + "Table is read with catalog-vended credentials, which this backend cannot use") + } + } + if (!BackendsApiManager.getSettings.supportIcebergEqualityDeleteRead()) { val notSupport = table match { case t: SparkTable => @@ -230,7 +249,8 @@ case class IcebergScanTransformer( getPartitionSchema, metadataColumnNames, icebergFieldIds, - icebergInitialDefaults) + icebergInitialDefaults, + icebergVendedReadProperties) case _ => throw new GlutenNotSupportException() } numSplits.add(splitInfo.asInstanceOf[LocalFilesNode].getPaths.size()) diff --git a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala index db1bb024afa..78b24e87524 100644 --- a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala +++ b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala @@ -32,7 +32,7 @@ import org.apache.iceberg._ import org.apache.iceberg.spark.SparkSchemaUtil import java.lang.{Class, Long => JLong} -import java.util.{ArrayList => JArrayList, HashMap => JHashMap, List => JList, Map => JMap} +import java.util.{ArrayList => JArrayList, Collections, HashMap => JHashMap, List => JList, Map => JMap} import java.util.Locale import scala.collection.JavaConverters._ @@ -42,6 +42,22 @@ object GlutenIcebergSourceUtil { private val InputFileBlockStartCol = "input_file_block_start" private val InputFileBlockLengthCol = "input_file_block_length" + // Contract with the native backend: the split's read_properties map carries the + // table location under LocationKey plus the vended S3 credentials under their + // Iceberg FileIO property names. + val LocationKey = "location" + private val S3AccessKeyId = "s3.access-key-id" + private val S3SecretAccessKey = "s3.secret-access-key" + // Carried verbatim when the catalog vends them alongside the key pair. + private val OptionalCredentialKeys = Seq( + "s3.session-token", + "s3.session-token-expires-at-ms", + "s3.endpoint", + "s3.path-style-access", + "s3.region", + "client.region" + ) + def getClassOfSparkBatchQueryScan(): Class[SparkBatchQueryScan] = { classOf[SparkBatchQueryScan] } @@ -55,12 +71,64 @@ object GlutenIcebergSourceUtil { } } + /** + * The per-table S3 credentials the catalog vended for this scan's table, empty when the table has + * none - which is the case whenever the files are readable with the process credentials. + * + * The credentials are read on the driver at planning time and travel with the split; executors + * cannot re-vend them, so a scan has to start within the vend TTL. + */ + def vendedReadProperties(sparkScan: Scan): JMap[String, String] = sparkScan match { + case scan: SparkBatchQueryScan => + val table = scan.table() + val ioProperties = + try { + table.io().properties() + } catch { + // FileIO.properties() is implemented by S3FileIO and ResolvingFileIO but + // defaults to throwing on FileIOs that do not expose their configuration. + // Such a FileIO cannot be carrying vended credentials. + case _: UnsupportedOperationException => Collections.emptyMap[String, String]() + } + extractVendedReadProperties( + ioProperties, + BackendsApiManager.getTransformerApiInstance.encodeFilePathIfNeed(table.location())) + case _ => Collections.emptyMap() + } + + /** + * Keeps the vended credential set only when the access-key/secret pair is present, carrying the + * optional companions verbatim and the table location under [[LocationKey]]. + */ + private[source] def extractVendedReadProperties( + ioProperties: JMap[String, String], + encodedTableLocation: String): JMap[String, String] = { + val accessKeyId = ioProperties.get(S3AccessKeyId) + val secretAccessKey = ioProperties.get(S3SecretAccessKey) + if (accessKeyId == null || secretAccessKey == null) { + return Collections.emptyMap() + } + val readProperties = new JHashMap[String, String]() + readProperties.put(S3AccessKeyId, accessKeyId) + readProperties.put(S3SecretAccessKey, secretAccessKey) + OptionalCredentialKeys.foreach { + key => + val value = ioProperties.get(key) + if (value != null) { + readProperties.put(key, value) + } + } + readProperties.put(LocationKey, encodedTableLocation) + readProperties + } + def genSplitInfo( partition: SparkDataSourceRDDPartition, readPartitionSchema: StructType, metadataColumnNames: Seq[String], fieldIds: JMap[String, Integer], - initialDefaults: JMap[String, String]): SplitInfo = { + initialDefaults: JMap[String, String], + readProperties: JMap[String, String]): SplitInfo = { val paths = new JArrayList[String]() val starts = new JArrayList[JLong]() val lengths = new JArrayList[JLong]() @@ -94,7 +162,7 @@ object GlutenIcebergSourceUtil { case o => throw new GlutenNotSupportException(s"Unsupported input partition type: $o") } - IcebergLocalFilesBuilder.makeIcebergLocalFiles( + val localFiles = IcebergLocalFilesBuilder.makeIcebergLocalFiles( partition.index, paths, starts, @@ -110,6 +178,10 @@ object GlutenIcebergSourceUtil { fieldIds, initialDefaults ) + if (!readProperties.isEmpty) { + localFiles.setReadProperties(readProperties) + } + localFiles } def getFieldIds(sparkScan: Scan): JHashMap[String, Integer] = { diff --git a/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeReadPropertiesTest.java b/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeReadPropertiesTest.java new file mode 100644 index 00000000000..8d083243886 --- /dev/null +++ b/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeReadPropertiesTest.java @@ -0,0 +1,66 @@ +/* + * 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.gluten.substrait.rel; + +import io.substrait.proto.ReadRel; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +public class IcebergLocalFilesNodeReadPropertiesTest { + + private static IcebergLocalFilesNode newNode() { + return new IcebergLocalFilesNode( + 0, + Collections.singletonList("s3a://warehouse/ns/db/table/data/00000-0-data.parquet"), + Collections.singletonList(0L), + Collections.singletonList(100L), + Collections.singletonList(Collections.emptyMap()), + LocalFilesNode.ReadFileFormat.ParquetReadFormat, + Collections.emptyList(), + Collections.singletonList(Collections.emptyList()), + Collections.singletonList(Collections.emptyMap()), + Collections.emptyMap(), + Collections.emptyMap()); + } + + @Test + public void serializesTableScopedReadProperties() { + Map readProperties = new HashMap<>(); + readProperties.put("location", "s3://warehouse/ns/db/table"); + readProperties.put("s3.access-key-id", "ASIAVENDED"); + readProperties.put("s3.secret-access-key", "secret"); + readProperties.put("s3.session-token", "token"); + + IcebergLocalFilesNode node = newNode(); + node.setReadProperties(readProperties); + + ReadRel.LocalFiles localFiles = node.toProtobuf(); + + Assert.assertEquals(readProperties, localFiles.getReadPropertiesMap()); + } + + @Test + public void omitsReadPropertiesWhenTheTableHasNone() { + ReadRel.LocalFiles localFiles = newNode().toProtobuf(); + + Assert.assertEquals(Collections.emptyMap(), localFiles.getReadPropertiesMap()); + } +} diff --git a/gluten-iceberg/src/test/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtilSuite.scala b/gluten-iceberg/src/test/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtilSuite.scala new file mode 100644 index 00000000000..5eef81e0403 --- /dev/null +++ b/gluten-iceberg/src/test/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtilSuite.scala @@ -0,0 +1,78 @@ +/* + * 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.spark.source + +import org.scalatest.funsuite.AnyFunSuite + +import java.util.{HashMap => JHashMap} + +class GlutenIcebergSourceUtilSuite extends AnyFunSuite { + + private val location = "s3://warehouse/ns/db/table" + + test("the vended credential set is extracted with the location and its companions") { + val ioProperties = new JHashMap[String, String]() + ioProperties.put("s3.access-key-id", "ASIAVENDED") + ioProperties.put("s3.secret-access-key", "secret") + ioProperties.put("s3.session-token", "token") + ioProperties.put("s3.session-token-expires-at-ms", "1780000000000") + ioProperties.put("client.region", "us-west-2") + ioProperties.put("io-impl", "org.apache.iceberg.aws.s3.S3FileIO") + + val extracted = GlutenIcebergSourceUtil.extractVendedReadProperties(ioProperties, location) + + val expected = new JHashMap[String, String]() + expected.put("s3.access-key-id", "ASIAVENDED") + expected.put("s3.secret-access-key", "secret") + expected.put("s3.session-token", "token") + expected.put("s3.session-token-expires-at-ms", "1780000000000") + expected.put("client.region", "us-west-2") + expected.put(GlutenIcebergSourceUtil.LocationKey, location) + // io-impl is not a credential, so it must not ride along. + assert(extracted == expected) + } + + test("a table without vended credentials extracts nothing") { + val noCredentials = new JHashMap[String, String]() + noCredentials.put("io-impl", "org.apache.iceberg.aws.s3.S3FileIO") + noCredentials.put("client.region", "us-west-2") + assert( + GlutenIcebergSourceUtil.extractVendedReadProperties(noCredentials, location).isEmpty, + "a region without an access-key/secret pair is not a vended credential set" + ) + + val secretOnly = new JHashMap[String, String]() + secretOnly.put("s3.secret-access-key", "secret") + assert(GlutenIcebergSourceUtil.extractVendedReadProperties(secretOnly, location).isEmpty) + + val accessKeyOnly = new JHashMap[String, String]() + accessKeyOnly.put("s3.access-key-id", "ASIAVENDED") + assert(GlutenIcebergSourceUtil.extractVendedReadProperties(accessKeyOnly, location).isEmpty) + } + + test("the session token is optional") { + val staticKeys = new JHashMap[String, String]() + staticKeys.put("s3.access-key-id", "AKIASTATIC") + staticKeys.put("s3.secret-access-key", "secret") + + val expected = new JHashMap[String, String]() + expected.put("s3.access-key-id", "AKIASTATIC") + expected.put("s3.secret-access-key", "secret") + expected.put(GlutenIcebergSourceUtil.LocationKey, location) + assert(GlutenIcebergSourceUtil.extractVendedReadProperties(staticKeys, location) == expected) + } +} diff --git a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java index dfea5dd7531..6f27affde50 100644 --- a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java +++ b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java @@ -59,6 +59,10 @@ public enum ReadFileFormat { private Boolean iterAsInput = false; private StructType fileSchema; private Map fileReadProperties; + // Table-scoped storage properties required to open this split's files + // natively, e.g. Iceberg catalog-vended S3 credentials. Serialized into + // ReadRel.LocalFiles.read_properties when non-empty. + private Map readProperties; LocalFilesNode( Integer index, @@ -112,6 +116,7 @@ protected LocalFilesNode(LocalFilesNode other, List> otherMe this.fileFormat = other.fileFormat; this.preferredLocations.addAll(other.preferredLocations); this.fileReadProperties = other.fileReadProperties; + this.readProperties = other.readProperties; this.iterAsInput = other.iterAsInput; this.fileSchema = other.fileSchema; this.otherMetadataColumns.addAll(otherMetadataColumns); @@ -130,6 +135,10 @@ public void setFileSchema(StructType schema) { this.fileSchema = schema; } + public void setReadProperties(Map readProperties) { + this.readProperties = readProperties; + } + private NamedStruct buildNamedStruct() { NamedStruct.Builder namedStructBuilder = NamedStruct.newBuilder(); @@ -284,6 +293,9 @@ public ReadRel.LocalFiles toProtobuf() { processFileBuilder(fileBuilder, i); localFilesBuilder.addItems(fileBuilder.build()); } + if (readProperties != null && !readProperties.isEmpty()) { + localFilesBuilder.putAllReadProperties(readProperties); + } return localFilesBuilder.build(); } } diff --git a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto index e2396912085..e683ab8ef92 100644 --- a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto +++ b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto @@ -111,6 +111,12 @@ message ReadRel { repeated FileOrFiles items = 1; substrait.extensions.AdvancedExtension advanced_extension = 10; + // Table-scoped storage properties the native reader needs to open the + // listed files, e.g. the S3 credentials an Iceberg REST catalog vends for + // one table. A LocalFiles is single-table by construction, so table + // granularity is split granularity. Emitted only when present. + map read_properties = 11; + // Many files consist of indivisible chunks (e.g. parquet row groups // or CSV rows). If a slice partially selects an indivisible chunk // then the consumer should employ some rule to decide which slice to diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala index cf2026af630..ac3d4a7c63c 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala @@ -154,6 +154,12 @@ trait BackendSettingsApi { def supportIcebergInitialDefaultRead(): Boolean = false + /** + * Whether the backend reads the files of an Iceberg table with the credentials the catalog vended + * for it, carried in LocalFiles.read_properties. + */ + def supportIcebergVendedCredentialsRead(): Boolean = false + def reorderColumnsForPartitionWrite(): Boolean = false def enableEnhancedFeatures(): Boolean = false From 557f86fc16d8ce03cdef881c5b4bc310d5763e49 Mon Sep 17 00:00:00 2001 From: Guangyu Yang Date: Wed, 19 Aug 2026 12:03:21 -0400 Subject: [PATCH 2/2] [MINOR] Build CI against the pending Velox PR (revert before merge) The native half of the previous commit needs the S3 file system to consume FileOptions::tokenProvider, which is facebookincubator/velox#18570 and not yet in the pinned Velox branch. Point UPSTREAM_VELOX_PR_ID at it so CI can build this PR, and drop this commit once the Velox change is merged and the pin is advanced past it. --- ep/build-velox/src/get-velox.sh | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh index 8713a3b8ae0..9e31b545690 100755 --- a/ep/build-velox/src/get-velox.sh +++ b/ep/build-velox/src/get-velox.sh @@ -25,7 +25,9 @@ RUN_SETUP_SCRIPT=ON ENABLE_ENHANCED_FEATURES=OFF # Developer use only for testing Velox PR. -UPSTREAM_VELOX_PR_ID="" +# TODO: reset to "" once facebookincubator/velox#18570 (per-path S3 credentials +# from the query TokenProvider) is merged and the Velox pin above includes it. +UPSTREAM_VELOX_PR_ID="18570" OS=`uname -s`