Skip to content
Open
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
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
11 changes: 11 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,7 @@
#include "iceberg/catalog/rest/catalog_properties.h"

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

Expand Down Expand Up @@ -61,4 +62,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
10 changes: 10 additions & 0 deletions src/iceberg/catalog/rest/catalog_properties.h
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,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 +61,8 @@ 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 The prefix for HTTP headers.
inline static constexpr std::string_view kHeaderPrefix = "header.";

Expand All @@ -80,6 +85,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
9 changes: 9 additions & 0 deletions src/iceberg/catalog/rest/http_client.cc
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,15 @@ std::unordered_map<std::string, std::string> HttpResponse::headers() const {
return impl_->headers();
}

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

namespace {

/// \brief Default error type for unparseable REST responses.
Expand Down
40 changes: 21 additions & 19 deletions src/iceberg/catalog/rest/http_client.h
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,9 @@ class ICEBERG_REST_EXPORT HttpResponse {
/// \brief Get the headers of the response as a map.
std::unordered_map<std::string, std::string> headers() const;

/// \brief Create a response for use in unit tests.
static HttpResponse MakeForTesting(int32_t status_code, std::string body);

private:
friend class HttpClient;
class Impl;
Expand All @@ -71,44 +74,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: 24 additions & 0 deletions src/iceberg/catalog/rest/json_serde.cc
Original file line number Diff line number Diff line change
Expand Up @@ -531,6 +531,15 @@ Result<nlohmann::json> ScanTaskFieldsToJson(
json[kFileScanTasks] = std::move(tasks_json);
}

if (!response.storage_credentials.empty()) {
nlohmann::json creds_json = nlohmann::json::array();
for (const auto& cred : response.storage_credentials) {
ICEBERG_ASSIGN_OR_RAISE(auto entry, StorageCredentialToJson(cred));
creds_json.push_back(std::move(entry));
}
json[kStorageCredentials] = std::move(creds_json);
}

return json;
}

Expand Down Expand Up @@ -571,6 +580,21 @@ Status ScanTaskFieldsFromJson(
FileScanTasksFromJson(file_scan_tasks_json, response.delete_files,
partition_specs_by_id, schema));
}

// 4. storage_credentials
if (json.contains(kStorageCredentials)) {
ICEBERG_ASSIGN_OR_RAISE(auto creds_json,
GetJsonValue<nlohmann::json>(json, kStorageCredentials));
if (!creds_json.is_array()) {
return JsonParseError("Cannot parse storage credentials from non-array: {}",
SafeDumpJson(creds_json));
}
for (const auto& entry : creds_json) {
ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(entry));
response.storage_credentials.push_back(std::move(cred));
}
}

return {};
}

