feat: S3 credential vending for REST catalog

Implement per-table credential vending in RestCatalog, following the pattern
from iceberg-java's RESTSessionCatalog and PyIceberg's REST catalog.

- Set X-Iceberg-Access-Delegation header at session level on HttpClient
- ResolveTableFileIO merges catalog properties with per-table config and
  storage credentials (longest-prefix match), creating per-table FileIO
  via FileIORegistry when credentials are vended
- StorageCredential struct with JSON serde and prefix-based resolution
- Applied consistently to LoadTable, CreateTable, RegisterTable, StageCreateTable
- Table exposes io_properties() for consumers that need vended configuration
- Table::Refresh() updates io_properties_ to prevent stale credentials
- Comprehensive unit tests (property merging, storage credential resolution,
  io_properties lifecycle) and REST integration tests

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Sreesh Maheshwar

src/iceberg/catalog/rest/constant.h

@@ -38,6 +38,9 @@ inline const std::string kMimeTypeFormUrlEncoded = "application/x-www-form-urlen
 inline const std::string kUserAgentPrefix = "iceberg-cpp/";
 inline const std::string kUserAgent = "iceberg-cpp/" ICEBERG_VERSION_STRING;
 
+inline const std::string kHeaderAccessDelegation = "X-Iceberg-Access-Delegation";
+inline const std::string kAccessDelegationVendedCredentials = "vended-credentials";
+
 inline const std::string kQueryParamParent = "parent";
 inline const std::string kQueryParamPageToken = "page_token";
 

src/iceberg/catalog/rest/json_serde.cc

@@ -72,6 +72,8 @@ constexpr std::string_view kStack = "stack";
 constexpr std::string_view kError = "error";
 constexpr std::string_view kIdentifier = "identifier";
 constexpr std::string_view kRequirements = "requirements";
+constexpr std::string_view kStorageCredentials = "storage-credentials";
+constexpr std::string_view kPrefix = "prefix";
 constexpr std::string_view kAccessToken = "access_token";
 constexpr std::string_view kTokenType = "token_type";
 constexpr std::string_view kExpiresIn = "expires_in";
@@ -217,12 +219,35 @@ Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json
   return request;
 }
 
+// StorageCredential serialization
+nlohmann::json ToJson(const StorageCredential& cred) {
+  nlohmann::json json;
+  json[kPrefix] = cred.prefix;
+  json[kConfig] = cred.config;
+  return json;
+}
+
+Result<StorageCredential> StorageCredentialFromJson(const nlohmann::json& json) {
+  StorageCredential cred;
+  ICEBERG_ASSIGN_OR_RAISE(cred.prefix, GetJsonValue<std::string>(json, kPrefix));
+  ICEBERG_ASSIGN_OR_RAISE(cred.config,
+                          GetJsonValueOrDefault<decltype(cred.config)>(json, kConfig));
+  return cred;
+}
+
 // LoadTableResult (used by CreateTableResponse, LoadTableResponse)
 nlohmann::json ToJson(const LoadTableResult& result) {
   nlohmann::json json;
   SetOptionalStringField(json, kMetadataLocation, result.metadata_location);
   json[kMetadata] = ToJson(*result.metadata);
   SetContainerField(json, kConfig, result.config);
+  if (!result.storage_credentials.empty()) {
+    nlohmann::json creds_json = nlohmann::json::array();
+    for (const auto& cred : result.storage_credentials) {
+      creds_json.push_back(ToJson(cred));
+    }
+    json[kStorageCredentials] = std::move(creds_json);
+  }
   return json;
 }
 
@@ -235,6 +260,14 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
   ICEBERG_ASSIGN_OR_RAISE(result.config,
                           GetJsonValueOrDefault<decltype(result.config)>(json, kConfig));
+  if (json.contains(kStorageCredentials)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto creds_json,
+                            GetJsonValue<nlohmann::json>(json, kStorageCredentials));
+    for (const auto& cred_json : creds_json) {
+      ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(cred_json));
+      result.storage_credentials.push_back(std::move(cred));
+    }
+  }
   ICEBERG_RETURN_UNEXPECTED(result.Validate());
   return result;
 }
