2020#include " iceberg/catalog/rest/rest_catalog.h"
2121
2222#include < memory>
23+ #include < mutex>
24+ #include < string>
2325#include < string_view>
2426#include < tuple>
2527#include < unordered_map>
4244#include " iceberg/catalog/rest/rest_util.h"
4345#include " iceberg/catalog/rest/types.h"
4446#include " iceberg/json_serde_internal.h"
47+ #include " iceberg/logging/log_macros.h"
4548#include " iceberg/metrics/metrics_reporters.h"
4649#include " iceberg/partition_spec.h"
4750#include " iceberg/result.h"
@@ -60,6 +63,41 @@ namespace iceberg::rest {
6063
6164namespace {
6265
66+ class RestStorageCredentialProvider final : public StorageCredentialProvider {
67+ public:
68+ RestStorageCredentialProvider (std::shared_ptr<HttpClient> client,
69+ std::shared_ptr<auth::AuthSession> session,
70+ std::string path)
71+ : client_(std::move(client)),
72+ session_ (std::move(session)),
73+ path_(std::move(path)) {}
74+
75+ Result<std::vector<StorageCredential>> Load () override {
76+ // The provider can be shared by several FileIOs; keep fetches serialized.
77+ std::lock_guard lock (mutex_);
78+ ICEBERG_ASSIGN_OR_RAISE (const auto response,
79+ client_->Get (path_, /* params=*/ {}, /* headers=*/ {},
80+ *TableErrorHandler::Instance (), *session_));
81+ // Parse errors can contain credential data; return a fixed message.
82+ auto json = FromJsonString (response.body ());
83+ if (!json.has_value ()) {
84+ return JsonParseError (" Malformed LoadCredentials response" );
85+ }
86+ auto result = LoadCredentialsResponseFromJson (*json);
87+ if (!result.has_value ()) {
88+ return std::unexpected<Error>(
89+ {.kind = result.error ().kind , .message = " Malformed LoadCredentials response" });
90+ }
91+ return std::move (result->storage_credentials );
92+ }
93+
94+ private:
95+ std::shared_ptr<HttpClient> client_;
96+ std::shared_ptr<auth::AuthSession> session_;
97+ std::string path_;
98+ std::mutex mutex_;
99+ };
100+
63101// / \brief Get the default set of endpoints for backwards compatibility according to the
64102// / iceberg rest spec.
65103std::unordered_set<Endpoint> GetDefaultEndpoints () {
@@ -508,12 +546,45 @@ Result<std::shared_ptr<auth::AuthSession>> RestCatalog::TableAuthSession(
508546 std::move (contextual_session));
509547}
510548
549+ std::shared_ptr<StorageCredentialProvider> RestCatalog::MakeStorageCredentialProvider (
550+ const TableIdentifier& identifier,
551+ std::shared_ptr<auth::AuthSession> table_session) const {
552+ if (!supported_endpoints_.contains (Endpoint::TableCredentials ())) {
553+ // Not an error, but it surfaces much later as credentials expiring.
554+ ICEBERG_LOG_DEBUG (
555+ " Catalog does not advertise {}; vended credentials for '{}' will not be "
556+ " refreshed" ,
557+ Endpoint::TableCredentials ().ToString (), ToString (identifier));
558+ return nullptr ;
559+ }
560+ auto path = paths_->Credentials (identifier);
561+ if (!path.has_value ()) {
562+ ICEBERG_LOG_WARN (
563+ " Cannot build the credentials path for '{}' ({}); its vended credentials "
564+ " will not be refreshed" ,
565+ ToString (identifier), path.error ().message );
566+ return nullptr ;
567+ }
568+ auto client = client_;
569+ auto credentials_path = std::move (path.value ());
570+ auto session = std::move (table_session);
571+ return std::make_shared<RestStorageCredentialProvider>(
572+ std::move (client), std::move (session), std::move (credentials_path));
573+ }
574+
511575Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO (
512- const SessionContext& /* context*/ ,
576+ const SessionContext& /* context*/ , const TableIdentifier& identifier,
513577 const std::unordered_map<std::string, std::string>& table_config,
514- const std::vector<StorageCredential>& storage_credentials) const {
578+ const std::vector<StorageCredential>& storage_credentials,
579+ std::shared_ptr<auth::AuthSession> table_session) const {
515580 if (!table_config.empty () || !storage_credentials.empty ()) {
516- return MakeTableFileIO (config_.configs (), table_config, storage_credentials);
581+ // Only vended credentials expire, so only they need a provider.
582+ std::shared_ptr<StorageCredentialProvider> provider;
583+ if (!storage_credentials.empty ()) {
584+ provider = MakeStorageCredentialProvider (identifier, std::move (table_session));
585+ }
586+ return MakeTableFileIO (config_.configs (), table_config, storage_credentials,
587+ std::move (provider));
517588 }
518589
519590 return file_io_;
@@ -772,11 +843,12 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
772843 /* stage_create=*/ true , *contextual_session));
773844 auto table_config = std::move (result.config );
774845 auto storage_credentials = std::move (result.storage_credentials );
775- ICEBERG_ASSIGN_OR_RAISE (auto table_io,
776- TableFileIO (context, table_config, storage_credentials));
846+ // Before the FileIO: refreshing its credentials reuses the table session.
777847 ICEBERG_ASSIGN_OR_RAISE (
778848 auto table_session,
779849 TableAuthSession (identifier, table_config, std::move (contextual_session)));
850+ ICEBERG_ASSIGN_OR_RAISE (auto table_io, TableFileIO (context, identifier, table_config,
851+ storage_credentials, table_session));
780852 ICEBERG_ASSIGN_OR_RAISE (auto reporter, MakeTableReporter (identifier, table_session));
781853 auto table_catalog = std::make_shared<TableScopedCatalog>(
782854 shared_from_this (), context, identifier, table_config, std::move (table_session),
@@ -890,11 +962,12 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
890962 std::shared_ptr<auth::AuthSession> contextual_session) {
891963 auto table_config = std::move (result.config );
892964 auto storage_credentials = std::move (result.storage_credentials );
893- ICEBERG_ASSIGN_OR_RAISE (auto table_io,
894- TableFileIO (context, table_config, storage_credentials));
965+ // Before the FileIO: refreshing its credentials reuses the table session.
895966 ICEBERG_ASSIGN_OR_RAISE (
896967 auto table_session,
897968 TableAuthSession (identifier, table_config, std::move (contextual_session)));
969+ ICEBERG_ASSIGN_OR_RAISE (auto table_io, TableFileIO (context, identifier, table_config,
970+ storage_credentials, table_session));
898971 ICEBERG_ASSIGN_OR_RAISE (auto reporter, MakeTableReporter (identifier, table_session));
899972 auto table_catalog = std::make_shared<TableScopedCatalog>(
900973 shared_from_this (), context, identifier, table_config, table_session, table_io);
0 commit comments