Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/iceberg/catalog/rest/constant.h
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down
34 changes: 34 additions & 0 deletions src/iceberg/catalog/rest/json_serde.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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;
}

Expand All @@ -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;
}
Expand Down Expand Up @@ -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
1 change: 1 addition & 0 deletions src/iceberg/catalog/rest/json_serde_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
33 changes: 26 additions & 7 deletions src/iceberg/catalog/rest/rest_catalog.cc
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,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()));

Expand Down Expand Up @@ -361,8 +363,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(
Expand Down Expand Up @@ -404,10 +411,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);
}

Expand Down Expand Up @@ -474,9 +485,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());
}

Expand All @@ -498,8 +513,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());
}

Expand Down
80 changes: 75 additions & 5 deletions src/iceberg/catalog/rest/rest_file_io.cc
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,9 @@
#include "iceberg/catalog/rest/rest_file_io.h"

#include <string>
#include <unordered_map>

#include "iceberg/catalog/rest/catalog_properties.h"
#include "iceberg/file_io_registry.h"
#include "iceberg/util/macros.h"

Expand Down Expand Up @@ -69,12 +71,13 @@ 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 — enabling per-table ResolveTableFileIO (vending).
io_impl = std::string(FileIORegistry::kArrowLocalFileIO);
Comment on lines +74 to +76

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is change from previously returning InvalidArgument, is this approach to credential vending fine?

} 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)) {
Expand All @@ -92,4 +95,71 @@ 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,
std::string_view warehouse, const LoadTableResult& result) {
if (result.config.empty() && result.storage_credentials.empty()) {
return catalog_io;
}

// Merge order: catalog props < table config < storage credentials (highest priority).
const StorageCredential* cred = nullptr;
if (!result.metadata_location.empty()) {
cred = ResolveStorageCredential(result.storage_credentials, result.metadata_location);
}
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.

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

matches pyiceberg?

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
14 changes: 14 additions & 0 deletions src/iceberg/catalog/rest/rest_file_io.h
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,13 @@

#include <cstdint>
#include <memory>
#include <string>
#include <string_view>
#include <unordered_map>

#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"
Expand All @@ -44,4 +47,15 @@ ICEBERG_REST_EXPORT std::string_view BuiltinFileIOName(BuiltinFileIOKind kind);
ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
const RestCatalogProperties& config);

/// \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,
std::string_view warehouse, const LoadTableResult& result);

} // namespace iceberg::rest
3 changes: 2 additions & 1 deletion src/iceberg/catalog/rest/types.cc
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,8 @@ bool CreateTableRequest::operator==(const CreateTableRequest& other) const {
}

bool LoadTableResult::operator==(const LoadTableResult& other) const {
if (metadata_location != other.metadata_location || config != other.config) {
if (metadata_location != other.metadata_location || config != other.config ||
storage_credentials != other.storage_credentials) {
return false;
}

Expand Down
15 changes: 14 additions & 1 deletion src/iceberg/catalog/rest/types.h
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
13 changes: 13 additions & 0 deletions src/iceberg/file_io.h
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
#include <span>
#include <string>
#include <string_view>
#include <unordered_map>
#include <vector>

#include "iceberg/iceberg_export.h"
Expand Down Expand Up @@ -118,6 +119,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.

@smaheshwar-pltr smaheshwar-pltr Apr 17, 2026 •

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Real-world engine validation: Trino

Trino (a major Java query engine) validates this exact pattern. Its IcebergRestCatalogFileSystemFactory extracts vended credentials by calling baseTable.io().properties() directly — it does NOT use Java's SupportsStorageCredentials interface.

The flow: BaseTable.io().properties() → IcebergTableCredentials → IcebergRestCatalogFileSystemFactory.create(identity, fileIoProperties) → extracts S3/GCS/Azure keys → injects them into Trino's ConnectorIdentity.extraCredentials → Trino's filesystem factories use the vended creds.

See: trino/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/IcebergRestCatalogFileSystemFactory.java

const std::unordered_map<std::string, std::string>& properties() const {
return properties_;

@smaheshwar-pltr smaheshwar-pltr Apr 17, 2026 •

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why FileIO::properties() instead of Table::io_properties()?

We initially had io_properties() on the Table class, but after reviewing all three reference implementations:

  • Java: engines use table.io() directly; credentials live on FileIO via SupportsStorageCredentials
  • Python: engines read table.io.properties (a dict on FileIO) — e.g., the DataFusion integration explicitly passes self.io.properties
  • Rust: credentials flow through table.file_io().config().props()

No implementation exposes credentials on the Table object itself. FileIO::properties() matches the Python pattern and is the standard surface for engines that need to configure their own storage access.

}

/// \brief Create an input file handle for the given location.
virtual Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location);

Expand Down Expand Up @@ -165,6 +174,10 @@ class ICEBERG_EXPORT FileIO {
/// \param file_locations The locations of the files to delete.
/// \return void if all deletes succeed, or an error code if any delete fails.
virtual Status DeleteFiles(const std::vector<std::string>& file_locations);

private:
friend class FileIORegistry;

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

rev: PR comment on this please. friend is controversial

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

friend class FileIORegistry is intentional: properties_ should only be populated by the registry (which creates FileIO instances with merged config). Making it protected would let any subclass set arbitrary properties, breaking the invariant that properties reflect the config used to create the FileIO. The friend constrains this to the single correct code path.

If a second factory path emerges later (e.g., custom FileIO outside the registry), we can add a protected setter at that point.

std::unordered_map<std::string, std::string> properties_;
};

} // namespace iceberg
6 changes: 5 additions & 1 deletion src/iceberg/file_io_registry.cc
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
#include <mutex>
#include <utility>

#include "iceberg/util/macros.h"

namespace iceberg {

namespace {
Expand Down Expand Up @@ -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;

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FileIORegistry::Load() automatically populates properties_ on every FileIO it creates. This means any FileIO from the registry (including per-table FileIOs from ResolveTableFileIO) will have the merged properties available via properties() without each factory needing to set them manually.

return io;
}

} // namespace iceberg
1 change: 1 addition & 0 deletions src/iceberg/test/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,7 @@ if(ICEBERG_BUILD_REST)
add_rest_iceberg_test(rest_catalog_test
SOURCES
auth_manager_test.cc
credential_vending_test.cc
endpoint_test.cc
rest_file_io_test.cc
rest_json_serde_test.cc
Expand Down
Loading
Loading