feat(rest): bind a per-table FileIO from vended credentials

Author: plusplusjiajia

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -100,15 +100,25 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
   }
 
   // Configure region
-  if (const auto* region = FindProperty(properties, S3Properties::kRegion);
-      region != nullptr) {
+  // Prefer the standard `client.region`; fall back to legacy `s3.region`.
+  const auto* region = FindProperty(properties, S3Properties::kClientRegion);
+  if (region == nullptr) {
+    region = FindProperty(properties, S3Properties::kRegion);
+  }
+  if (region != nullptr) {
     options.region = *region;
   }
 
-  // Configure endpoint (for MinIO, LocalStack, etc.)
+  // Configure endpoint (for MinIO, LocalStack, OSS, etc.)
   if (const auto* endpoint = FindProperty(properties, S3Properties::kEndpoint);
       endpoint != nullptr) {
-    options.endpoint_override = *endpoint;
+    // `s3.endpoint` may be a full URI; Arrow wants host[:port], so strip the scheme.
+    std::string_view ep = *endpoint;
+    if (const auto pos = ep.find("://"); pos != std::string_view::npos) {
+      options.scheme = std::string(ep.substr(0, pos));
+      ep = ep.substr(pos + 3);
+    }
+    options.endpoint_override = std::string(ep);
   } else {
     // Fall back to AWS standard environment variables for endpoint override
     const char* s3_endpoint_env = std::getenv("AWS_ENDPOINT_URL_S3");

src/iceberg/arrow/s3/s3_properties.h

@@ -37,8 +37,10 @@ struct S3Properties {
   static constexpr std::string_view kSecretAccessKey = "s3.secret-access-key";
   /// AWS session token (for temporary credentials)
   static constexpr std::string_view kSessionToken = "s3.session-token";
-  /// AWS region
+  /// AWS region (legacy, non-standard key kept for compatibility)
   static constexpr std::string_view kRegion = "s3.region";
+  /// AWS region, standard Iceberg client property (preferred over kRegion).
+  static constexpr std::string_view kClientRegion = "client.region";
   /// Custom endpoint override (for MinIO, LocalStack, etc.)
   static constexpr std::string_view kEndpoint = "s3.endpoint";
   /// Whether to use path-style access (needed for MinIO)

src/iceberg/catalog/rest/rest_catalog.cc

@@ -38,6 +38,7 @@
 #include "iceberg/catalog/rest/rest_file_io.h"
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
+#include "iceberg/file_io_registry.h"
 #include "iceberg/json_serde_internal.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
@@ -486,13 +487,36 @@ Result<LoadTableResult> RestCatalog::LoadTableResultFor(
   return LoadTableResultFromJson(json);
 }
 
+Result<std::shared_ptr<FileIO>> RestCatalog::MakeFileIOForTable(
+    const LoadTableResult& load_result) const {
+  // Longest-prefix S3-family vended credential, matching the Java client's
+  // VendedCredentialsProvider.
+  const StorageCredential* s3_cred =
+      SelectS3StorageCredential(load_result.storage_credentials);
+  if (s3_cred == nullptr) {
+    return file_io_;
+  }
+
+  // Build an S3 FileIO from catalog + table config + the vended credential; its
+  // ResolvePath resolves oss:// and other S3-compatible schemes.
+  auto properties = config_.configs();
+  for (const auto& [key, value] : load_result.config) {
+    properties[key] = value;
+  }
+  for (const auto& [key, value] : s3_cred->config) {
+    properties[key] = value;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto io,
+      FileIORegistry::Load(std::string(FileIORegistry::kArrowS3FileIO), properties));
+  return std::shared_ptr<FileIO>(std::move(io));
+}
+
 Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& identifier) {
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFor(identifier));
-  // Binds the catalog's default FileIO; use LoadTableResultFor() for vended creds.
-  // TODO: Support table-specific auth config and per-table FileIO from the REST
-  // load response when table-scoped REST operations are introduced.
+  ICEBERG_ASSIGN_OR_RAISE(auto io, MakeFileIOForTable(load_result));
   return Table::Make(identifier, std::move(load_result.metadata),
-                     std::move(load_result.metadata_location), file_io_,
+                     std::move(load_result.metadata_location), std::move(io),
                      shared_from_this());
 }
 
@@ -514,10 +538,9 @@ Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
-  // TODO: Support table-specific auth config and per-table FileIO from the REST
-  // register response when table-scoped REST operations are introduced.
+  ICEBERG_ASSIGN_OR_RAISE(auto io, MakeFileIOForTable(load_result));
   return Table::Make(identifier, std::move(load_result.metadata),
-                     std::move(load_result.metadata_location), file_io_,
+                     std::move(load_result.metadata_location), std::move(io),
                      shared_from_this());
 }
 