@@ -528,5 +561,6 @@ ICEBERG_DEFINE_FROM_JSON(CreateTableRequest)
 ICEBERG_DEFINE_FROM_JSON(CommitTableRequest)
 ICEBERG_DEFINE_FROM_JSON(CommitTableResponse)
 ICEBERG_DEFINE_FROM_JSON(OAuthTokenResponse)
+ICEBERG_DEFINE_FROM_JSON(StorageCredential)
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/json_serde_internal.h

@@ -59,6 +59,7 @@ ICEBERG_DECLARE_JSON_SERDE(CreateTableRequest)
 ICEBERG_DECLARE_JSON_SERDE(CommitTableRequest)
 ICEBERG_DECLARE_JSON_SERDE(CommitTableResponse)
 ICEBERG_DECLARE_JSON_SERDE(OAuthTokenResponse)
+ICEBERG_DECLARE_JSON_SERDE(StorageCredential)
 
 #undef ICEBERG_DECLARE_JSON_SERDE
 

src/iceberg/catalog/rest/rest_catalog.cc

@@ -35,6 +35,7 @@
 #include "iceberg/catalog/rest/json_serde_internal.h"
 #include "iceberg/catalog/rest/resource_paths.h"
 #include "iceberg/catalog/rest/rest_file_io.h"
+#include "iceberg/catalog/rest/rest_file_io_internal.h"
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
@@ -166,7 +167,9 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
   // Get snapshot loading mode
   ICEBERG_ASSIGN_OR_RAISE(auto snapshot_mode, final_config.SnapshotLoadingMode());
 
-  auto client = std::make_unique<HttpClient>(final_config.ExtractHeaders());
+  auto default_headers = final_config.ExtractHeaders();
+  default_headers.emplace(kHeaderAccessDelegation, kAccessDelegationVendedCredentials);
+  auto client = std::make_unique<HttpClient>(std::move(default_headers));
   ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
                           auth_manager->CatalogSession(*client, final_config.configs()));
 
@@ -361,8 +364,13 @@ Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
   ICEBERG_ASSIGN_OR_RAISE(auto result,
                           CreateTableInternal(identifier, schema, spec, order, location,
                                               properties, /*stage_create=*/false));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto table_io,
+      ResolveTableFileIO(file_io_, config_.configs(),
+                         config_.Get(RestCatalogProperties::kWarehouse), result));
   return Table::Make(identifier, std::move(result.metadata),
-                     std::move(result.metadata_location), file_io_, shared_from_this());
+                     std::move(result.metadata_location), std::move(table_io),
+                     shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
@@ -404,10 +412,14 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
   ICEBERG_ASSIGN_OR_RAISE(auto result,
                           CreateTableInternal(identifier, schema, spec, order, location,
                                               properties, /*stage_create=*/true));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto table_io,
+      ResolveTableFileIO(file_io_, config_.configs(),
+                         config_.Get(RestCatalogProperties::kWarehouse), result));
   ICEBERG_ASSIGN_OR_RAISE(auto staged_table,
                           StagedTable::Make(identifier, std::move(result.metadata),
-                                            std::move(result.metadata_location), file_io_,
-                                            shared_from_this()));
+                                            std::move(result.metadata_location),
+                                            std::move(table_io), shared_from_this()));
   return Transaction::Make(std::move(staged_table), TransactionKind::kCreate);
 }
 
@@ -474,9 +486,13 @@ Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& ide
   ICEBERG_ASSIGN_OR_RAISE(const auto body, LoadTableInternal(identifier));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(body));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
-  /// FIXME: support per-table FileIO creation
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto table_io,
+      ResolveTableFileIO(file_io_, config_.configs(),
+                         config_.Get(RestCatalogProperties::kWarehouse), 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(table_io),
                      shared_from_this());
 }
 
@@ -498,8 +514,12 @@ 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));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto table_io,
+      ResolveTableFileIO(file_io_, config_.configs(),
+                         config_.Get(RestCatalogProperties::kWarehouse), 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(table_io),
                      shared_from_this());
 }
 

