wgtmac commented on code in PR #783:
URL: https://github.com/apache/iceberg-cpp/pull/783#discussion_r3510514160
##########
src/iceberg/catalog/rest/rest_catalog.cc:
##########
@@ -873,6 +874,20 @@ Result<std::shared_ptr<Table>>
RestCatalog::MakeTableFromLoadResult(
TableAuthSession(identifier, table_config,
std::move(contextual_session)));
auto table_catalog = std::make_shared<TableScopedCatalog>(
shared_from_this(), context, identifier, table_config, table_session);
+
+ if (supported_endpoints_.contains(Endpoint::PlanTableScan())) {
Review Comment:
This should also gate on the effective `scan-planning-mode`, not only
endpoint support. Java defaults to client-side planning and lets the table
config override the client config, so a table can otherwise be forced into REST
planning even when the server says `client`, or silently fall back when the
server says `server` but the endpoint is missing.
##########
src/iceberg/catalog/rest/rest_table_scan.cc:
##########
@@ -0,0 +1,269 @@
+/*
+ * 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_scan.h"
+
+#include <chrono>
+#include <thread>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/endpoint.h"
+#include "iceberg/catalog/rest/error_handlers.h"
+#include "iceberg/catalog/rest/http_client.h"
+#include "iceberg/catalog/rest/json_serde_internal.h"
+#include "iceberg/catalog/rest/resource_paths.h"
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_serde_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/result.h"
+#include "iceberg/schema.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+constexpr int64_t kMinSleepMs = 1'000;
+constexpr int64_t kMaxSleepMs = 60'000;
+constexpr int kMaxRetries = 10;
+constexpr int64_t kMaxWaitTimeMs = 5 * 60 * 1'000;
+
+#define ICEBERG_ENDPOINT_CHECK(endpoints, endpoint) \
+ do { \
+ if (!endpoints.contains(endpoint)) { \
+ return NotSupported("Not supported endpoint: {}", endpoint.ToString()); \
+ } \
+ } while (0)
+
+} // namespace
+
+// RestTableScan
+
+RestTableScan::RestTableScan(std::shared_ptr<TableMetadata> metadata,
+ std::shared_ptr<Schema> schema,
std::shared_ptr<FileIO> io,
+ internal::TableScanContext context,
+ RestScanContext rest_context)
+ : DataTableScan(std::move(metadata), std::move(schema), std::move(io),
+ std::move(context)),
+ rest_context_(std::move(rest_context)) {}
+
+Result<std::unique_ptr<DataTableScan>> RestTableScan::Make(
+ std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+ std::shared_ptr<FileIO> io, internal::TableScanContext context,
+ RestScanContext rest_context) {
+ ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+ ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
+ ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
+ return std::unique_ptr<DataTableScan>(
+ new RestTableScan(std::move(metadata), std::move(schema), std::move(io),
+ std::move(context), std::move(rest_context)));
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanFiles()
const {
+ TableMetadataCache metadata_cache(metadata_.get());
+ ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id,
metadata_cache.GetPartitionSpecsById());
+
+ std::string plan_id;
+ return PlanTableScan(plan_id, specs_by_id);
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>>
RestTableScan::PlanTableScan(
+ std::string& plan_id,
+ const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs)
const {
+ ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints,
Endpoint::PlanTableScan());
+
+ // Build request from scan context
+ PlanTableScanRequest request;
+ request.select = context_.selected_columns;
+ request.filter = context_.filter;
+ request.case_sensitive = context_.case_sensitive;
+ request.min_rows_requested = context_.min_rows_requested;
+
+ if (context_.from_snapshot_id.has_value() &&
context_.to_snapshot_id.has_value()) {
Review Comment:
We need to set `use-snapshot-schema` for snapshot/time-travel and
incremental scans. Java sends it for `useSnapshot` and start/end snapshot
scans, and the REST spec says time travel should use the snapshot schema.
Without it, schema-evolved tables can be planned against the current schema.
##########
src/iceberg/catalog/rest/rest_table_scan.cc:
##########
@@ -0,0 +1,269 @@
+/*
+ * 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_scan.h"
+
+#include <chrono>
+#include <thread>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/endpoint.h"
+#include "iceberg/catalog/rest/error_handlers.h"
+#include "iceberg/catalog/rest/http_client.h"
+#include "iceberg/catalog/rest/json_serde_internal.h"
+#include "iceberg/catalog/rest/resource_paths.h"
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_serde_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/result.h"
+#include "iceberg/schema.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+constexpr int64_t kMinSleepMs = 1'000;
+constexpr int64_t kMaxSleepMs = 60'000;
+constexpr int kMaxRetries = 10;
+constexpr int64_t kMaxWaitTimeMs = 5 * 60 * 1'000;
+
+#define ICEBERG_ENDPOINT_CHECK(endpoints, endpoint) \
+ do { \
+ if (!endpoints.contains(endpoint)) { \
+ return NotSupported("Not supported endpoint: {}", endpoint.ToString()); \
+ } \
+ } while (0)
+
+} // namespace
+
+// RestTableScan
+
+RestTableScan::RestTableScan(std::shared_ptr<TableMetadata> metadata,
+ std::shared_ptr<Schema> schema,
std::shared_ptr<FileIO> io,
+ internal::TableScanContext context,
+ RestScanContext rest_context)
+ : DataTableScan(std::move(metadata), std::move(schema), std::move(io),
+ std::move(context)),
+ rest_context_(std::move(rest_context)) {}
+
+Result<std::unique_ptr<DataTableScan>> RestTableScan::Make(
+ std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+ std::shared_ptr<FileIO> io, internal::TableScanContext context,
+ RestScanContext rest_context) {
+ ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+ ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
+ ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
+ return std::unique_ptr<DataTableScan>(
+ new RestTableScan(std::move(metadata), std::move(schema), std::move(io),
+ std::move(context), std::move(rest_context)));
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanFiles()
const {
+ TableMetadataCache metadata_cache(metadata_.get());
+ ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id,
metadata_cache.GetPartitionSpecsById());
+
+ std::string plan_id;
+ return PlanTableScan(plan_id, specs_by_id);
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>>
RestTableScan::PlanTableScan(
+ std::string& plan_id,
+ const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs)
const {
+ ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints,
Endpoint::PlanTableScan());
+
+ // Build request from scan context
+ PlanTableScanRequest request;
+ request.select = context_.selected_columns;
+ request.filter = context_.filter;
+ request.case_sensitive = context_.case_sensitive;
+ request.min_rows_requested = context_.min_rows_requested;
+
+ if (context_.from_snapshot_id.has_value() &&
context_.to_snapshot_id.has_value()) {
+ request.start_snapshot_id = context_.from_snapshot_id;
+ request.end_snapshot_id = context_.to_snapshot_id;
+ } else if (context_.snapshot_id.has_value()) {
+ request.snapshot_id = context_.snapshot_id;
+ }
+
+ if (!context_.columns_to_keep_stats.empty()) {
+ for (int32_t field_id : context_.columns_to_keep_stats) {
+ ICEBERG_ASSIGN_OR_RAISE(auto name,
schema_->FindColumnNameById(field_id));
+ if (name.has_value()) {
+ request.stats_fields.emplace_back(*name);
+ }
+ }
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(auto path,
+ rest_context_.paths->Plan(rest_context_.identifier));
+ ICEBERG_ASSIGN_OR_RAISE(auto request_json, ToJson(request));
+ ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(request_json));
+ ICEBERG_ASSIGN_OR_RAISE(
+ const auto response,
+ rest_context_.client->Post(path, json_request, /*headers=*/{},
+ *PlanErrorHandler::Instance(),
*rest_context_.session));
+ ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
+ ICEBERG_ASSIGN_OR_RAISE(auto result,
Review Comment:
Planning responses can include `storage-credentials`. Java switches to a
scan-scoped `FileIO` built from those credentials, and the spec expects clients
to use them for the returned tasks. Ignoring them means servers that vend
temporary storage credentials can plan successfully but reads may fail.
##########
src/iceberg/catalog/rest/rest_table_scan.cc:
##########
@@ -0,0 +1,269 @@
+/*
+ * 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_scan.h"
+
+#include <chrono>
+#include <thread>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/endpoint.h"
+#include "iceberg/catalog/rest/error_handlers.h"
+#include "iceberg/catalog/rest/http_client.h"
+#include "iceberg/catalog/rest/json_serde_internal.h"
+#include "iceberg/catalog/rest/resource_paths.h"
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_serde_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/result.h"
+#include "iceberg/schema.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+constexpr int64_t kMinSleepMs = 1'000;
+constexpr int64_t kMaxSleepMs = 60'000;
+constexpr int kMaxRetries = 10;
+constexpr int64_t kMaxWaitTimeMs = 5 * 60 * 1'000;
+
+#define ICEBERG_ENDPOINT_CHECK(endpoints, endpoint) \
+ do { \
+ if (!endpoints.contains(endpoint)) { \
+ return NotSupported("Not supported endpoint: {}", endpoint.ToString()); \
+ } \
+ } while (0)
+
+} // namespace
+
+// RestTableScan
+
+RestTableScan::RestTableScan(std::shared_ptr<TableMetadata> metadata,
+ std::shared_ptr<Schema> schema,
std::shared_ptr<FileIO> io,
+ internal::TableScanContext context,
+ RestScanContext rest_context)
+ : DataTableScan(std::move(metadata), std::move(schema), std::move(io),
+ std::move(context)),
+ rest_context_(std::move(rest_context)) {}
+
+Result<std::unique_ptr<DataTableScan>> RestTableScan::Make(
+ std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+ std::shared_ptr<FileIO> io, internal::TableScanContext context,
+ RestScanContext rest_context) {
+ ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+ ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
+ ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
+ return std::unique_ptr<DataTableScan>(
+ new RestTableScan(std::move(metadata), std::move(schema), std::move(io),
+ std::move(context), std::move(rest_context)));
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanFiles()
const {
+ TableMetadataCache metadata_cache(metadata_.get());
+ ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id,
metadata_cache.GetPartitionSpecsById());
+
+ std::string plan_id;
+ return PlanTableScan(plan_id, specs_by_id);
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>>
RestTableScan::PlanTableScan(
+ std::string& plan_id,
+ const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs)
const {
+ ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints,
Endpoint::PlanTableScan());
+
+ // Build request from scan context
+ PlanTableScanRequest request;
+ request.select = context_.selected_columns;
+ request.filter = context_.filter;
+ request.case_sensitive = context_.case_sensitive;
+ request.min_rows_requested = context_.min_rows_requested;
+
+ if (context_.from_snapshot_id.has_value() &&
context_.to_snapshot_id.has_value()) {
+ request.start_snapshot_id = context_.from_snapshot_id;
+ request.end_snapshot_id = context_.to_snapshot_id;
+ } else if (context_.snapshot_id.has_value()) {
+ request.snapshot_id = context_.snapshot_id;
+ }
+
+ if (!context_.columns_to_keep_stats.empty()) {
+ for (int32_t field_id : context_.columns_to_keep_stats) {
+ ICEBERG_ASSIGN_OR_RAISE(auto name,
schema_->FindColumnNameById(field_id));
+ if (name.has_value()) {
+ request.stats_fields.emplace_back(*name);
+ }
+ }
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(auto path,
+ rest_context_.paths->Plan(rest_context_.identifier));
+ ICEBERG_ASSIGN_OR_RAISE(auto request_json, ToJson(request));
+ ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(request_json));
+ ICEBERG_ASSIGN_OR_RAISE(
+ const auto response,
+ rest_context_.client->Post(path, json_request, /*headers=*/{},
+ *PlanErrorHandler::Instance(),
*rest_context_.session));
+ ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
+ ICEBERG_ASSIGN_OR_RAISE(auto result,
+ PlanTableScanResponseFromJson(json, specs,
*schema_));
+ ICEBERG_RETURN_UNEXPECTED(result.Validate());
+
+ plan_id = result.plan_id;
+
+ switch (result.plan_status) {
+ case PlanStatus::kCompleted:
+ return ResolveScanTasks(result.plan_tasks, result.file_scan_tasks,
specs);
Review Comment:
Once a `plan-id` is returned, the server may hold resources until all plan
tasks are fetched or the plan is cancelled. If resolving paginated tasks fails
partway through, this returns without cancelling the remaining plan; Java
cancels from the scan-task iterable cleanup path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
