Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
36 commits
Select commit Hold shift + click to select a range
3803eae
feat(rest): add scan plan endpoint support to REST catalog client
Jun 26, 2026
cf003c8
Address review comments and add tests for scan plan endpoints
Sep 13, 2026
4e0b00f
Fix API compatibility after rebase on main
Sep 13, 2026
0a28427
feat(rest): support storage credentials in scan planning responses
Sep 13, 2026
e40bc52
style: apply clang-format to fix CI pre-commit failures
Sep 13, 2026
be13f91
fix: export TableScanContext so Validate() is linkable from iceberg_rest
Sep 13, 2026
1d7cf26
fix: explicitly default TableScanBuilder move ctor for Windows DLL ex…
Sep 13, 2026
443443e
feat(rest): server-side planning for incremental append scans, PlanFi…
Sep 29, 2026
72a0ff6
fix: remove unnecessary virtual from DataTableScan::PlanFiles()
Sep 29, 2026
050fa34
test(rest): add tests for PlanFilesStream, RestIncrementalAppendScan,…
Sep 29, 2026
d8782b1
fix(rest): call CancelPlanning on all error paths after plan_id is known
Sep 29, 2026
70b16cb
fix(rest): reset scan IO slot at the start of each planning call
Sep 29, 2026
e7debd0
test(rest): add body-inspecting matchers and tests for cancel/credent…
Sep 29, 2026
d28d0e0
test(rest): replace misleading stream comment with true laziness and …
Sep 29, 2026
424af75
test(rest): add RestCatalog::LoadTable scan-planning-mode selection t…
Sep 29, 2026
98657c5
style: apply clang-format to rest catalog and scan files
Sep 29, 2026
ad31c27
style: use raw string literal in rest_catalog_unit_test
Sep 30, 2026
0ca7b8c
fix(rest): resolve incremental append scan range via base class
Oct 11, 2026
24fb9da
fix(rest): expose vended credentials via RestIncrementalAppendScan::io()
Oct 11, 2026
4cdd996
fix(rest): use table schema for branch scans and keep RestTable after…
Oct 11, 2026
7eb96b3
fix(rest): send stats-fields and projected select in scan plan requests
Oct 11, 2026
0e230fa
fix(rest): keep plan tasks lazy for submitted plans in PlanFilesStream
Oct 11, 2026
300cc15
docs(rest): document serial plan task fetching in REST scans
Oct 11, 2026
e1db3ca
fix(rest): cancel plan when the plan response fails validation
Oct 11, 2026
d6c50c4
feat(rest): make scan planning poll timeout and retries configurable
Oct 11, 2026
1c4237d
test: run use_snapshot_schema checks once instead of per format version
Oct 11, 2026
b09f0ad
test(rest): rename stream cancel tests that consume no tasks
Oct 11, 2026
b22effb
fix(rest): use iceberg::unexpected in REST scan planning
Oct 11, 2026
91a1e42
feat(rest): install a plan-scoped credential provider on scan FileIO
Oct 11, 2026
7830f1a
fix(rest): reject plan-id in failed plan responses, matching Java
Oct 11, 2026
80e3de8
fix(rest): omit planId on credential refresh when the plan has no id
Oct 11, 2026
5949bfa
fix(rest): ignore storage credentials in fetchScanTasks responses
Oct 11, 2026
bfc9266
test(rest): use braced return in incremental scan test builder
Oct 11, 2026
ee0b146
build: add REST table scan headers to the C++20 compatibility header
Oct 11, 2026
9d92d39
fix: export GetProjectedIdsVisitor for use from iceberg_rest
Oct 11, 2026
b40f915
fix(rest): fix scan planning and simplify tests
wgtmac Oct 14, 2026
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
2 changes: 2 additions & 0 deletions src/iceberg/catalog/rest/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ set(ICEBERG_REST_SOURCES
rest_catalog.cc
rest_file_io.cc
rest_metrics_reporter.cc
rest_table.cc
rest_table_scan.cc
rest_util.cc
types.cc)

Expand Down
12 changes: 12 additions & 0 deletions src/iceberg/catalog/rest/catalog_properties.cc
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
#include "iceberg/catalog/rest/catalog_properties.h"

#include <algorithm>
#include <cstdint>
#include <optional>
#include <string>
#include <string_view>

Expand Down Expand Up @@ -61,4 +63,14 @@ Result<SnapshotMode> RestCatalogProperties::SnapshotLoadingMode() const {
}
}

