This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 79fdb868 feat(rest): load table credentials from the LoadCredentials
endpoint (#899)
79fdb868 is described below
commit 79fdb86853450397f64a4bc22ad93d1cf0eb44c9
Author: Jiajia Li <[email protected]>
AuthorDate: Thu Oct 8 23:46:13 2026 +0800
feat(rest): load table credentials from the LoadCredentials endpoint (#899)
---
src/iceberg/catalog/rest/json_serde.cc | 42 ++++++++++---
src/iceberg/catalog/rest/json_serde_internal.h | 4 ++
src/iceberg/catalog/rest/rest_catalog.cc | 87 +++++++++++++++++++++++---
src/iceberg/catalog/rest/rest_catalog.h | 12 +++-
src/iceberg/catalog/rest/rest_file_io.cc | 16 ++++-
src/iceberg/catalog/rest/rest_file_io.h | 3 +-
src/iceberg/catalog/rest/types.h | 15 +++++
src/iceberg/file_io.h | 13 ++++
src/iceberg/resolving_file_io.cc | 34 ++++++++--
src/iceberg/resolving_file_io.h | 14 ++++-
src/iceberg/storage_credential.h | 12 ++++
src/iceberg/test/resolving_file_io_test.cc | 65 +++++++++++++++++++
src/iceberg/test/rest_file_io_test.cc | 48 ++++++++++++++
src/iceberg/test/rest_json_serde_test.cc | 26 ++++++++
14 files changed, 362 insertions(+), 29 deletions(-)
diff --git a/src/iceberg/catalog/rest/json_serde.cc
b/src/iceberg/catalog/rest/json_serde.cc
index 3ce753f1..098f2ba0 100644
--- a/src/iceberg/catalog/rest/json_serde.cc
+++ b/src/iceberg/catalog/rest/json_serde.cc
@@ -152,6 +152,25 @@ Result<StorageCredential> StorageCredentialFromJson(const
nlohmann::json& json)
return credential;
}
+/// \brief Reads the optional `storage-credentials` array shared by the
+/// LoadTable and LoadCredentials responses.
+Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
+ const nlohmann::json& json) {
+ std::vector<StorageCredential> credentials;
+ auto it = json.find(kStorageCredentials);
+ if (it == json.end() || it->is_null()) {
+ return credentials;
+ }
+ if (!it->is_array()) {
+ return JsonParseError("Cannot parse storage credentials from non-array");
+ }
+ for (const auto& entry : *it) {
+ ICEBERG_ASSIGN_OR_RAISE(auto credential, StorageCredentialFromJson(entry));
+ credentials.push_back(std::move(credential));
+ }
+ return credentials;
+}
+
template <typename Value>
Result<std::map<int32_t, Value>> KeyValueMapFromJson(const nlohmann::json&
json,
std::string_view key) {
@@ -738,19 +757,24 @@ 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 (auto it = json.find(kStorageCredentials); it != json.end() &&
!it->is_null()) {
- if (!it->is_array()) {
- return JsonParseError("Cannot parse storage credentials from non-array");
- }
- for (const auto& entry : *it) {
- ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(entry));
- result.storage_credentials.push_back(std::move(cred));
- }
- }
+ ICEBERG_ASSIGN_OR_RAISE(result.storage_credentials,
StorageCredentialsFromJson(json));
ICEBERG_RETURN_UNEXPECTED(result.Validate());
return result;
}
+Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
+ const nlohmann::json& json) {
+ // Required here, unlike in LoadTable: reading a malformed response as "no
+ // credentials" would look like a refresh that succeeded and dropped them.
+ if (auto it = json.find(kStorageCredentials); it == json.end() ||
it->is_null()) {
+ return JsonParseError("Missing '{}'", kStorageCredentials);
+ }
+ LoadCredentialsResponse response;
+ ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials,
StorageCredentialsFromJson(json));
+ ICEBERG_RETURN_UNEXPECTED(response.Validate());
+ return response;
+}
+
nlohmann::json ToJson(const ListNamespacesResponse& response) {
nlohmann::json json;
SetOptionalStringField(json, kNextPageToken, response.next_page_token);
diff --git a/src/iceberg/catalog/rest/json_serde_internal.h
b/src/iceberg/catalog/rest/json_serde_internal.h
index 6e148e0d..aedd2cd1 100644
--- a/src/iceberg/catalog/rest/json_serde_internal.h
+++ b/src/iceberg/catalog/rest/json_serde_internal.h
@@ -71,6 +71,10 @@ template <>
ICEBERG_REST_EXPORT Result<LoadTableResult> FromJson(const nlohmann::json&
json);
ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const LoadTableResult&
model);
+// Response-only model: a client never serializes it, so no ToJson.
+ICEBERG_REST_EXPORT Result<LoadCredentialsResponse>
LoadCredentialsResponseFromJson(
+ const nlohmann::json& json);
+
ICEBERG_REST_EXPORT Result<CreateTableRequest> CreateTableRequestFromJson(
const nlohmann::json& json);
template <>
diff --git a/src/iceberg/catalog/rest/rest_catalog.cc
b/src/iceberg/catalog/rest/rest_catalog.cc
index 9635aebe..e2aaf5f3 100644
--- a/src/iceberg/catalog/rest/rest_catalog.cc
+++ b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -20,6 +20,8 @@
#include "iceberg/catalog/rest/rest_catalog.h"
#include <memory>
+#include <mutex>
+#include <string>
#include <string_view>
#include <tuple>
#include <unordered_map>
@@ -42,6 +44,7 @@
#include "iceberg/catalog/rest/rest_util.h"
#include "iceberg/catalog/rest/types.h"
#include "iceberg/json_serde_internal.h"
+#include "iceberg/logging/log_macros.h"
#include "iceberg/metrics/metrics_reporters.h"
#include "iceberg/partition_spec.h"
#include "iceberg/result.h"
@@ -60,6 +63,41 @@ namespace iceberg::rest {
namespace {
+class RestStorageCredentialProvider final : public StorageCredentialProvider {
+ public:
+ RestStorageCredentialProvider(std::shared_ptr<HttpClient> client,
+ std::shared_ptr<auth::AuthSession> session,
+ std::string path)
+ : client_(std::move(client)),
+ session_(std::move(session)),
+ path_(std::move(path)) {}
+
+ Result<std::vector<StorageCredential>> Load() override {
+ // The provider can be shared by several FileIOs; keep fetches serialized.
+ std::lock_guard lock(mutex_);
+ ICEBERG_ASSIGN_OR_RAISE(const auto response,
+ client_->Get(path_, /*params=*/{}, /*headers=*/{},
+ *TableErrorHandler::Instance(),
*session_));
+ // Parse errors can contain credential data; return a fixed message.
+ auto json = FromJsonString(response.body());
+ if (!json.has_value()) {
+ return JsonParseError("Malformed LoadCredentials response");
+ }
+ auto result = LoadCredentialsResponseFromJson(*json);
+ if (!result.has_value()) {
+ return std::unexpected<Error>(
+ {.kind = result.error().kind, .message = "Malformed LoadCredentials
response"});
+ }
+ return std::move(result->storage_credentials);
+ }
+
+ private:
+ std::shared_ptr<HttpClient> client_;
+ std::shared_ptr<auth::AuthSession> session_;
+ std::string path_;
+ std::mutex mutex_;
+};
+
/// \brief Get the default set of endpoints for backwards compatibility
according to the
/// iceberg rest spec.
std::unordered_set<Endpoint> GetDefaultEndpoints() {
@@ -508,12 +546,45 @@ Result<std::shared_ptr<auth::AuthSession>>
RestCatalog::TableAuthSession(
std::move(contextual_session));
}
+std::shared_ptr<StorageCredentialProvider>
RestCatalog::MakeStorageCredentialProvider(
+ const TableIdentifier& identifier,
+ std::shared_ptr<auth::AuthSession> table_session) const {
+ if (!supported_endpoints_.contains(Endpoint::TableCredentials())) {
+ // Not an error, but it surfaces much later as credentials expiring.
+ ICEBERG_LOG_DEBUG(
+ "Catalog does not advertise {}; vended credentials for '{}' will not
be "
+ "refreshed",
+ Endpoint::TableCredentials().ToString(), ToString(identifier));
+ return nullptr;
+ }
+ auto path = paths_->Credentials(identifier);
+ if (!path.has_value()) {
+ ICEBERG_LOG_WARN(
+ "Cannot build the credentials path for '{}' ({}); its vended
credentials "
+ "will not be refreshed",
+ ToString(identifier), path.error().message);
+ return nullptr;
+ }
+ auto client = client_;
+ auto credentials_path = std::move(path.value());
+ auto session = std::move(table_session);
+ return std::make_shared<RestStorageCredentialProvider>(
+ std::move(client), std::move(session), std::move(credentials_path));
+}
+
Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO(
- const SessionContext& /*context*/,
+ const SessionContext& /*context*/, const TableIdentifier& identifier,
const std::unordered_map<std::string, std::string>& table_config,
- const std::vector<StorageCredential>& storage_credentials) const {
+ const std::vector<StorageCredential>& storage_credentials,
+ std::shared_ptr<auth::AuthSession> table_session) const {
if (!table_config.empty() || !storage_credentials.empty()) {
- return MakeTableFileIO(config_.configs(), table_config,
storage_credentials);
+ // Only vended credentials expire, so only they need a provider.
+ std::shared_ptr<StorageCredentialProvider> provider;
+ if (!storage_credentials.empty()) {
+ provider = MakeStorageCredentialProvider(identifier,
std::move(table_session));
+ }
+ return MakeTableFileIO(config_.configs(), table_config,
storage_credentials,
+ std::move(provider));
}
return file_io_;
@@ -772,11 +843,12 @@ Result<std::shared_ptr<Transaction>>
RestCatalog::StageCreateTable(
/*stage_create=*/true, *contextual_session));
auto table_config = std::move(result.config);
auto storage_credentials = std::move(result.storage_credentials);
- ICEBERG_ASSIGN_OR_RAISE(auto table_io,
- TableFileIO(context, table_config,
storage_credentials));
+ // Before the FileIO: refreshing its credentials reuses the table session.
ICEBERG_ASSIGN_OR_RAISE(
auto table_session,
TableAuthSession(identifier, table_config,
std::move(contextual_session)));
+ ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier,
table_config,
+ storage_credentials,
table_session));
ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier,
table_session));
auto table_catalog = std::make_shared<TableScopedCatalog>(
shared_from_this(), context, identifier, table_config,
std::move(table_session),
@@ -890,11 +962,12 @@ Result<std::shared_ptr<Table>>
RestCatalog::MakeTableFromLoadResult(
std::shared_ptr<auth::AuthSession> contextual_session) {
auto table_config = std::move(result.config);
auto storage_credentials = std::move(result.storage_credentials);
- ICEBERG_ASSIGN_OR_RAISE(auto table_io,
- TableFileIO(context, table_config,
storage_credentials));
+ // Before the FileIO: refreshing its credentials reuses the table session.
ICEBERG_ASSIGN_OR_RAISE(
auto table_session,
TableAuthSession(identifier, table_config,
std::move(contextual_session)));
+ ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier,
table_config,
+ storage_credentials,
table_session));
ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier,
table_session));
auto table_catalog = std::make_shared<TableScopedCatalog>(
shared_from_this(), context, identifier, table_config, table_session,
table_io);
diff --git a/src/iceberg/catalog/rest/rest_catalog.h
b/src/iceberg/catalog/rest/rest_catalog.h
index 65b0b5ea..6f65e6d6 100644
--- a/src/iceberg/catalog/rest/rest_catalog.h
+++ b/src/iceberg/catalog/rest/rest_catalog.h
@@ -85,9 +85,17 @@ class ICEBERG_REST_EXPORT RestCatalog final
std::shared_ptr<auth::AuthSession> contextual_session);
Result<std::shared_ptr<FileIO>> TableFileIO(
- const SessionContext& context,
+ const SessionContext& context, const TableIdentifier& identifier,
const std::unordered_map<std::string, std::string>& table_config,
- const std::vector<StorageCredential>& storage_credentials) const;
+ const std::vector<StorageCredential>& storage_credentials,
+ std::shared_ptr<auth::AuthSession> table_session) const;
+
+ /// \brief Build a provider for this table's vended credentials.
+ ///
+ /// Returns nullptr when the catalog does not serve LoadCredentials.
+ std::shared_ptr<StorageCredentialProvider> MakeStorageCredentialProvider(
+ const TableIdentifier& identifier,
+ std::shared_ptr<auth::AuthSession> table_session) const;
Result<std::vector<Namespace>> ListNamespaces(const Namespace& ns,
auth::AuthSession& session)
const;
diff --git a/src/iceberg/catalog/rest/rest_file_io.cc
b/src/iceberg/catalog/rest/rest_file_io.cc
index cdc1204f..4cd10270 100644
--- a/src/iceberg/catalog/rest/rest_file_io.cc
+++ b/src/iceberg/catalog/rest/rest_file_io.cc
@@ -21,11 +21,13 @@
#include <string>
#include <unordered_map>
+#include <utility>
#include <vector>
#include "iceberg/catalog/rest/types.h"
#include "iceberg/file_io.h"
#include "iceberg/file_io_registry.h"
+#include "iceberg/logging/log_macros.h"
#include "iceberg/resolving_file_io.h"
#include "iceberg/util/macros.h"
@@ -57,7 +59,8 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const
RestCatalogProperties& c
Result<std::unique_ptr<FileIO>> MakeTableFileIO(
const std::unordered_map<std::string, std::string>& catalog_config,
const std::unordered_map<std::string, std::string>& table_config,
- const std::vector<StorageCredential>& storage_credentials) {
+ const std::vector<StorageCredential>& storage_credentials,
+ std::shared_ptr<StorageCredentialProvider> provider) {
const auto default_properties = MergeFileIOProperties(catalog_config,
table_config);
ICEBERG_ASSIGN_OR_RAISE(
auto io,
MakeCatalogFileIO(RestCatalogProperties::FromMap(default_properties)));
@@ -65,7 +68,16 @@ Result<std::unique_ptr<FileIO>> MakeTableFileIO(
if (storage_credentials.empty()) {
return io;
} else if (auto* credentialed = io->AsSupportsStorageCredentials()) {
-
ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(storage_credentials));
+ // First, so the FileIO never briefly holds credentials it cannot replace.
+ auto status =
credentialed->InitializeStorageCredentials(storage_credentials,
+
std::move(provider));
+ if (!status && !storage_credentials.empty() &&
+ status.error().kind == ErrorKind::kNotSupported) {
+ ICEBERG_LOG_WARN("Configured FileIO cannot refresh vended storage
credentials: {}",
+ status.error().message);
+ status = credentialed->SetStorageCredentials(storage_credentials);
+ }
+ ICEBERG_RETURN_UNEXPECTED(status);
} else {
return NotSupported("Configured FileIO does not support vended storage
credentials");
}
diff --git a/src/iceberg/catalog/rest/rest_file_io.h
b/src/iceberg/catalog/rest/rest_file_io.h
index e2316c3e..5eb9f97f 100644
--- a/src/iceberg/catalog/rest/rest_file_io.h
+++ b/src/iceberg/catalog/rest/rest_file_io.h
@@ -44,6 +44,7 @@ ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>>
MakeCatalogFileIO(
ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeTableFileIO(
const std::unordered_map<std::string, std::string>& catalog_config,
const std::unordered_map<std::string, std::string>& table_config,
- const std::vector<StorageCredential>& storage_credentials);
+ const std::vector<StorageCredential>& storage_credentials,
+ std::shared_ptr<StorageCredentialProvider> provider = nullptr);
} // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/types.h b/src/iceberg/catalog/rest/types.h
index 20a59fa5..a47a8e85 100644
--- a/src/iceberg/catalog/rest/types.h
+++ b/src/iceberg/catalog/rest/types.h
@@ -209,6 +209,21 @@ using CreateTableResponse = LoadTableResult;
/// \brief Alias of LoadTableResult used as the body of LoadTableResponse
using LoadTableResponse = LoadTableResult;
+/// \brief Response body of the LoadCredentials API.
+struct ICEBERG_REST_EXPORT LoadCredentialsResponse {
+ std::vector<StorageCredential> storage_credentials;
+
+ /// \brief Validates the LoadCredentialsResponse.
+ Status Validate() const {
+ for (const auto& credential : storage_credentials) {
+ ICEBERG_RETURN_UNEXPECTED(credential.Validate());
+ }
+ return {};
+ }
+
+ bool operator==(const LoadCredentialsResponse& other) const = default;
+};
+
/// \brief Response body for listing namespaces.
struct ICEBERG_REST_EXPORT ListNamespacesResponse {
PageToken next_page_token;
diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h
index e22e5cf2..4dd96530 100644
--- a/src/iceberg/file_io.h
+++ b/src/iceberg/file_io.h
@@ -29,6 +29,7 @@
#include <span>
#include <string>
#include <string_view>
+#include <utility>
#include <vector>
#include "iceberg/iceberg_export.h"
@@ -198,6 +199,18 @@ class ICEBERG_EXPORT SupportsStorageCredentials {
/// By value because a concurrent install may replace them. An implementation
/// that delegates may report what was installed on it.
virtual std::vector<StorageCredential> credentials() const = 0;
+
+ /// \brief Install initial credentials and their refresh source before first
use.
+ ///
+ /// The default implementation supports static credentials only.
+ virtual Status InitializeStorageCredentials(
+ const std::vector<StorageCredential>& storage_credentials,
+ std::shared_ptr<StorageCredentialProvider> provider) {
+ if (provider) {
+ return NotSupported("Credential refresh is not supported");
+ }
+ return SetStorageCredentials(storage_credentials);
+ }
};
} // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.cc b/src/iceberg/resolving_file_io.cc
index 91be67ea..ddf19b6f 100644
--- a/src/iceberg/resolving_file_io.cc
+++ b/src/iceberg/resolving_file_io.cc
@@ -24,6 +24,7 @@
#include <utility>
#include "iceberg/file_io_registry.h"
+#include "iceberg/logging/log_macros.h"
#include "iceberg/util/location_util.h"
#include "iceberg/util/macros.h"
#include "iceberg/util/string_util.h"
@@ -44,13 +45,19 @@ Result<std::shared_ptr<FileIO>>
ResolvingFileIO::FileIOForPath(
// without static keys may wait on the EC2 metadata service), which would
// stall every other operation. Forwards all credentials; each implementation
// applies the prefixes it understands.
- auto load = [&](const std::vector<StorageCredential>& credentials)
+ auto load = [&](const std::vector<StorageCredential>& credentials,
+ const std::shared_ptr<StorageCredentialProvider>& provider)
-> Result<std::shared_ptr<FileIO>> {
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<FileIO> io,
FileIORegistry::Load(name, properties_));
- if (!credentials.empty()) {
- if (auto* credentialed = io->AsSupportsStorageCredentials()) {
-
ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(credentials));
+ if (auto* credentialed = io->AsSupportsStorageCredentials()) {
+ auto status = credentialed->InitializeStorageCredentials(credentials,
provider);
+ if (!status && provider && status.error().kind ==
ErrorKind::kNotSupported) {
+ ICEBERG_LOG_WARN("FileIO '{}' cannot refresh vended storage
credentials", name);
+ status = credentialed->SetStorageCredentials(credentials);
+ }
+ if (!status) {
+ return std::unexpected(status.error());
}
}
return io;
@@ -59,6 +66,7 @@ Result<std::shared_ptr<FileIO>>
ResolvingFileIO::FileIOForPath(
while (true) {
uint64_t generation = 0;
std::vector<StorageCredential> credentials;
+ std::shared_ptr<StorageCredentialProvider> provider;
{
std::shared_lock lock(mutex_);
if (const auto cached = io_by_name_.find(name); cached !=
io_by_name_.end()) {
@@ -66,13 +74,14 @@ Result<std::shared_ptr<FileIO>>
ResolvingFileIO::FileIOForPath(
}
generation = credential_generation_;
credentials = storage_credentials_;
+ provider = provider_;
}
// Declared before the lock, so a delegate that is not cached is torn down
// only after the lock is released.
- auto loaded = load(credentials);
+ auto loaded = load(credentials, provider);
std::unique_lock lock(mutex_);
if (generation != credential_generation_) {
- continue; // Credentials were replaced mid-load; load again with them.
+ continue; // Replaced mid-load; load again with what is installed now.
}
if (const auto cached = io_by_name_.find(name); cached !=
io_by_name_.end()) {
return cached->second;
@@ -137,4 +146,17 @@ std::vector<StorageCredential>
ResolvingFileIO::credentials() const {
return storage_credentials_;
}
+Status ResolvingFileIO::InitializeStorageCredentials(
+ const std::vector<StorageCredential>& storage_credentials,
+ std::shared_ptr<StorageCredentialProvider> provider) {
+ std::unique_lock lock(mutex_);
+ if (!io_by_name_.empty()) {
+ return InvalidArgument("Storage credentials must be initialized before
first use");
+ }
+ storage_credentials_ = storage_credentials;
+ provider_ = std::move(provider);
+ ++credential_generation_;
+ return {};
+}
+
} // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.h b/src/iceberg/resolving_file_io.h
index 9a5cf047..05e11eb8 100644
--- a/src/iceberg/resolving_file_io.h
+++ b/src/iceberg/resolving_file_io.h
@@ -40,8 +40,8 @@ namespace iceberg {
/// \brief FileIO that resolves and caches implementations by registry name.
///
-/// Vended credentials are forwarded to every resolved implementation that
-/// supports them; each applies what it understands.
+/// Vended credentials and any credential provider are forwarded to every
resolved
+/// implementation that supports them; each applies what it understands.
class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
public SupportsStorageCredentials
{
public:
@@ -62,8 +62,17 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
Status SetStorageCredentials(
const std::vector<StorageCredential>& storage_credentials) override;
+ /// \brief Return the credentials installed on this resolver.
+ ///
+ /// Not necessarily the ones in use: a resolved implementation refreshes its
+ /// own without reporting back.
std::vector<StorageCredential> credentials() const override;
+ /// \brief Install initial credentials before the first delegate is loaded.
+ Status InitializeStorageCredentials(
+ const std::vector<StorageCredential>& storage_credentials,
+ std::shared_ptr<StorageCredentialProvider> provider) override;
+
SupportsStorageCredentials* AsSupportsStorageCredentials() override { return
this; }
private:
@@ -74,6 +83,7 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
// Guards lazy resolution and credential state.
mutable std::shared_mutex mutex_;
std::vector<StorageCredential> storage_credentials_;
+ std::shared_ptr<StorageCredentialProvider> provider_;
// Bumped by every credential install, so a delegate loaded from an older set
// never reaches the cache.
uint64_t credential_generation_ = 0;
diff --git a/src/iceberg/storage_credential.h b/src/iceberg/storage_credential.h
index 746bad08..d2109961 100644
--- a/src/iceberg/storage_credential.h
+++ b/src/iceberg/storage_credential.h
@@ -24,6 +24,7 @@
#include <string>
#include <unordered_map>
+#include <vector>
#include "iceberg/iceberg_export.h"
#include "iceberg/result.h"
@@ -48,4 +49,15 @@ struct ICEBERG_EXPORT StorageCredential {
bool operator==(const StorageCredential& other) const = default;
};
+/// \brief Source for loading the current vended credentials.
+///
+/// Implementations must be safe for concurrent calls.
+class ICEBERG_EXPORT StorageCredentialProvider {
+ public:
+ virtual ~StorageCredentialProvider() = default;
+
+ /// \brief Load the current credentials. Returns the whole vended list.
+ virtual Result<std::vector<StorageCredential>> Load() = 0;
+};
+
} // namespace iceberg
diff --git a/src/iceberg/test/resolving_file_io_test.cc
b/src/iceberg/test/resolving_file_io_test.cc
index 2189b64a..a3b19487 100644
--- a/src/iceberg/test/resolving_file_io_test.cc
+++ b/src/iceberg/test/resolving_file_io_test.cc
@@ -56,9 +56,28 @@ class RecordingFileIO : public FileIO {
std::vector<std::vector<std::string>> deleted_batches;
};
+class StaticStorageCredentialProvider : public StorageCredentialProvider {
+ public:
+ explicit StaticStorageCredentialProvider(std::vector<StorageCredential>
credentials)
+ : credentials_(std::move(credentials)) {}
+
+ Result<std::vector<StorageCredential>> Load() override { return
credentials_; }
+
+ private:
+ std::vector<StorageCredential> credentials_;
+};
+
class RecordingCredentialedFileIO : public RecordingFileIO,
public SupportsStorageCredentials {
public:
+ Status InitializeStorageCredentials(
+ const std::vector<StorageCredential>& storage_credentials,
+ std::shared_ptr<StorageCredentialProvider> provider) override {
+ credentials_ = storage_credentials;
+ provider_ = std::move(provider);
+ return {};
+ }
+
Status SetStorageCredentials(
const std::vector<StorageCredential>& storage_credentials) override {
credentials_ = storage_credentials;
@@ -69,8 +88,19 @@ class RecordingCredentialedFileIO : public RecordingFileIO,
SupportsStorageCredentials* AsSupportsStorageCredentials() override { return
this; }
+ Status Refresh() {
+ if (!provider_) {
+ return NotFound("no provider installed");
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto refreshed, provider_->Load());
+ return SetStorageCredentials(refreshed);
+ }
+
+ bool has_provider() const { return provider_ != nullptr; }
+
private:
std::vector<StorageCredential> credentials_;
+ std::shared_ptr<StorageCredentialProvider> provider_;
};
// File-scope recording state: registry factories are process-global, so they
@@ -318,4 +348,39 @@ TEST(ResolvingFileIOTest,
LoadsWithoutTheLockAndDropsStaleDelegates) {
EXPECT_EQ(last->credentials(), fresh);
}
+TEST(ResolvingFileIOTest, ForwardsCredentialProviderToResolvedImplementations)
{
+ RegisterRecordingFileIOs();
+ ResolvingFileIO io({});
+
+ std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config =
{{"k2", "v2"}}}};
+ auto provider = std::make_shared<StaticStorageCredentialProvider>(refreshed);
+ EXPECT_THAT(io.InitializeStorageCredentials(
+ {{.prefix = "s3", .config = {{"k1", "v1"}}}}, provider),
+ IsOk());
+
+ std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+ ASSERT_NE(last_s3_io, nullptr);
+ ASSERT_TRUE(last_s3_io->has_provider());
+ EXPECT_THAT(last_s3_io->Refresh(), IsOk());
+ EXPECT_EQ(last_s3_io->credentials(), refreshed);
+}
+
+TEST(ResolvingFileIOTest, RejectsCredentialProviderAfterFirstUse) {
+ RegisterRecordingFileIOs();
+ ResolvingFileIO io({});
+
+ EXPECT_THAT(io.SetStorageCredentials({{.prefix = "s3", .config = {{"k1",
"v1"}}}}),
+ IsOk());
+ std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+ ASSERT_NE(last_s3_io, nullptr);
+ EXPECT_FALSE(last_s3_io->has_provider());
+ EXPECT_EQ(s3_factory_calls, 1);
+
+ auto provider =
+
std::make_shared<StaticStorageCredentialProvider>(std::vector<StorageCredential>{});
+ EXPECT_THAT(io.InitializeStorageCredentials({}, provider),
+ IsError(ErrorKind::kInvalidArgument));
+ EXPECT_EQ(s3_factory_calls, 1);
+}
+
} // namespace iceberg
diff --git a/src/iceberg/test/rest_file_io_test.cc
b/src/iceberg/test/rest_file_io_test.cc
index 2fce9dac..31a027ab 100644
--- a/src/iceberg/test/rest_file_io_test.cc
+++ b/src/iceberg/test/rest_file_io_test.cc
@@ -50,11 +50,31 @@ class MockFileIO : public FileIO {
Status DeleteFile(const std::string& /*file_location*/) override { return
{}; }
};
+class StaticStorageCredentialProvider : public StorageCredentialProvider {
+ public:
+ explicit StaticStorageCredentialProvider(std::vector<StorageCredential>
credentials)
+ : credentials_(std::move(credentials)) {}
+
+ Result<std::vector<StorageCredential>> Load() override { return
credentials_; }
+
+ private:
+ std::vector<StorageCredential> credentials_;
+};
+
std::vector<StorageCredential> captured_storage_credentials;
std::unordered_map<std::string, std::string> captured_file_io_properties;
+std::shared_ptr<StorageCredentialProvider> captured_provider;
class MockCredentialedFileIO : public MockFileIO, public
SupportsStorageCredentials {
public:
+ Status InitializeStorageCredentials(
+ const std::vector<StorageCredential>& credentials,
+ std::shared_ptr<StorageCredentialProvider> provider) override {
+ captured_storage_credentials = credentials;
+ captured_provider = std::move(provider);
+ return {};
+ }
+
Status SetStorageCredentials(
const std::vector<StorageCredential>& credentials) override {
captured_storage_credentials = credentials;
@@ -155,6 +175,34 @@ TEST(RestFileIOTest,
TableFileIOMergesConfigAndCredentials) {
EXPECT_EQ(credentialed->credentials(), captured_storage_credentials);
}
+TEST(RestFileIOTest, InstallsCredentialProviderOnlyForVendedCredentials) {
+ const std::string custom_impl = "rest-file-io-test-provider";
+ FileIORegistry::Register(
+ custom_impl,
+ {.create = [](const std::unordered_map<std::string, std::string>&
/*properties*/)
+ -> Result<std::unique_ptr<FileIO>> {
+ return std::make_unique<MockCredentialedFileIO>();
+ }});
+ const std::unordered_map<std::string, std::string> table_config{
+ {"io-impl", custom_impl}};
+
+ captured_provider = nullptr;
+ std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config =
{{"k", "v2"}}}};
+ auto provider = std::make_shared<StaticStorageCredentialProvider>(refreshed);
+ auto result = MakeTableFileIO({}, table_config,
+ {{.prefix = "s3", .config = {{"k", "v1"}}}},
provider);
+ ASSERT_THAT(result, IsOk());
+ ASSERT_TRUE(captured_provider);
+ EXPECT_EQ(captured_provider, provider);
+ EXPECT_THAT(captured_provider->Load(), IsOk());
+ EXPECT_EQ(captured_provider->Load().value(), refreshed);
+
+ captured_provider = nullptr;
+ result = MakeTableFileIO({}, table_config, /*storage_credentials=*/{},
provider);
+ ASSERT_THAT(result, IsOk());
+ EXPECT_FALSE(captured_provider);
+}
+
TEST(RestFileIOTest, TableImplOverridesWarehouseScheme) {
captured_file_io_properties.clear();
FileIORegistry::Register(
diff --git a/src/iceberg/test/rest_json_serde_test.cc
b/src/iceberg/test/rest_json_serde_test.cc
index ec41e4a6..70b760b4 100644
--- a/src/iceberg/test/rest_json_serde_test.cc
+++ b/src/iceberg/test/rest_json_serde_test.cc
@@ -1229,6 +1229,32 @@ INSTANTIATE_TEST_SUITE_P(
return info.param.test_name;
});
+TEST(LoadCredentialsResponseTest, ParsesVendedCredentials) {
+ auto json = nlohmann::json::parse(
+
R"({"storage-credentials":[{"prefix":"s3","config":{"s3.access-key-id":"AKIAtest"}}]})");
+ auto response = LoadCredentialsResponseFromJson(json);
+ ASSERT_THAT(response, IsOk());
+ EXPECT_EQ(response->storage_credentials,
+ (std::vector<StorageCredential>{
+ {.prefix = "s3", .config = {{"s3.access-key-id",
"AKIAtest"}}}}));
+}
+
+TEST(LoadCredentialsResponseTest, RejectsResponseWithoutCredentials) {
+ for (std::string_view body : {"{}", R"({"storage-credentials": null})"}) {
+ auto response =
LoadCredentialsResponseFromJson(nlohmann::json::parse(body));
+ EXPECT_THAT(response, IsError(ErrorKind::kJsonParseError));
+ EXPECT_THAT(response, HasErrorMessage("Missing 'storage-credentials'"));
+ }
+}
+
+TEST(LoadCredentialsResponseTest, RejectsNonArrayCredentials) {
+ auto response = LoadCredentialsResponseFromJson(
+ nlohmann::json::parse(R"({"storage-credentials":"oops"})"));
+ EXPECT_THAT(response, IsError(ErrorKind::kJsonParseError));
+ EXPECT_THAT(response,
+ HasErrorMessage("Cannot parse storage credentials from
non-array"));
+}
+
DECLARE_ROUNDTRIP_TEST(CommitTableRequest)
INSTANTIATE_TEST_SUITE_P(