Expand Down
4 changes: 4 additions & 0 deletions src/iceberg/catalog/rest/meson.build
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@ iceberg_rest_sources = files(
'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 Expand Up @@ -95,6 +97,8 @@ install_headers(
'resource_paths.h',
'rest_catalog.h',
'rest_file_io.h',
'rest_table.h',
'rest_table_scan.h',
'rest_util.h',
'type_fwd.h',
'types.h',
Expand Down
46 changes: 45 additions & 1 deletion src/iceberg/catalog/rest/rest_catalog.cc
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,12 @@
#include "iceberg/catalog/rest/resource_paths.h"
#include "iceberg/catalog/rest/rest_file_io.h"
#include "iceberg/catalog/rest/rest_metrics_reporter_internal.h"
#include "iceberg/catalog/rest/rest_table.h"
#include "iceberg/catalog/rest/rest_util.h"
#include "iceberg/catalog/rest/types.h"
#include "iceberg/json_serde_internal.h"
#include "iceberg/logging/log_level.h"
#include "iceberg/logging/logger.h"
#include "iceberg/metrics/metrics_reporters.h"
#include "iceberg/partition_spec.h"
#include "iceberg/result.h"
Expand Down Expand Up @@ -454,7 +457,7 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(

RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
std::shared_ptr<HttpClient> client,
std::unique_ptr<ResourcePaths> paths,
std::shared_ptr<ResourcePaths> paths,
std::unordered_set<Endpoint> endpoints,
std::unique_ptr<auth::AuthManager> auth_manager,
std::shared_ptr<auth::AuthSession> catalog_session,
Expand Down Expand Up @@ -899,6 +902,47 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
auto table_catalog = std::make_shared<TableScopedCatalog>(
shared_from_this(), context, identifier, table_config, table_session, table_io);

// Determine effective scan planning mode: table config overrides client config.
ICEBERG_ASSIGN_OR_RAISE(auto client_mode,
RestCatalogProperties::ScanPlanningModeFrom(config_.configs()));
ICEBERG_ASSIGN_OR_RAISE(auto server_mode,
RestCatalogProperties::ScanPlanningModeFrom(table_config));

if (client_mode.has_value() && server_mode.has_value() &&
*client_mode != *server_mode) {
Log(LogLevel::kWarn,
"Scan planning mode mismatch for table {}: client config={}, server config={}. "
"Server config will take precedence.",
identifier.ToString(),
*client_mode == ScanPlanningMode::kClient ? "client" : "server",
*server_mode == ScanPlanningMode::kClient ? "client" : "server");
}

ScanPlanningMode effective_mode =
server_mode.value_or(client_mode.value_or(ScanPlanningMode::kClient));

if (effective_mode == ScanPlanningMode::kServer) {
if (!supported_endpoints_.contains(Endpoint::PlanTableScan())) {
return NotSupported(
"Server requires server-side scan planning for table {} but does not support "
"the PlanTableScan endpoint.",
identifier.ToString());
}
RestScanContext rest_ctx{
.client = client_,
.paths = paths_,
.session = table_session,
.supported_endpoints = supported_endpoints_,
.identifier = identifier,
.catalog_config = config_.configs(),
.table_config = table_config,
};
return RestTable::Make(identifier, std::move(result.metadata),
std::move(result.metadata_location), std::move(table_io),
std::move(table_catalog), RestTableName(name_, identifier),
reporter, std::move(rest_ctx));
}

return Table::Make(identifier, std::move(result.metadata),
std::move(result.metadata_location), std::move(table_io),
std::move(table_catalog), RestTableName(name_, identifier),
Expand Down
4 changes: 2 additions & 2 deletions src/iceberg/catalog/rest/rest_catalog.h
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ class ICEBERG_REST_EXPORT RestCatalog final
class TableScopedCatalog;

RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
std::shared_ptr<HttpClient> client, std::unique_ptr<ResourcePaths> paths,
std::shared_ptr<HttpClient> client, std::shared_ptr<ResourcePaths> paths,
std::unordered_set<Endpoint> endpoints,
std::unique_ptr<auth::AuthManager> auth_manager,
std::shared_ptr<auth::AuthSession> catalog_session,
Expand Down Expand Up @@ -193,7 +193,7 @@ class ICEBERG_REST_EXPORT RestCatalog final
RestCatalogProperties config_;
std::shared_ptr<FileIO> file_io_;
std::shared_ptr<HttpClient> client_;
std::unique_ptr<ResourcePaths> paths_;
std::shared_ptr<ResourcePaths> paths_;
std::string name_;
std::unordered_set<Endpoint> supported_endpoints_;
std::unique_ptr<auth::AuthManager> auth_manager_;
Expand Down
62 changes: 62 additions & 0 deletions src/iceberg/catalog/rest/rest_table.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
/*
* 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.
*/

#include "iceberg/catalog/rest/rest_table.h"

#include <memory>
#include <utility>

#include "iceberg/catalog/rest/rest_table_scan.h"
#include "iceberg/result.h"
#include "iceberg/table_metadata.h"
#include "iceberg/util/macros.h"

namespace iceberg::rest {

RestTable::RestTable(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
std::string metadata_location, std::shared_ptr<FileIO> io,
std::shared_ptr<Catalog> catalog, std::string full_name,
std::shared_ptr<MetricsReporter> reporter,
RestScanContext rest_context)
: Table(std::move(identifier), std::move(metadata), std::move(metadata_location),
std::move(io), std::move(catalog), std::move(full_name), std::move(reporter)),
rest_context_(std::move(rest_context)) {}

RestTable::~RestTable() = default;

Result<std::shared_ptr<RestTable>> RestTable::Make(
TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
std::string metadata_location, std::shared_ptr<FileIO> io,
std::shared_ptr<Catalog> catalog, std::string full_name,
std::shared_ptr<MetricsReporter> reporter, RestScanContext rest_context) {
if (metadata == nullptr) {
return InvalidArgument("Metadata cannot be null");
}
return std::shared_ptr<RestTable>(
new RestTable(std::move(identifier), std::move(metadata),
std::move(metadata_location), std::move(io), std::move(catalog),
std::move(full_name), std::move(reporter), std::move(rest_context)));
}

Result<std::unique_ptr<DataTableScanBuilder>> RestTable::NewScan() const {
return std::make_unique<RestTableScanBuilder>(metadata_, io_, full_name_, reporter_,
rest_context_);
}

} // namespace iceberg::rest
Loading
Loading