Result<std::optional<ScanPlanningMode>> RestCatalogProperties::ScanPlanningModeFrom(
const std::unordered_map<std::string, std::string>& config) {
auto it = config.find(kScanPlanningMode.key());
if (it == config.end()) return std::nullopt;
std::string lower = StringUtils::ToLower(it->second);
if (lower == "client") return ScanPlanningMode::kClient;
if (lower == "server") return ScanPlanningMode::kServer;
return InvalidArgument("Invalid scan planning mode: '{}'.", it->second);
}

} // namespace iceberg::rest
17 changes: 17 additions & 0 deletions src/iceberg/catalog/rest/catalog_properties.h
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

#pragma once

#include <cstdint>
#include <string>
#include <unordered_map>

Expand All @@ -34,6 +35,9 @@ namespace iceberg::rest {
/// \brief Snapshot loading mode for REST catalog.
enum class SnapshotMode : uint8_t { kAll, kRefs };

/// \brief Scan planning mode for REST catalog.
enum class ScanPlanningMode : uint8_t { kClient, kServer };

/// \brief Configuration class for a REST Catalog.
class ICEBERG_REST_EXPORT RestCatalogProperties
: public ConfigBase<RestCatalogProperties> {
Expand All @@ -58,6 +62,14 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
/// \brief Whether to report metrics to the REST catalog server (default: true).
inline static Entry<std::string> kMetricsReportingEnabled{
"rest-metrics-reporting-enabled", "true"};
/// \brief The scan planning mode (client or server).
inline static Entry<std::string> kScanPlanningMode{"scan-planning-mode", "client"};
/// \brief Maximum time to wait for a submitted scan plan, in milliseconds.
inline static Entry<int64_t> kScanPlanningPollTimeoutMs{
"rest-scan-planning.poll-timeout-ms", 300000};
/// \brief Extra polls after the first fetch of a submitted scan plan.
inline static Entry<int32_t> kScanPlanningPollNumRetries{
"rest-scan-planning.poll-num-retries", 10};
/// \brief The prefix for HTTP headers.
inline static constexpr std::string_view kHeaderPrefix = "header.";

Expand All @@ -80,6 +92,11 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
/// "REFS", or an error if the value is invalid. Parsing is
/// case-insensitive to match Java behavior.
Result<SnapshotMode> SnapshotLoadingMode() const;

/// \brief Get the scan planning mode from the given config map, returning
/// std::nullopt if the key is absent.
static Result<std::optional<ScanPlanningMode>> ScanPlanningModeFrom(
const std::unordered_map<std::string, std::string>& config);
};

} // namespace iceberg::rest
8 changes: 8 additions & 0 deletions src/iceberg/catalog/rest/http_client.cc
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,14 @@ class HttpResponse::Impl {
};

HttpResponse::HttpResponse() = default;

HttpResponse::HttpResponse(int32_t status_code, std::string body) {
cpr::Response cpr_response;
cpr_response.status_code = status_code;
cpr_response.text = std::move(body);
impl_ = std::make_unique<HttpResponse::Impl>(std::move(cpr_response));
}