src/iceberg/catalog/rest/rest_file_io.cc

@@ -20,7 +20,10 @@
 #include "iceberg/catalog/rest/rest_file_io.h"
 
 #include <string>
+#include <unordered_map>
 
+#include "iceberg/catalog/rest/catalog_properties.h"
+#include "iceberg/catalog/rest/rest_file_io_internal.h"
 #include "iceberg/file_io_registry.h"
 #include "iceberg/util/macros.h"
 
@@ -69,12 +72,14 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
 
   if (io_impl.empty()) {
     if (warehouse.empty()) {
-      return InvalidArgument(R"("{}" or "{}" property is required to create FileIO)",
-                             RestCatalogProperties::kIOImpl.key(),
-                             RestCatalogProperties::kWarehouse.key());
+      // No io-impl or warehouse configured. Fall back to a local FileIO as a
+      // default — the per-table ResolveTableFileIO will resolve the real FileIO
+      // when a table is loaded (using the table's location to detect S3, etc.).
+      io_impl = std::string(FileIORegistry::kArrowLocalFileIO);
+    } else {
+      ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse));
+      io_impl = std::string(BuiltinFileIOName(detected_kind));
     }
-    ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse));
-    io_impl = std::string(BuiltinFileIOName(detected_kind));
   }
 
   if (!warehouse.empty() && IsBuiltinImpl(io_impl)) {
@@ -92,4 +97,73 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
   return FileIORegistry::Load(io_impl, config.configs());
 }
 
+namespace {
+
+const StorageCredential* ResolveStorageCredential(
+    const std::vector<StorageCredential>& credentials, std::string_view location) {
+  const StorageCredential* best = nullptr;
+  for (const auto& cred : credentials) {
+    if (location.starts_with(cred.prefix)) {
+      if (!best || cred.prefix.size() > best->prefix.size()) {
+        best = &cred;
+      }
+    }
+  }
+  return best;
+}
+
+std::unordered_map<std::string, std::string> MergeTableProperties(
+    const std::unordered_map<std::string, std::string>& catalog_props,
+    const std::unordered_map<std::string, std::string>& table_config,
+    const std::unordered_map<std::string, std::string>& credential_config) {
+  auto merged = catalog_props;
+  for (const auto& [k, v] : table_config) {
+    merged[k] = v;
+  }
+  for (const auto& [k, v] : credential_config) {
+    merged[k] = v;
+  }
+  return merged;
+}
+
+}  // namespace
+
+Result<std::shared_ptr<FileIO>> ResolveTableFileIO(
+    const std::shared_ptr<FileIO>& catalog_io,
+    const std::unordered_map<std::string, std::string>& catalog_props,
+    const std::string& warehouse, const LoadTableResult& result) {
+  if (result.config.empty() && result.storage_credentials.empty()) {
+    return catalog_io;
+  }
+
+  // TODO: Python also merges result.metadata->properties into the chain here
+  // (metadata props < catalog < table config < credentials). Java does not.
+  // Consider adding this for parity with Python if needed.
+  const StorageCredential* cred = nullptr;
+  if (!result.metadata_location.empty()) {
+    cred = ResolveStorageCredential(result.storage_credentials, result.metadata_location);
+  }
+  static const std::unordered_map<std::string, std::string> kEmpty;
+  auto merged =
+      MergeTableProperties(catalog_props, result.config, cred ? cred->config : kEmpty);
+
+  // Detect FileIO type: explicit io-impl > warehouse scheme > metadata_location scheme.
+  std::string io_impl;
+  if (auto it = merged.find(std::string(RestCatalogProperties::kIOImpl.key()));
+      it != merged.end()) {
+    io_impl = it->second;
+  } else if (!warehouse.empty()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto kind, DetectBuiltinFileIO(warehouse));
+    io_impl = std::string(BuiltinFileIOName(kind));
+  } else if (!result.metadata_location.empty()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto kind, DetectBuiltinFileIO(result.metadata_location));
+    io_impl = std::string(BuiltinFileIOName(kind));
+  } else {
+    return catalog_io;
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, FileIORegistry::Load(io_impl, merged));
+  return std::shared_ptr<FileIO>(std::move(table_io));
+}
+
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_file_io_internal.h

