[GLUTEN-12701][CORE][VL] Support Iceberg REST-catalog vended credentials in native scans

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()

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()
 

cpp/velox/compute/VeloxPlanConverter.cc

@@ -149,6 +149,9 @@ std::shared_ptr<SplitInfo> 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();

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<velox::core::QueryCtx> 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<velox::filesystems::TokenProvider> WholeStageResultIterator::createFsTokenProvider() const {
+#ifdef ENABLE_S3
+  std::vector<std::unordered_map<std::string, std::string>> 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<ColumnarBatch> WholeStageResultIterator::next() {
   while (true) {
     if (!cursor_->moveNext()) {

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<facebook::velox::core::QueryCtx> 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<facebook::velox::filesystems::TokenProvider> createFsTokenProvider() const;
+
   /// Get all the children plan node ids with postorder traversal.
   void getOrderedNodeIds(
       const std::shared_ptr<const facebook::velox::core::PlanNode>&,

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<std::optional<facebook::velox::FileProperties>> 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<std::string, std::string> readProperties;
+
   /// The schema of the table being scanned.
   RowTypePtr tableSchema;
 

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)

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 <gtest/gtest.h>
+
+#include "velox/common/file/PlainUserNameTokenProvider.h"
+
+namespace gluten {
+namespace {
+
+using namespace facebook::velox::filesystems;
+
+std::unordered_map<std::string, std::string> readProperties(
+    const std::string& location,
+    const std::string& accessKeyId,
+    const std::string& secretAccessKey,
+    const std::string& sessionToken = "") {
+  std::unordered_map<std::string, std::string> properties{
+      {kReadPropertiesLocation, location},
+      {kReadPropertiesAccessKeyId, accessKeyId},
+      {kReadPropertiesSecretAccessKey, secretAccessKey}};
+  if (!sessionToken.empty()) {
+    properties[kReadPropertiesSessionToken] = sessionToken;
+  }
+  return properties;
+}
+
+std::shared_ptr<S3AccessToken> tokenFor(const TokenProvider& provider, const std::string& path) {
+  return std::dynamic_pointer_cast<S3AccessToken>(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

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 <cstring>
+#include <functional>
+
+#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<std::string, std::string>& properties, const char* key) {
+  const auto it = properties.find(key);
+  return it == properties.end() ? "" : it->second;
+}
+
+} // namespace
+
+GlutenS3TokenProvider::GlutenS3TokenProvider(std::map<std::string, S3TableCredentials> credentialsByPrefix)
+    : credentialsByPrefix_(std::move(credentialsByPrefix)) {
+  const std::hash<std::string> 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> GlutenS3TokenProvider::create(
+    const std::vector<std::unordered_map<std::string, std::string>>& readProperties) {
+  std::map<std::string, S3TableCredentials> 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<GlutenS3TokenProvider>(std::move(credentialsByPrefix));
+}
+
+bool GlutenS3TokenProvider::equals(const facebook::velox::filesystems::TokenProvider& other) const {
+  const auto* typedOther = dynamic_cast<const GlutenS3TokenProvider*>(&other);
+  return typedOther != nullptr && credentialsByPrefix_ == typedOther->credentialsByPrefix_;
+}
+
+size_t GlutenS3TokenProvider::hash() const {
+  return hash_;
+}
+
+std::shared_ptr<facebook::velox::filesystems::AccessToken> GlutenS3TokenProvider::getToken(
+    const facebook::velox::filesystems::AccessTokenKey& key) const {
+  const auto* s3Key = dynamic_cast<const facebook::velox::filesystems::S3AccessTokenKey*>(&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<facebook::velox::filesystems::S3AccessToken>(
+      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