HttpResponse::~HttpResponse() = default;
HttpResponse::HttpResponse(HttpResponse&&) noexcept = default;
HttpResponse& HttpResponse::operator=(HttpResponse&&) noexcept = default;
Expand Down
38 changes: 19 additions & 19 deletions src/iceberg/catalog/rest/http_client.h
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ namespace iceberg::rest {
class ICEBERG_REST_EXPORT HttpResponse {
public:
HttpResponse();
HttpResponse(int32_t status_code, std::string body);
~HttpResponse();

HttpResponse(const HttpResponse&) = delete;
Expand All @@ -71,44 +72,43 @@ class ICEBERG_REST_EXPORT HttpResponse {
class ICEBERG_REST_EXPORT HttpClient {
public:
explicit HttpClient(std::unordered_map<std::string, std::string> default_headers = {});
~HttpClient();
virtual ~HttpClient();

HttpClient(const HttpClient&) = delete;
HttpClient& operator=(const HttpClient&) = delete;
HttpClient(HttpClient&&) = delete;
HttpClient& operator=(HttpClient&&) = delete;

/// \brief Sends a GET request.
Result<HttpResponse> Get(const std::string& path,
const std::unordered_map<std::string, std::string>& params,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler, auth::AuthSession& session);
virtual Result<HttpResponse> Get(
const std::string& path, const std::unordered_map<std::string, std::string>& params,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler, auth::AuthSession& session);

/// \brief Sends a POST request.
Result<HttpResponse> Post(const std::string& path, const std::string& body,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler,
auth::AuthSession& session);
virtual Result<HttpResponse> Post(
const std::string& path, const std::string& body,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler, auth::AuthSession& session);

/// \brief Sends a POST request with form data.
Result<HttpResponse> PostForm(
virtual Result<HttpResponse> PostForm(
const std::string& path,
const std::unordered_map<std::string, std::string>& form_data,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler, auth::AuthSession& session);

/// \brief Sends a HEAD request.
Result<HttpResponse> Head(const std::string& path,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler,
auth::AuthSession& session);
virtual Result<HttpResponse> Head(
const std::string& path,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler, auth::AuthSession& session);

/// \brief Sends a DELETE request.
Result<HttpResponse> Delete(const std::string& path,
const std::unordered_map<std::string, std::string>& params,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler,
auth::AuthSession& session);
virtual Result<HttpResponse> Delete(
const std::string& path, const std::unordered_map<std::string, std::string>& params,
const std::unordered_map<std::string, std::string>& headers,
const ErrorHandler& error_handler, auth::AuthSession& session);

private:
std::unordered_map<std::string, std::string> default_headers_;
Expand Down
24 changes: 22 additions & 2 deletions src/iceberg/catalog/rest/json_serde.cc
Original file line number Diff line number Diff line change
Expand Up @@ -128,8 +128,7 @@ Result<StorageCredential> StorageCredentialFromJson(const nlohmann::json& json)
return credential;
}

/// \brief Reads the optional `storage-credentials` array shared by the
/// LoadTable and LoadCredentials responses.
/// \brief Reads the optional `storage-credentials` array.
Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
const nlohmann::json& json) {
std::vector<StorageCredential> credentials;
Expand All @@ -146,6 +145,21 @@ Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
}
return credentials;
}

/// \brief Writes `storage-credentials` to json if any are present.
Status StorageCredentialsToJson(const std::vector<StorageCredential>& credentials,
nlohmann::json& json) {
if (credentials.empty()) {
return {};
}
nlohmann::json creds_json = nlohmann::json::array();
for (const auto& cred : credentials) {
ICEBERG_ASSIGN_OR_RAISE(auto entry, StorageCredentialToJson(cred));
creds_json.push_back(std::move(entry));
}
json[kStorageCredentials] = std::move(creds_json);
return {};
}
} // namespace

Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
Expand Down Expand Up @@ -829,6 +843,8 @@ Result<nlohmann::json> ToJson(const PlanTableScanRequest& request) {
if (request.snapshot_id.has_value()) {
json[kSnapshotId] = request.snapshot_id.value();
}
// FIXME: Preserve explicit empty projections as select: []; omitting select
// means the default projection. Java currently has the same issue.
if (!request.select.empty()) {
json[kSelect] = request.select;
}
Expand Down Expand Up @@ -884,6 +900,7 @@ Result<PlanTableScanResponse> PlanTableScanResponseFromJson(
}
ICEBERG_RETURN_UNEXPECTED(
ScanTaskFieldsFromJson(json, response, partition_specs_by_id, schema));
ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials, StorageCredentialsFromJson(json));
ICEBERG_RETURN_UNEXPECTED(response.Validate());
return response;
}
Expand All @@ -895,6 +912,7 @@ Result<nlohmann::json> ToJson(
const Schema& schema) {
ICEBERG_ASSIGN_OR_RAISE(nlohmann::json json,
ScanTaskFieldsToJson(response, partition_specs_by_id, schema));
ICEBERG_RETURN_UNEXPECTED(StorageCredentialsToJson(response.storage_credentials, json));
json[kPlanStatus] = ToString(response.plan_status);
if (!response.plan_id.empty()) {
json[kPlanId] = response.plan_id;
Expand All @@ -921,6 +939,7 @@ Result<FetchPlanningResultResponse> FetchPlanningResultResponseFromJson(
}
ICEBERG_RETURN_UNEXPECTED(
ScanTaskFieldsFromJson(json, response, partition_specs_by_id, schema));
ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials, StorageCredentialsFromJson(json));
ICEBERG_RETURN_UNEXPECTED(response.Validate());
return response;
}
Expand All @@ -932,6 +951,7 @@ Result<nlohmann::json> ToJson(
const Schema& schema) {
ICEBERG_ASSIGN_OR_RAISE(nlohmann::json json,
ScanTaskFieldsToJson(response, partition_specs_by_id, schema));
ICEBERG_RETURN_UNEXPECTED(StorageCredentialsToJson(response.storage_credentials, json));
json[kPlanStatus] = ToString(response.plan_status);
if (response.error.has_value()) {
json[kError] = ToJson(*response.error)[kError];
Expand Down
Loading
Loading