@@ -0,0 +1,47 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/catalog/rest/iceberg_rest_export.h"
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/file_io.h"
+#include "iceberg/result.h"
+
+/// \file iceberg/catalog/rest/rest_file_io_internal.h
+/// Internal helpers for per-table FileIO resolution in the REST catalog.
+
+namespace iceberg::rest {
+
+/// \brief Resolve a per-table FileIO from a LoadTableResult.
+///
+/// Merges catalog properties, table config, and the best-matching storage
+/// credential (longest prefix match on metadata_location), then creates a
+/// per-table FileIO via FileIORegistry.  Falls back to \p catalog_io when
+/// no per-table config is present or the FileIO type cannot be detected.
+ICEBERG_REST_EXPORT Result<std::shared_ptr<FileIO>> ResolveTableFileIO(
+    const std::shared_ptr<FileIO>& catalog_io,
+    const std::unordered_map<std::string, std::string>& catalog_props,
+    const std::string& warehouse, const LoadTableResult& result);
+
+}  // namespace iceberg::rest

src/iceberg/catalog/rest/types.h

@@ -169,12 +169,25 @@ struct ICEBERG_REST_EXPORT CreateTableRequest {
 /// \brief An opaque token that allows clients to make use of pagination for list APIs.
 using PageToken = std::string;
 
+/// \brief A storage credential returned by the REST catalog server.
+///
+/// Each credential has a prefix (a location prefix like "s3://bucket/path") and
+/// a config map containing cloud-provider-specific properties (e.g., temporary
+/// access keys). Clients resolve the best credential via longest prefix match
+/// against the table's metadata location.
+struct ICEBERG_REST_EXPORT StorageCredential {
+  std::string prefix;
+  std::unordered_map<std::string, std::string> config;
+
+  bool operator==(const StorageCredential&) const = default;
+};
+
 /// \brief Result body for table create/load/register APIs.
 struct ICEBERG_REST_EXPORT LoadTableResult {
   std::string metadata_location;
   std::shared_ptr<TableMetadata> metadata;  // required
   std::unordered_map<std::string, std::string> config;
-  // TODO(Li Feiyang): Add std::shared_ptr<StorageCredential> storage_credential;
+  std::vector<StorageCredential> storage_credentials;
 
   /// \brief Validates the LoadTableResult.
   Status Validate() const {

src/iceberg/file_io.h

@@ -22,6 +22,7 @@
 #include <optional>
 #include <string>
 #include <string_view>
+#include <unordered_map>
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
@@ -42,6 +43,14 @@ class ICEBERG_EXPORT FileIO {
   FileIO() = default;
   virtual ~FileIO() = default;
 
+  /// \brief Returns the configuration properties used to initialize this FileIO.
+  ///
+  /// Engines that need to configure their own storage access (e.g., for credential
+  /// vending) can read these properties to obtain the resolved credentials.
+  const std::unordered_map<std::string, std::string>& properties() const {
+    return properties_;
+  }
+
   /// \brief Read the content of the file at the given location.
   ///
   /// \param file_location The location of the file to read.
@@ -73,6 +82,10 @@ class ICEBERG_EXPORT FileIO {
   virtual Status DeleteFile(const std::string& file_location) {
     return NotImplemented("DeleteFile not implemented");
   }
+
+ private:
+  friend class FileIORegistry;
+  std::unordered_map<std::string, std::string> properties_;
 };
 
 }  // namespace iceberg

src/iceberg/file_io_registry.cc

@@ -20,6 +20,8 @@
 #include <mutex>
 #include <utility>
 
+#include "iceberg/util/macros.h"
+
 namespace iceberg {
 
 namespace {
@@ -55,7 +57,9 @@ Result<std::unique_ptr<FileIO>> FileIORegistry::Load(
     }
     factory = it->second;
   }
-  return factory(properties);
+  ICEBERG_ASSIGN_OR_RAISE(auto io, factory(properties));
+  io->properties_ = properties;
+  return io;
 }
 
 }  // namespace iceberg