src/iceberg/catalog/rest/rest_catalog.h

@@ -113,6 +113,11 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
 
+  /// \brief Builds a per-table FileIO from the load response config and vended
+  /// `storage-credentials`, or the catalog default FileIO when none apply.
+  Result<std::shared_ptr<FileIO>> MakeFileIOForTable(
+      const LoadTableResult& load_result) const;
+
   Result<LoadTableResult> CreateTableInternal(
       const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
       const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,

src/iceberg/catalog/rest/rest_file_io.cc

@@ -20,7 +20,9 @@
 #include "iceberg/catalog/rest/rest_file_io.h"
 
 #include <string>
+#include <vector>
 
+#include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io_registry.h"
 #include "iceberg/util/macros.h"
 
@@ -92,4 +94,16 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
   return FileIORegistry::Load(io_impl, config.configs());
 }
 
+const StorageCredential* SelectS3StorageCredential(
+    const std::vector<StorageCredential>& credentials) {
+  const StorageCredential* best = nullptr;
+  for (const auto& cred : credentials) {
+    if (cred.prefix.starts_with("s3") &&
+        (best == nullptr || cred.prefix.size() > best->prefix.size())) {
+      best = &cred;
+    }
+  }
+  return best;
+}
+
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_file_io.h

@@ -22,9 +22,11 @@
 #include <cstdint>
 #include <memory>
 #include <string_view>
+#include <vector>
 
 #include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
+#include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
 #include "iceberg/result.h"
@@ -44,4 +46,9 @@ ICEBERG_REST_EXPORT std::string_view BuiltinFileIOName(BuiltinFileIOKind kind);
 ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
     const RestCatalogProperties& config);
 
+/// \brief Returns the longest-prefix S3-family vended credential (prefix
+/// starting with "s3"), or nullptr if none.
+ICEBERG_REST_EXPORT const StorageCredential* SelectS3StorageCredential(
+    const std::vector<StorageCredential>& credentials);
+
 }  // namespace iceberg::rest

src/iceberg/test/rest_file_io_test.cc

@@ -19,9 +19,12 @@
 
 #include "iceberg/catalog/rest/rest_file_io.h"
 
+#include <vector>
+
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
+#include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io_registry.h"
 #include "iceberg/test/matchers.h"
 
@@ -143,4 +146,33 @@ TEST(RestFileIOTest, MakeCatalogFileIOSkipsCheckWhenWarehouseAbsent) {
   ASSERT_THAT(result, IsOk());
 }
 
+TEST(RestFileIOTest, SelectS3StorageCredentialPicksLongestS3Prefix) {
+  std::vector<StorageCredential> credentials = {
+      {.prefix = "s3", .config = {{"s3.access-key-id", "a"}}},
+      {.prefix = "s3://bucket/data", .config = {{"s3.access-key-id", "b"}}},
+      {.prefix = "s3://bucket", .config = {{"s3.access-key-id", "c"}}},
+  };
+  const auto* cred = SelectS3StorageCredential(credentials);
+  ASSERT_NE(cred, nullptr);
+  EXPECT_EQ(cred->prefix, "s3://bucket/data");
+}
+
+TEST(RestFileIOTest, SelectS3StorageCredentialIgnoresNonS3Prefixes) {
+  std::vector<StorageCredential> credentials = {
+      {.prefix = "gs://bucket", .config = {{"k", "v"}}},
+      {.prefix = "s3", .config = {{"s3.access-key-id", "a"}}},
+  };
+  const auto* cred = SelectS3StorageCredential(credentials);
+  ASSERT_NE(cred, nullptr);
+  EXPECT_EQ(cred->prefix, "s3");
+}
+
+TEST(RestFileIOTest, SelectS3StorageCredentialReturnsNullWhenNoneMatch) {
+  std::vector<StorageCredential> credentials = {
+      {.prefix = "gs://bucket", .config = {{"k", "v"}}},
+  };
+  EXPECT_EQ(SelectS3StorageCredential(credentials), nullptr);
+  EXPECT_EQ(SelectS3StorageCredential({}), nullptr);
+}
+
 }  // namespace iceberg::rest