wgtmac commented on code in PR #783: URL: https://github.com/apache/iceberg-cpp/pull/783#discussion_r4034231128
########## src/iceberg/catalog/rest/rest_table_scan.h: ########## @@ -0,0 +1,137 @@ +/* + * 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. + */ + +#pragma once + +#include <memory> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/iceberg_rest_export.h" +#include "iceberg/metrics/metrics_reporter.h" +#include "iceberg/result.h" +#include "iceberg/storage_credential.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_scan.h" +#include "iceberg/type_fwd.h" + +/// \file iceberg/catalog/rest/rest_table_scan.h +/// REST-specific table scan that delegates scan planning to the REST catalog server. + +namespace iceberg::rest { + +class HttpClient; +class ResourcePaths; + +namespace auth { +class AuthSession; +} // namespace auth + +/// \brief HTTP context shared between RestTable and RestTableScan. +struct ICEBERG_REST_EXPORT RestScanContext { + std::shared_ptr<HttpClient> client; + std::shared_ptr<ResourcePaths> paths; + std::shared_ptr<auth::AuthSession> session; + std::unordered_set<Endpoint> supported_endpoints; + TableIdentifier identifier; + /// Catalog-level config, used with table_config to build a scan-scoped FileIO + /// when the server vends storage credentials in a planning response. + std::unordered_map<std::string, std::string> catalog_config; + /// Table-level config merged with catalog_config for scan-scoped FileIO creation. + std::unordered_map<std::string, std::string> table_config; +}; + +/// \brief A DataTableScan that delegates PlanFiles() to the REST catalog server +/// via the scan planning endpoints (planTableScan / fetchPlanningResult / +/// cancelPlanning / fetchScanTasks). +class ICEBERG_REST_EXPORT RestTableScan : public DataTableScan { + public: + ~RestTableScan() override = default; + + static Result<std::unique_ptr<DataTableScan>> Make( + std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema, + std::shared_ptr<FileIO> io, internal::TableScanContext context, + RestScanContext rest_context); + + /// \brief Plans files via the REST scan planning endpoints. + Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const override; Review Comment: After #873, callers can use `PlanFilesStream()`. This class only overrides `PlanFiles()`, so the stream path still reads manifests locally and never calls `/plan`. This bypasses server-side planning. ########## src/iceberg/catalog/rest/rest_table_scan.cc: ########## @@ -0,0 +1,294 @@ +/* + * 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/rest_file_io.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; + request.use_snapshot_schema = true; + } else if (context_.snapshot_id.has_value()) { + request.snapshot_id = context_.snapshot_id; + request.use_snapshot_schema = context_.use_snapshot_schema; + } + + 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: { + ICEBERG_RETURN_UNEXPECTED(ApplyStorageCredentials(result.storage_credentials)); + auto tasks = ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs); + if (!tasks.has_value()) CancelPlanning(plan_id); + return tasks; + } + case PlanStatus::kSubmitted: + return FetchPlanningResult(plan_id, specs); + case PlanStatus::kFailed: + return IOError("Scan planning failed: {}", + result.error ? result.error->message : "unknown error"); + case PlanStatus::kCancelled: + return IOError("Scan planning was cancelled for plan_id={}", plan_id); + } + return IOError("Unexpected plan status"); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchPlanningResult( + const std::string& plan_id, + const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs) const { + ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints, + Endpoint::FetchPlanningResult()); + + ICEBERG_ASSIGN_OR_RAISE(auto path, + rest_context_.paths->Plan(rest_context_.identifier, plan_id)); + + auto delay_ms = kMinSleepMs; + auto start = std::chrono::steady_clock::now(); + + for (int retry = 0; retry <= kMaxRetries; ++retry) { + ICEBERG_ASSIGN_OR_RAISE( Review Comment: After a `plan-id` exists, any GET, JSON parse, or response validation error here returns without calling `CancelPlanning()`. The server may keep the plan resources. ########## src/iceberg/catalog/rest/rest_table.h: ########## @@ -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. + */ + +#pragma once + +#include <memory> +#include <string> + +#include "iceberg/catalog/rest/iceberg_rest_export.h" +#include "iceberg/catalog/rest/rest_table_scan.h" +#include "iceberg/metrics/metrics_reporter.h" +#include "iceberg/result.h" +#include "iceberg/table.h" +#include "iceberg/type_fwd.h" + +/// \file iceberg/catalog/rest/rest_table.h +/// A Table subclass that uses server-side distributed scan planning via the REST catalog. + +namespace iceberg::rest { + +/// \brief A Table whose NewScan() returns a RestTableScanBuilder, delegating +/// PlanFiles() to the REST catalog server's scan planning endpoints. +class ICEBERG_REST_EXPORT RestTable final : public Table { + public: + static Result<std::shared_ptr<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); + + ~RestTable() override; + + /// \brief Returns a RestTableScanBuilder that will delegate PlanFiles() to the + /// REST catalog server. + Result<std::unique_ptr<DataTableScanBuilder>> NewScan() const override; Review Comment: This only overrides `NewScan()`. `NewIncrementalAppendScan()` and `NewIncrementalChangelogScan()` still use the base table path and plan manifests locally, even when `scan-planning-mode=server`. ########## src/iceberg/catalog/rest/rest_table_scan.h: ########## @@ -0,0 +1,137 @@ +/* + * 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. + */ + +#pragma once + +#include <memory> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/iceberg_rest_export.h" +#include "iceberg/metrics/metrics_reporter.h" +#include "iceberg/result.h" +#include "iceberg/storage_credential.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_scan.h" +#include "iceberg/type_fwd.h" + +/// \file iceberg/catalog/rest/rest_table_scan.h +/// REST-specific table scan that delegates scan planning to the REST catalog server. + +namespace iceberg::rest { + +class HttpClient; +class ResourcePaths; + +namespace auth { +class AuthSession; +} // namespace auth + +/// \brief HTTP context shared between RestTable and RestTableScan. +struct ICEBERG_REST_EXPORT RestScanContext { + std::shared_ptr<HttpClient> client; + std::shared_ptr<ResourcePaths> paths; + std::shared_ptr<auth::AuthSession> session; + std::unordered_set<Endpoint> supported_endpoints; + TableIdentifier identifier; + /// Catalog-level config, used with table_config to build a scan-scoped FileIO + /// when the server vends storage credentials in a planning response. + std::unordered_map<std::string, std::string> catalog_config; + /// Table-level config merged with catalog_config for scan-scoped FileIO creation. + std::unordered_map<std::string, std::string> table_config; +}; + +/// \brief A DataTableScan that delegates PlanFiles() to the REST catalog server +/// via the scan planning endpoints (planTableScan / fetchPlanningResult / +/// cancelPlanning / fetchScanTasks). +class ICEBERG_REST_EXPORT RestTableScan : public DataTableScan { + public: + ~RestTableScan() override = default; + + static Result<std::unique_ptr<DataTableScan>> Make( + std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema, + std::shared_ptr<FileIO> io, internal::TableScanContext context, + RestScanContext rest_context); + + /// \brief Plans files via the REST scan planning endpoints. + Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const override; + + /// \brief Returns the FileIO to use when reading scan results. + /// + /// If the server vended storage credentials during planning, returns a FileIO + /// initialised with those credentials; otherwise returns the table's FileIO. + /// Must be called after PlanFiles(). + const std::shared_ptr<FileIO>& effective_io() const; Review Comment: The scan is returned as `std::unique_ptr<DataTableScan>`, but `DataTableScan::io()` is not virtual and still returns the table IO. A normal caller cannot reach `effective_io()` without a downcast, so the vended credentials are not used by the normal read path. ########## src/iceberg/test/rest_table_scan_test.cc: ########## @@ -0,0 +1,467 @@ +/* + * 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 <memory> +#include <optional> +#include <string> +#include <unordered_map> +#include <unordered_set> + +#include <gmock/gmock.h> +#include <gtest/gtest.h> +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/rest_table.h" +#include "iceberg/file_io.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_scan.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg::rest { + +using ::testing::_; +using ::testing::Return; + +// -------------------------------------------------------------------------- +// Mock HTTP client that overrides the virtual methods of HttpClient. +// The base class constructor creates a cpr::ConnectionPool, which is a +// lightweight allocation (no network connections are opened at construction). +// -------------------------------------------------------------------------- +class MockHttpClient : public HttpClient { + public: + MockHttpClient() : HttpClient({}) {} + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); +}; + +// -------------------------------------------------------------------------- +// Minimal FileIO stub (no real I/O needed for server-side scan planning tests) +// -------------------------------------------------------------------------- +class NoOpFileIO : public FileIO { + public: + Result<std::string> ReadFile(const std::string&, std::optional<size_t>) override { + return IOError("NoOpFileIO"); + } + Status WriteFile(const std::string&, std::string_view) override { return {}; } + Status DeleteFile(const std::string&) override { return {}; } +}; + +// -------------------------------------------------------------------------- +// Test fixture shared by RestTableScan tests. +// -------------------------------------------------------------------------- +class RestTableScanTest : public ::testing::Test { + protected: + void SetUp() override { + schema_ = std::make_shared<Schema>( + std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "data", string())}); + + auto spec = PartitionSpec::Unpartitioned(); + + constexpr int64_t kSnapshotId = 1000L; + auto snapshot = std::make_shared<Snapshot>( + Snapshot{.snapshot_id = kSnapshotId, + .sequence_number = 1L, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L), + .manifest_list = "/tmp/manifest-list.avro", + .schema_id = schema_->schema_id()}); + + metadata_ = std::make_shared<TableMetadata>( + TableMetadata{.format_version = 2, + .table_uuid = "test-uuid", + .location = "/tmp/table", + .last_sequence_number = 1L, + .last_updated_ms = TimePointMsFromUnixMs(1609459200000L), + .last_column_id = 2, + .schemas = {schema_}, + .current_schema_id = schema_->schema_id(), + .partition_specs = {spec}, + .default_spec_id = spec->spec_id(), + .last_partition_id = 999, + .current_snapshot_id = kSnapshotId, + .snapshots = {snapshot}, + .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ + .snapshot_id = kSnapshotId, + .retention = SnapshotRef::Branch{}})}}}); + + file_io_ = std::make_shared<NoOpFileIO>(); + + mock_client_ = std::make_shared<MockHttpClient>(); + + ICEBERG_UNWRAP_OR_FAIL(paths_, + ResourcePaths::Make("http://test-server";, /*prefix=*/"", + /*namespace_separator=*/"%1F")); + + session_ = auth::AuthSession::MakeDefault(/*headers=*/{}); + + identifier_ = TableIdentifier{.ns = Namespace{{"default"}}, .name = "my_table"}; + + all_plan_endpoints_ = {Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + } + + // Pass std::nullopt to get the full set of plan endpoints (default). + // Pass an explicit set (including empty) to use exactly that set. + RestScanContext MakeContext( + std::optional<std::unordered_set<Endpoint>> endpoints = std::nullopt) { + auto effective = endpoints.has_value() ? std::move(*endpoints) : all_plan_endpoints_; + return RestScanContext{ + .client = mock_client_, + .paths = paths_, + .session = session_, + .supported_endpoints = std::move(effective), + .identifier = identifier_, + }; + } + + Result<std::unique_ptr<DataTableScan>> MakeScan(RestScanContext ctx) { + return RestTableScan::Make(metadata_, schema_, file_io_, internal::TableScanContext{}, + std::move(ctx)); + } + + std::shared_ptr<Schema> schema_; + std::shared_ptr<TableMetadata> metadata_; + std::shared_ptr<FileIO> file_io_; + std::shared_ptr<MockHttpClient> mock_client_; + std::shared_ptr<ResourcePaths> paths_; + std::shared_ptr<auth::AuthSession> session_; + TableIdentifier identifier_; + std::unordered_set<Endpoint> all_plan_endpoints_; +}; + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED immediately, no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompleted) { + constexpr std::string_view kResponseBody = R"({"status":"completed"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED with a non-empty plan-id (still valid). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompletedWithPlanId) { + constexpr std::string_view kResponseBody = + R"({"status":"completed","plan-id":"plan-abc"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns SUBMITTED → poll returns COMPLETED. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesSubmittedThenCompleted) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-poll-1"})"; + constexpr std::string_view kCompletedBody = R"({"status":"completed"})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + EXPECT_CALL(*mock_client_, Get(_, _, _, _, _)) Review Comment: There is no case where a `plan-id` is returned and the poll GET, JSON parsing, or response validation fails. Those errors currently skip cancellation. ########## src/iceberg/test/rest_table_scan_test.cc: ########## @@ -0,0 +1,467 @@ +/* + * 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 <memory> +#include <optional> +#include <string> +#include <unordered_map> +#include <unordered_set> + +#include <gmock/gmock.h> +#include <gtest/gtest.h> +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/rest_table.h" +#include "iceberg/file_io.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_scan.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg::rest { + +using ::testing::_; +using ::testing::Return; + +// -------------------------------------------------------------------------- +// Mock HTTP client that overrides the virtual methods of HttpClient. +// The base class constructor creates a cpr::ConnectionPool, which is a +// lightweight allocation (no network connections are opened at construction). +// -------------------------------------------------------------------------- +class MockHttpClient : public HttpClient { + public: + MockHttpClient() : HttpClient({}) {} + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); +}; + +// -------------------------------------------------------------------------- +// Minimal FileIO stub (no real I/O needed for server-side scan planning tests) +// -------------------------------------------------------------------------- +class NoOpFileIO : public FileIO { + public: + Result<std::string> ReadFile(const std::string&, std::optional<size_t>) override { + return IOError("NoOpFileIO"); + } + Status WriteFile(const std::string&, std::string_view) override { return {}; } + Status DeleteFile(const std::string&) override { return {}; } +}; + +// -------------------------------------------------------------------------- +// Test fixture shared by RestTableScan tests. +// -------------------------------------------------------------------------- +class RestTableScanTest : public ::testing::Test { + protected: + void SetUp() override { + schema_ = std::make_shared<Schema>( + std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "data", string())}); + + auto spec = PartitionSpec::Unpartitioned(); + + constexpr int64_t kSnapshotId = 1000L; + auto snapshot = std::make_shared<Snapshot>( + Snapshot{.snapshot_id = kSnapshotId, + .sequence_number = 1L, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L), + .manifest_list = "/tmp/manifest-list.avro", + .schema_id = schema_->schema_id()}); + + metadata_ = std::make_shared<TableMetadata>( + TableMetadata{.format_version = 2, + .table_uuid = "test-uuid", + .location = "/tmp/table", + .last_sequence_number = 1L, + .last_updated_ms = TimePointMsFromUnixMs(1609459200000L), + .last_column_id = 2, + .schemas = {schema_}, + .current_schema_id = schema_->schema_id(), + .partition_specs = {spec}, + .default_spec_id = spec->spec_id(), + .last_partition_id = 999, + .current_snapshot_id = kSnapshotId, + .snapshots = {snapshot}, + .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ + .snapshot_id = kSnapshotId, + .retention = SnapshotRef::Branch{}})}}}); + + file_io_ = std::make_shared<NoOpFileIO>(); + + mock_client_ = std::make_shared<MockHttpClient>(); + + ICEBERG_UNWRAP_OR_FAIL(paths_, + ResourcePaths::Make("http://test-server";, /*prefix=*/"", + /*namespace_separator=*/"%1F")); + + session_ = auth::AuthSession::MakeDefault(/*headers=*/{}); + + identifier_ = TableIdentifier{.ns = Namespace{{"default"}}, .name = "my_table"}; + + all_plan_endpoints_ = {Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + } + + // Pass std::nullopt to get the full set of plan endpoints (default). + // Pass an explicit set (including empty) to use exactly that set. + RestScanContext MakeContext( + std::optional<std::unordered_set<Endpoint>> endpoints = std::nullopt) { + auto effective = endpoints.has_value() ? std::move(*endpoints) : all_plan_endpoints_; + return RestScanContext{ + .client = mock_client_, + .paths = paths_, + .session = session_, + .supported_endpoints = std::move(effective), + .identifier = identifier_, + }; + } + + Result<std::unique_ptr<DataTableScan>> MakeScan(RestScanContext ctx) { + return RestTableScan::Make(metadata_, schema_, file_io_, internal::TableScanContext{}, + std::move(ctx)); + } + + std::shared_ptr<Schema> schema_; + std::shared_ptr<TableMetadata> metadata_; + std::shared_ptr<FileIO> file_io_; + std::shared_ptr<MockHttpClient> mock_client_; + std::shared_ptr<ResourcePaths> paths_; + std::shared_ptr<auth::AuthSession> session_; + TableIdentifier identifier_; + std::unordered_set<Endpoint> all_plan_endpoints_; +}; + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED immediately, no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompleted) { + constexpr std::string_view kResponseBody = R"({"status":"completed"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED with a non-empty plan-id (still valid). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompletedWithPlanId) { + constexpr std::string_view kResponseBody = + R"({"status":"completed","plan-id":"plan-abc"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns SUBMITTED → poll returns COMPLETED. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesSubmittedThenCompleted) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-poll-1"})"; + constexpr std::string_view kCompletedBody = R"({"status":"completed"})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + EXPECT_CALL(*mock_client_, Get(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kCompletedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns FAILED → scan returns IOError. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesFailed) { + constexpr std::string_view kFailedBody = + R"({"status":"failed","error":{"message":"server error","type":"ServerError","code":500}})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kFailedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// PlanFiles: PlanTableScan endpoint missing → NotSupported error. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesEndpointNotSupported) { + ICEBERG_UNWRAP_OR_FAIL(auto scan, + MakeScan(MakeContext(std::unordered_set<Endpoint>{}))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// PlanFiles with plan-tasks: server returns COMPLETED with opaque task token, +// then FetchScanTasks is called and returns no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesWithPlanTasks) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-1","plan-tasks":["tok-1"]})"; + // FetchScanTasksResponse requires at least one of plan-tasks or file-scan-tasks + // present. + constexpr std::string_view kTasksResponse = R"({"file-scan-tasks":[]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kTasksResponse)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// Cancel is called when FetchScanTasks fails after a COMPLETED response that +// included plan-tasks. This mirrors the Java cancelPlan-on-close behavior. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelCalledWhenFetchScanTasksFails) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-cancel-1","plan-tasks":["tok-a"]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, "{}"))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// Cancel is a no-op when plan_id is empty (server returned no plan-id). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelIsNoOpWithEmptyPlanId) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-tasks":["tok-b"]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)).Times(0); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// Cancel is a no-op when CancelPlanning endpoint is not advertised. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelIsNoOpWhenEndpointNotAdvertised) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-2","plan-tasks":["tok-c"]})"; + + std::unordered_set<Endpoint> endpoints_without_cancel = { + Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::FetchScanTasks()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)).Times(0); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_cancel))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// FetchPlanningResult: FetchPlanningResult endpoint missing → NotSupported. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, FetchPlanningResultEndpointNotSupported) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-3"})"; + + std::unordered_set<Endpoint> endpoints_without_fetch = { + Endpoint::PlanTableScan(), Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_fetch))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// FetchScanTasks: endpoint missing → NotSupported. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, FetchScanTasksEndpointNotSupported) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-4","plan-tasks":["tok-d"]})"; + + std::unordered_set<Endpoint> endpoints_without_tasks = {Endpoint::PlanTableScan(), + Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, "{}"))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_tasks))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// use_snapshot_schema: UseSnapshot() sets it to true in the builder context. +// RestTableScanBuilder propagates context from DataTableScanBuilder. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, UseSnapshotPropagatesUseSnapshotSchemaInContext) { + constexpr int64_t kSnapshotId = 1000L; + RestTableScanBuilder builder(metadata_, file_io_, "test.my_table", nullptr, + MakeContext(std::nullopt)); + builder.UseSnapshot(kSnapshotId); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder.Build()); + EXPECT_TRUE(scan->context().use_snapshot_schema); +} + +// -------------------------------------------------------------------------- +// use_snapshot_schema: default scan does not set use_snapshot_schema. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, DefaultScanDoesNotSetUseSnapshotSchema) { + RestTableScanBuilder builder(metadata_, file_io_, "test.my_table", nullptr, + MakeContext(std::nullopt)); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder.Build()); + EXPECT_FALSE(scan->context().use_snapshot_schema); +} + +// -------------------------------------------------------------------------- +// Storage credentials in COMPLETED response: effective_io() returns a +// credential-scoped IO, not the original table IO. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, StorageCredentialsInPlanResponseUpdatesEffectiveIO) { + constexpr std::string_view kResponseBody = R"({ + "status": "completed", + "storage-credentials": [ + {"prefix": "s3://bucket/prefix", "config": {"key": "value"}} + ] + })"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); + + auto* rest_scan = dynamic_cast<RestTableScan*>(scan.get()); + ASSERT_NE(rest_scan, nullptr); + // effective_io() must return a credential-scoped IO, not the original file_io_. + EXPECT_NE(rest_scan->effective_io().get(), file_io_.get()); +} + +// -------------------------------------------------------------------------- +// No storage credentials: effective_io() falls back to the table's FileIO. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, NoStorageCredentialsEffectiveIoFallsBackToTableIO) { Review Comment: This uses a fresh scan. It does not catch stale credentials when the same scan is planned twice, first with credentials and then without them. ########## src/iceberg/test/rest_table_scan_test.cc: ########## @@ -0,0 +1,467 @@ +/* + * 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 <memory> +#include <optional> +#include <string> +#include <unordered_map> +#include <unordered_set> + +#include <gmock/gmock.h> +#include <gtest/gtest.h> +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/rest_table.h" +#include "iceberg/file_io.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_scan.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg::rest { + +using ::testing::_; +using ::testing::Return; + +// -------------------------------------------------------------------------- +// Mock HTTP client that overrides the virtual methods of HttpClient. +// The base class constructor creates a cpr::ConnectionPool, which is a +// lightweight allocation (no network connections are opened at construction). +// -------------------------------------------------------------------------- +class MockHttpClient : public HttpClient { + public: + MockHttpClient() : HttpClient({}) {} + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); +}; + +// -------------------------------------------------------------------------- +// Minimal FileIO stub (no real I/O needed for server-side scan planning tests) +// -------------------------------------------------------------------------- +class NoOpFileIO : public FileIO { + public: + Result<std::string> ReadFile(const std::string&, std::optional<size_t>) override { + return IOError("NoOpFileIO"); + } + Status WriteFile(const std::string&, std::string_view) override { return {}; } + Status DeleteFile(const std::string&) override { return {}; } +}; + +// -------------------------------------------------------------------------- +// Test fixture shared by RestTableScan tests. +// -------------------------------------------------------------------------- +class RestTableScanTest : public ::testing::Test { + protected: + void SetUp() override { + schema_ = std::make_shared<Schema>( + std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "data", string())}); + + auto spec = PartitionSpec::Unpartitioned(); + + constexpr int64_t kSnapshotId = 1000L; + auto snapshot = std::make_shared<Snapshot>( + Snapshot{.snapshot_id = kSnapshotId, + .sequence_number = 1L, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L), + .manifest_list = "/tmp/manifest-list.avro", + .schema_id = schema_->schema_id()}); + + metadata_ = std::make_shared<TableMetadata>( + TableMetadata{.format_version = 2, + .table_uuid = "test-uuid", + .location = "/tmp/table", + .last_sequence_number = 1L, + .last_updated_ms = TimePointMsFromUnixMs(1609459200000L), + .last_column_id = 2, + .schemas = {schema_}, + .current_schema_id = schema_->schema_id(), + .partition_specs = {spec}, + .default_spec_id = spec->spec_id(), + .last_partition_id = 999, + .current_snapshot_id = kSnapshotId, + .snapshots = {snapshot}, + .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ + .snapshot_id = kSnapshotId, + .retention = SnapshotRef::Branch{}})}}}); + + file_io_ = std::make_shared<NoOpFileIO>(); + + mock_client_ = std::make_shared<MockHttpClient>(); + + ICEBERG_UNWRAP_OR_FAIL(paths_, + ResourcePaths::Make("http://test-server";, /*prefix=*/"", + /*namespace_separator=*/"%1F")); + + session_ = auth::AuthSession::MakeDefault(/*headers=*/{}); + + identifier_ = TableIdentifier{.ns = Namespace{{"default"}}, .name = "my_table"}; + + all_plan_endpoints_ = {Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + } + + // Pass std::nullopt to get the full set of plan endpoints (default). + // Pass an explicit set (including empty) to use exactly that set. + RestScanContext MakeContext( + std::optional<std::unordered_set<Endpoint>> endpoints = std::nullopt) { + auto effective = endpoints.has_value() ? std::move(*endpoints) : all_plan_endpoints_; + return RestScanContext{ + .client = mock_client_, + .paths = paths_, + .session = session_, + .supported_endpoints = std::move(effective), + .identifier = identifier_, + }; + } + + Result<std::unique_ptr<DataTableScan>> MakeScan(RestScanContext ctx) { + return RestTableScan::Make(metadata_, schema_, file_io_, internal::TableScanContext{}, + std::move(ctx)); + } + + std::shared_ptr<Schema> schema_; + std::shared_ptr<TableMetadata> metadata_; + std::shared_ptr<FileIO> file_io_; + std::shared_ptr<MockHttpClient> mock_client_; + std::shared_ptr<ResourcePaths> paths_; + std::shared_ptr<auth::AuthSession> session_; + TableIdentifier identifier_; + std::unordered_set<Endpoint> all_plan_endpoints_; +}; + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED immediately, no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompleted) { + constexpr std::string_view kResponseBody = R"({"status":"completed"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED with a non-empty plan-id (still valid). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompletedWithPlanId) { + constexpr std::string_view kResponseBody = + R"({"status":"completed","plan-id":"plan-abc"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns SUBMITTED → poll returns COMPLETED. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesSubmittedThenCompleted) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-poll-1"})"; + constexpr std::string_view kCompletedBody = R"({"status":"completed"})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + EXPECT_CALL(*mock_client_, Get(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kCompletedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns FAILED → scan returns IOError. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesFailed) { + constexpr std::string_view kFailedBody = + R"({"status":"failed","error":{"message":"server error","type":"ServerError","code":500}})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kFailedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// PlanFiles: PlanTableScan endpoint missing → NotSupported error. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesEndpointNotSupported) { + ICEBERG_UNWRAP_OR_FAIL(auto scan, + MakeScan(MakeContext(std::unordered_set<Endpoint>{}))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// PlanFiles with plan-tasks: server returns COMPLETED with opaque task token, +// then FetchScanTasks is called and returns no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesWithPlanTasks) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-1","plan-tasks":["tok-1"]})"; + // FetchScanTasksResponse requires at least one of plan-tasks or file-scan-tasks + // present. + constexpr std::string_view kTasksResponse = R"({"file-scan-tasks":[]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kTasksResponse)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// Cancel is called when FetchScanTasks fails after a COMPLETED response that +// included plan-tasks. This mirrors the Java cancelPlan-on-close behavior. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelCalledWhenFetchScanTasksFails) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-cancel-1","plan-tasks":["tok-a"]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, "{}"))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// Cancel is a no-op when plan_id is empty (server returned no plan-id). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelIsNoOpWithEmptyPlanId) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-tasks":["tok-b"]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)).Times(0); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// Cancel is a no-op when CancelPlanning endpoint is not advertised. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelIsNoOpWhenEndpointNotAdvertised) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-2","plan-tasks":["tok-c"]})"; + + std::unordered_set<Endpoint> endpoints_without_cancel = { + Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::FetchScanTasks()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)).Times(0); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_cancel))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// FetchPlanningResult: FetchPlanningResult endpoint missing → NotSupported. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, FetchPlanningResultEndpointNotSupported) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-3"})"; + + std::unordered_set<Endpoint> endpoints_without_fetch = { + Endpoint::PlanTableScan(), Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_fetch))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// FetchScanTasks: endpoint missing → NotSupported. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, FetchScanTasksEndpointNotSupported) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-4","plan-tasks":["tok-d"]})"; + + std::unordered_set<Endpoint> endpoints_without_tasks = {Endpoint::PlanTableScan(), + Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, "{}"))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_tasks))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// use_snapshot_schema: UseSnapshot() sets it to true in the builder context. +// RestTableScanBuilder propagates context from DataTableScanBuilder. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, UseSnapshotPropagatesUseSnapshotSchemaInContext) { + constexpr int64_t kSnapshotId = 1000L; + RestTableScanBuilder builder(metadata_, file_io_, "test.my_table", nullptr, + MakeContext(std::nullopt)); + builder.UseSnapshot(kSnapshotId); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder.Build()); + EXPECT_TRUE(scan->context().use_snapshot_schema); Review Comment: This checks the builder context, not the JSON sent to the server. None of the POST expectations inspect the body, so request fields can be missing and these tests still pass. ########## src/iceberg/test/rest_table_scan_test.cc: ########## @@ -0,0 +1,467 @@ +/* + * 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 <memory> +#include <optional> +#include <string> +#include <unordered_map> +#include <unordered_set> + +#include <gmock/gmock.h> +#include <gtest/gtest.h> +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/rest_table.h" +#include "iceberg/file_io.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_scan.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg::rest { + +using ::testing::_; +using ::testing::Return; + +// -------------------------------------------------------------------------- +// Mock HTTP client that overrides the virtual methods of HttpClient. +// The base class constructor creates a cpr::ConnectionPool, which is a +// lightweight allocation (no network connections are opened at construction). +// -------------------------------------------------------------------------- +class MockHttpClient : public HttpClient { + public: + MockHttpClient() : HttpClient({}) {} + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); +}; + +// -------------------------------------------------------------------------- +// Minimal FileIO stub (no real I/O needed for server-side scan planning tests) +// -------------------------------------------------------------------------- +class NoOpFileIO : public FileIO { + public: + Result<std::string> ReadFile(const std::string&, std::optional<size_t>) override { + return IOError("NoOpFileIO"); + } + Status WriteFile(const std::string&, std::string_view) override { return {}; } + Status DeleteFile(const std::string&) override { return {}; } +}; + +// -------------------------------------------------------------------------- +// Test fixture shared by RestTableScan tests. +// -------------------------------------------------------------------------- +class RestTableScanTest : public ::testing::Test { + protected: + void SetUp() override { + schema_ = std::make_shared<Schema>( + std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "data", string())}); + + auto spec = PartitionSpec::Unpartitioned(); + + constexpr int64_t kSnapshotId = 1000L; + auto snapshot = std::make_shared<Snapshot>( + Snapshot{.snapshot_id = kSnapshotId, + .sequence_number = 1L, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L), + .manifest_list = "/tmp/manifest-list.avro", + .schema_id = schema_->schema_id()}); + + metadata_ = std::make_shared<TableMetadata>( + TableMetadata{.format_version = 2, + .table_uuid = "test-uuid", + .location = "/tmp/table", + .last_sequence_number = 1L, + .last_updated_ms = TimePointMsFromUnixMs(1609459200000L), + .last_column_id = 2, + .schemas = {schema_}, + .current_schema_id = schema_->schema_id(), + .partition_specs = {spec}, + .default_spec_id = spec->spec_id(), + .last_partition_id = 999, + .current_snapshot_id = kSnapshotId, + .snapshots = {snapshot}, + .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ + .snapshot_id = kSnapshotId, + .retention = SnapshotRef::Branch{}})}}}); + + file_io_ = std::make_shared<NoOpFileIO>(); + + mock_client_ = std::make_shared<MockHttpClient>(); + + ICEBERG_UNWRAP_OR_FAIL(paths_, + ResourcePaths::Make("http://test-server";, /*prefix=*/"", + /*namespace_separator=*/"%1F")); + + session_ = auth::AuthSession::MakeDefault(/*headers=*/{}); + + identifier_ = TableIdentifier{.ns = Namespace{{"default"}}, .name = "my_table"}; + + all_plan_endpoints_ = {Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + } + + // Pass std::nullopt to get the full set of plan endpoints (default). + // Pass an explicit set (including empty) to use exactly that set. + RestScanContext MakeContext( + std::optional<std::unordered_set<Endpoint>> endpoints = std::nullopt) { + auto effective = endpoints.has_value() ? std::move(*endpoints) : all_plan_endpoints_; + return RestScanContext{ + .client = mock_client_, + .paths = paths_, + .session = session_, + .supported_endpoints = std::move(effective), + .identifier = identifier_, + }; + } + + Result<std::unique_ptr<DataTableScan>> MakeScan(RestScanContext ctx) { + return RestTableScan::Make(metadata_, schema_, file_io_, internal::TableScanContext{}, + std::move(ctx)); + } + + std::shared_ptr<Schema> schema_; + std::shared_ptr<TableMetadata> metadata_; + std::shared_ptr<FileIO> file_io_; + std::shared_ptr<MockHttpClient> mock_client_; + std::shared_ptr<ResourcePaths> paths_; + std::shared_ptr<auth::AuthSession> session_; + TableIdentifier identifier_; + std::unordered_set<Endpoint> all_plan_endpoints_; +}; + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED immediately, no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompleted) { + constexpr std::string_view kResponseBody = R"({"status":"completed"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); Review Comment: These tests only call `PlanFiles()`. After #873, `PlanFilesStream()` is a separate caller path, so this suite does not catch the local-planning bypass. ########## src/iceberg/catalog/rest/rest_table_scan.cc: ########## @@ -0,0 +1,294 @@ +/* + * 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/rest_file_io.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; + request.use_snapshot_schema = true; + } else if (context_.snapshot_id.has_value()) { + request.snapshot_id = context_.snapshot_id; + request.use_snapshot_schema = context_.use_snapshot_schema; + } + + 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: { + ICEBERG_RETURN_UNEXPECTED(ApplyStorageCredentials(result.storage_credentials)); + auto tasks = ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs); + if (!tasks.has_value()) CancelPlanning(plan_id); + return tasks; + } + case PlanStatus::kSubmitted: + return FetchPlanningResult(plan_id, specs); + case PlanStatus::kFailed: + return IOError("Scan planning failed: {}", + result.error ? result.error->message : "unknown error"); + case PlanStatus::kCancelled: + return IOError("Scan planning was cancelled for plan_id={}", plan_id); + } + return IOError("Unexpected plan status"); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchPlanningResult( + const std::string& plan_id, + const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs) const { + ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints, + Endpoint::FetchPlanningResult()); + + ICEBERG_ASSIGN_OR_RAISE(auto path, + rest_context_.paths->Plan(rest_context_.identifier, plan_id)); + + auto delay_ms = kMinSleepMs; + auto start = std::chrono::steady_clock::now(); + + for (int retry = 0; retry <= kMaxRetries; ++retry) { + ICEBERG_ASSIGN_OR_RAISE( + const auto response, + rest_context_.client->Get(path, /*params=*/{}, /*headers=*/{}, + *PlanErrorHandler::Instance(), *rest_context_.session)); + ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body())); + ICEBERG_ASSIGN_OR_RAISE(auto result, + FetchPlanningResultResponseFromJson(json, specs, *schema_)); + ICEBERG_RETURN_UNEXPECTED(result.Validate()); + + switch (result.plan_status) { + case PlanStatus::kCompleted: { + ICEBERG_RETURN_UNEXPECTED(ApplyStorageCredentials(result.storage_credentials)); + auto tasks = ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs); + if (!tasks.has_value()) CancelPlanning(plan_id); + return tasks; + } + case PlanStatus::kSubmitted: { + auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>( + std::chrono::steady_clock::now() - start) + .count(); + if (elapsed_ms >= kMaxWaitTimeMs) { + CancelPlanning(plan_id); + return IOError("Scan planning timed out after {}ms waiting for plan_id={}", + elapsed_ms, plan_id); + } + std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms)); + delay_ms = std::min(delay_ms * 2, kMaxSleepMs); + continue; + } + case PlanStatus::kFailed: + CancelPlanning(plan_id); + return IOError("Scan planning failed: {}", + result.error ? result.error->message : "unknown error"); + case PlanStatus::kCancelled: + return IOError("Scan planning was cancelled for plan_id={}", plan_id); + } + } + + CancelPlanning(plan_id); + return IOError("Scan planning exceeded max retries ({}) for plan_id={}", kMaxRetries, + plan_id); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchScanTasks( + const std::string& plan_task, + const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs) const { + ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints, Endpoint::FetchScanTasks()); + + ICEBERG_ASSIGN_OR_RAISE(auto path, + rest_context_.paths->FetchScanTasks(rest_context_.identifier)); + FetchScanTasksRequest request{.planTask = plan_task}; + ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request))); + ICEBERG_ASSIGN_OR_RAISE(const auto response, + rest_context_.client->Post(path, json_request, /*headers=*/{}, + *PlanTaskErrorHandler::Instance(), + *rest_context_.session)); + ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body())); + ICEBERG_ASSIGN_OR_RAISE(auto result, + FetchScanTasksResponseFromJson(json, specs, *schema_)); + ICEBERG_RETURN_UNEXPECTED(result.Validate()); + ICEBERG_RETURN_UNEXPECTED(ApplyStorageCredentials(result.storage_credentials)); + + return ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::ResolveScanTasks( + const std::optional<std::vector<std::string>>& plan_tasks, + const std::optional<std::vector<std::shared_ptr<FileScanTask>>>& file_scan_tasks, + const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs) const { + std::vector<std::shared_ptr<FileScanTask>> result; + + if (file_scan_tasks.has_value()) { + result.insert(result.end(), file_scan_tasks->begin(), file_scan_tasks->end()); + } + + if (plan_tasks.has_value()) { + for (const auto& plan_task : *plan_tasks) { + ICEBERG_ASSIGN_OR_RAISE(auto tasks, FetchScanTasks(plan_task, specs)); + result.insert(result.end(), tasks.begin(), tasks.end()); + } + } + + return result; +} + +void RestTableScan::CancelPlanning(const std::string& plan_id) const { + if (plan_id.empty()) return; + if (!rest_context_.supported_endpoints.contains(Endpoint::CancelPlanning())) return; + + auto path = rest_context_.paths->Plan(rest_context_.identifier, plan_id); + if (!path.has_value()) return; + + // Best-effort: ignore errors. + std::ignore = + rest_context_.client->Delete(*path, /*params=*/{}, /*headers=*/{}, + *PlanErrorHandler::Instance(), *rest_context_.session); +} + +const std::shared_ptr<FileIO>& RestTableScan::effective_io() const { + return scan_io_ ? scan_io_ : io_; +} + +Status RestTableScan::ApplyStorageCredentials( + const std::vector<StorageCredential>& credentials) const { + if (credentials.empty()) return {}; Review Comment: An empty credential list leaves `scan_io_` unchanged. If this scan is planned twice and only the first response has credentials, the second result still uses the old credentials. ########## src/iceberg/test/rest_table_scan_test.cc: ########## @@ -0,0 +1,467 @@ +/* + * 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 <memory> +#include <optional> +#include <string> +#include <unordered_map> +#include <unordered_set> + +#include <gmock/gmock.h> +#include <gtest/gtest.h> +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/rest_table.h" +#include "iceberg/file_io.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_scan.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg::rest { + +using ::testing::_; +using ::testing::Return; + +// -------------------------------------------------------------------------- +// Mock HTTP client that overrides the virtual methods of HttpClient. +// The base class constructor creates a cpr::ConnectionPool, which is a +// lightweight allocation (no network connections are opened at construction). +// -------------------------------------------------------------------------- +class MockHttpClient : public HttpClient { + public: + MockHttpClient() : HttpClient({}) {} + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); +}; + +// -------------------------------------------------------------------------- +// Minimal FileIO stub (no real I/O needed for server-side scan planning tests) +// -------------------------------------------------------------------------- +class NoOpFileIO : public FileIO { + public: + Result<std::string> ReadFile(const std::string&, std::optional<size_t>) override { + return IOError("NoOpFileIO"); + } + Status WriteFile(const std::string&, std::string_view) override { return {}; } + Status DeleteFile(const std::string&) override { return {}; } +}; + +// -------------------------------------------------------------------------- +// Test fixture shared by RestTableScan tests. +// -------------------------------------------------------------------------- +class RestTableScanTest : public ::testing::Test { + protected: + void SetUp() override { + schema_ = std::make_shared<Schema>( + std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "data", string())}); + + auto spec = PartitionSpec::Unpartitioned(); + + constexpr int64_t kSnapshotId = 1000L; + auto snapshot = std::make_shared<Snapshot>( + Snapshot{.snapshot_id = kSnapshotId, + .sequence_number = 1L, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L), + .manifest_list = "/tmp/manifest-list.avro", + .schema_id = schema_->schema_id()}); + + metadata_ = std::make_shared<TableMetadata>( + TableMetadata{.format_version = 2, + .table_uuid = "test-uuid", + .location = "/tmp/table", + .last_sequence_number = 1L, + .last_updated_ms = TimePointMsFromUnixMs(1609459200000L), + .last_column_id = 2, + .schemas = {schema_}, + .current_schema_id = schema_->schema_id(), + .partition_specs = {spec}, + .default_spec_id = spec->spec_id(), + .last_partition_id = 999, + .current_snapshot_id = kSnapshotId, + .snapshots = {snapshot}, + .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ + .snapshot_id = kSnapshotId, + .retention = SnapshotRef::Branch{}})}}}); + + file_io_ = std::make_shared<NoOpFileIO>(); + + mock_client_ = std::make_shared<MockHttpClient>(); + + ICEBERG_UNWRAP_OR_FAIL(paths_, + ResourcePaths::Make("http://test-server";, /*prefix=*/"", + /*namespace_separator=*/"%1F")); + + session_ = auth::AuthSession::MakeDefault(/*headers=*/{}); + + identifier_ = TableIdentifier{.ns = Namespace{{"default"}}, .name = "my_table"}; + + all_plan_endpoints_ = {Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + } + + // Pass std::nullopt to get the full set of plan endpoints (default). + // Pass an explicit set (including empty) to use exactly that set. + RestScanContext MakeContext( + std::optional<std::unordered_set<Endpoint>> endpoints = std::nullopt) { + auto effective = endpoints.has_value() ? std::move(*endpoints) : all_plan_endpoints_; + return RestScanContext{ + .client = mock_client_, + .paths = paths_, + .session = session_, + .supported_endpoints = std::move(effective), + .identifier = identifier_, + }; + } + + Result<std::unique_ptr<DataTableScan>> MakeScan(RestScanContext ctx) { + return RestTableScan::Make(metadata_, schema_, file_io_, internal::TableScanContext{}, + std::move(ctx)); + } + + std::shared_ptr<Schema> schema_; + std::shared_ptr<TableMetadata> metadata_; + std::shared_ptr<FileIO> file_io_; + std::shared_ptr<MockHttpClient> mock_client_; + std::shared_ptr<ResourcePaths> paths_; + std::shared_ptr<auth::AuthSession> session_; + TableIdentifier identifier_; + std::unordered_set<Endpoint> all_plan_endpoints_; +}; + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED immediately, no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompleted) { + constexpr std::string_view kResponseBody = R"({"status":"completed"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED with a non-empty plan-id (still valid). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompletedWithPlanId) { + constexpr std::string_view kResponseBody = + R"({"status":"completed","plan-id":"plan-abc"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns SUBMITTED → poll returns COMPLETED. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesSubmittedThenCompleted) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-poll-1"})"; + constexpr std::string_view kCompletedBody = R"({"status":"completed"})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + EXPECT_CALL(*mock_client_, Get(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kCompletedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns FAILED → scan returns IOError. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesFailed) { + constexpr std::string_view kFailedBody = + R"({"status":"failed","error":{"message":"server error","type":"ServerError","code":500}})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kFailedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// PlanFiles: PlanTableScan endpoint missing → NotSupported error. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesEndpointNotSupported) { + ICEBERG_UNWRAP_OR_FAIL(auto scan, + MakeScan(MakeContext(std::unordered_set<Endpoint>{}))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// PlanFiles with plan-tasks: server returns COMPLETED with opaque task token, +// then FetchScanTasks is called and returns no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesWithPlanTasks) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-1","plan-tasks":["tok-1"]})"; + // FetchScanTasksResponse requires at least one of plan-tasks or file-scan-tasks + // present. + constexpr std::string_view kTasksResponse = R"({"file-scan-tasks":[]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kTasksResponse)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// Cancel is called when FetchScanTasks fails after a COMPLETED response that +// included plan-tasks. This mirrors the Java cancelPlan-on-close behavior. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelCalledWhenFetchScanTasksFails) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-cancel-1","plan-tasks":["tok-a"]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, "{}"))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// Cancel is a no-op when plan_id is empty (server returned no plan-id). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelIsNoOpWithEmptyPlanId) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-tasks":["tok-b"]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)).Times(0); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// Cancel is a no-op when CancelPlanning endpoint is not advertised. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelIsNoOpWhenEndpointNotAdvertised) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-2","plan-tasks":["tok-c"]})"; + + std::unordered_set<Endpoint> endpoints_without_cancel = { + Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::FetchScanTasks()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)).Times(0); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_cancel))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// FetchPlanningResult: FetchPlanningResult endpoint missing → NotSupported. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, FetchPlanningResultEndpointNotSupported) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-3"})"; + + std::unordered_set<Endpoint> endpoints_without_fetch = { + Endpoint::PlanTableScan(), Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_fetch))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// FetchScanTasks: endpoint missing → NotSupported. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, FetchScanTasksEndpointNotSupported) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-4","plan-tasks":["tok-d"]})"; + + std::unordered_set<Endpoint> endpoints_without_tasks = {Endpoint::PlanTableScan(), + Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, "{}"))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_tasks))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// use_snapshot_schema: UseSnapshot() sets it to true in the builder context. +// RestTableScanBuilder propagates context from DataTableScanBuilder. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, UseSnapshotPropagatesUseSnapshotSchemaInContext) { + constexpr int64_t kSnapshotId = 1000L; + RestTableScanBuilder builder(metadata_, file_io_, "test.my_table", nullptr, + MakeContext(std::nullopt)); + builder.UseSnapshot(kSnapshotId); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder.Build()); + EXPECT_TRUE(scan->context().use_snapshot_schema); +} + +// -------------------------------------------------------------------------- +// use_snapshot_schema: default scan does not set use_snapshot_schema. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, DefaultScanDoesNotSetUseSnapshotSchema) { + RestTableScanBuilder builder(metadata_, file_io_, "test.my_table", nullptr, + MakeContext(std::nullopt)); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder.Build()); + EXPECT_FALSE(scan->context().use_snapshot_schema); +} + +// -------------------------------------------------------------------------- +// Storage credentials in COMPLETED response: effective_io() returns a +// credential-scoped IO, not the original table IO. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, StorageCredentialsInPlanResponseUpdatesEffectiveIO) { + constexpr std::string_view kResponseBody = R"({ + "status": "completed", + "storage-credentials": [ + {"prefix": "s3://bucket/prefix", "config": {"key": "value"}} + ] + })"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); + + auto* rest_scan = dynamic_cast<RestTableScan*>(scan.get()); Review Comment: This downcast hides the public API problem. Real callers hold a `DataTableScan` and use `io()`, which still returns the original table IO. ########## src/iceberg/test/rest_table_scan_test.cc: ########## @@ -0,0 +1,467 @@ +/* + * 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 <memory> +#include <optional> +#include <string> +#include <unordered_map> +#include <unordered_set> + +#include <gmock/gmock.h> +#include <gtest/gtest.h> +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/rest_table.h" +#include "iceberg/file_io.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_scan.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg::rest { + +using ::testing::_; +using ::testing::Return; + +// -------------------------------------------------------------------------- +// Mock HTTP client that overrides the virtual methods of HttpClient. +// The base class constructor creates a cpr::ConnectionPool, which is a +// lightweight allocation (no network connections are opened at construction). +// -------------------------------------------------------------------------- +class MockHttpClient : public HttpClient { + public: + MockHttpClient() : HttpClient({}) {} + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); + + MOCK_METHOD(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), + (override)); +}; + +// -------------------------------------------------------------------------- +// Minimal FileIO stub (no real I/O needed for server-side scan planning tests) +// -------------------------------------------------------------------------- +class NoOpFileIO : public FileIO { + public: + Result<std::string> ReadFile(const std::string&, std::optional<size_t>) override { + return IOError("NoOpFileIO"); + } + Status WriteFile(const std::string&, std::string_view) override { return {}; } + Status DeleteFile(const std::string&) override { return {}; } +}; + +// -------------------------------------------------------------------------- +// Test fixture shared by RestTableScan tests. +// -------------------------------------------------------------------------- +class RestTableScanTest : public ::testing::Test { + protected: + void SetUp() override { + schema_ = std::make_shared<Schema>( + std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "data", string())}); + + auto spec = PartitionSpec::Unpartitioned(); + + constexpr int64_t kSnapshotId = 1000L; + auto snapshot = std::make_shared<Snapshot>( + Snapshot{.snapshot_id = kSnapshotId, + .sequence_number = 1L, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L), + .manifest_list = "/tmp/manifest-list.avro", + .schema_id = schema_->schema_id()}); + + metadata_ = std::make_shared<TableMetadata>( + TableMetadata{.format_version = 2, + .table_uuid = "test-uuid", + .location = "/tmp/table", + .last_sequence_number = 1L, + .last_updated_ms = TimePointMsFromUnixMs(1609459200000L), + .last_column_id = 2, + .schemas = {schema_}, + .current_schema_id = schema_->schema_id(), + .partition_specs = {spec}, + .default_spec_id = spec->spec_id(), + .last_partition_id = 999, + .current_snapshot_id = kSnapshotId, + .snapshots = {snapshot}, + .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ + .snapshot_id = kSnapshotId, + .retention = SnapshotRef::Branch{}})}}}); + + file_io_ = std::make_shared<NoOpFileIO>(); + + mock_client_ = std::make_shared<MockHttpClient>(); + + ICEBERG_UNWRAP_OR_FAIL(paths_, + ResourcePaths::Make("http://test-server";, /*prefix=*/"", + /*namespace_separator=*/"%1F")); + + session_ = auth::AuthSession::MakeDefault(/*headers=*/{}); + + identifier_ = TableIdentifier{.ns = Namespace{{"default"}}, .name = "my_table"}; + + all_plan_endpoints_ = {Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + } + + // Pass std::nullopt to get the full set of plan endpoints (default). + // Pass an explicit set (including empty) to use exactly that set. + RestScanContext MakeContext( + std::optional<std::unordered_set<Endpoint>> endpoints = std::nullopt) { + auto effective = endpoints.has_value() ? std::move(*endpoints) : all_plan_endpoints_; + return RestScanContext{ + .client = mock_client_, + .paths = paths_, + .session = session_, + .supported_endpoints = std::move(effective), + .identifier = identifier_, + }; + } + + Result<std::unique_ptr<DataTableScan>> MakeScan(RestScanContext ctx) { + return RestTableScan::Make(metadata_, schema_, file_io_, internal::TableScanContext{}, + std::move(ctx)); + } + + std::shared_ptr<Schema> schema_; + std::shared_ptr<TableMetadata> metadata_; + std::shared_ptr<FileIO> file_io_; + std::shared_ptr<MockHttpClient> mock_client_; + std::shared_ptr<ResourcePaths> paths_; + std::shared_ptr<auth::AuthSession> session_; + TableIdentifier identifier_; + std::unordered_set<Endpoint> all_plan_endpoints_; +}; + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED immediately, no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompleted) { + constexpr std::string_view kResponseBody = R"({"status":"completed"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns COMPLETED with a non-empty plan-id (still valid). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesCompletedWithPlanId) { + constexpr std::string_view kResponseBody = + R"({"status":"completed","plan-id":"plan-abc"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns SUBMITTED → poll returns COMPLETED. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesSubmittedThenCompleted) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-poll-1"})"; + constexpr std::string_view kCompletedBody = R"({"status":"completed"})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + EXPECT_CALL(*mock_client_, Get(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kCompletedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// PlanFiles: server returns FAILED → scan returns IOError. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesFailed) { + constexpr std::string_view kFailedBody = + R"({"status":"failed","error":{"message":"server error","type":"ServerError","code":500}})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kFailedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// PlanFiles: PlanTableScan endpoint missing → NotSupported error. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesEndpointNotSupported) { + ICEBERG_UNWRAP_OR_FAIL(auto scan, + MakeScan(MakeContext(std::unordered_set<Endpoint>{}))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// PlanFiles with plan-tasks: server returns COMPLETED with opaque task token, +// then FetchScanTasks is called and returns no file scan tasks. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, PlanFilesWithPlanTasks) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-1","plan-tasks":["tok-1"]})"; + // FetchScanTasksResponse requires at least one of plan-tasks or file-scan-tasks + // present. + constexpr std::string_view kTasksResponse = R"({"file-scan-tasks":[]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kTasksResponse)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); +} + +// -------------------------------------------------------------------------- +// Cancel is called when FetchScanTasks fails after a COMPLETED response that +// included plan-tasks. This mirrors the Java cancelPlan-on-close behavior. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelCalledWhenFetchScanTasksFails) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-cancel-1","plan-tasks":["tok-a"]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, "{}"))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// Cancel is a no-op when plan_id is empty (server returned no plan-id). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelIsNoOpWithEmptyPlanId) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-tasks":["tok-b"]})"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)).Times(0); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// Cancel is a no-op when CancelPlanning endpoint is not advertised. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, CancelIsNoOpWhenEndpointNotAdvertised) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-2","plan-tasks":["tok-c"]})"; + + std::unordered_set<Endpoint> endpoints_without_cancel = { + Endpoint::PlanTableScan(), Endpoint::FetchPlanningResult(), + Endpoint::FetchScanTasks()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(IOError("FetchScanTasks failed"))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)).Times(0); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_cancel))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); +} + +// -------------------------------------------------------------------------- +// FetchPlanningResult: FetchPlanningResult endpoint missing → NotSupported. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, FetchPlanningResultEndpointNotSupported) { + constexpr std::string_view kSubmittedBody = + R"({"status":"submitted","plan-id":"plan-3"})"; + + std::unordered_set<Endpoint> endpoints_without_fetch = { + Endpoint::PlanTableScan(), Endpoint::CancelPlanning(), Endpoint::FetchScanTasks()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kSubmittedBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_fetch))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// FetchScanTasks: endpoint missing → NotSupported. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, FetchScanTasksEndpointNotSupported) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-4","plan-tasks":["tok-d"]})"; + + std::unordered_set<Endpoint> endpoints_without_tasks = {Endpoint::PlanTableScan(), + Endpoint::FetchPlanningResult(), + Endpoint::CancelPlanning()}; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))); + EXPECT_CALL(*mock_client_, Delete(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, "{}"))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext(endpoints_without_tasks))); + auto result = scan->PlanFiles(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); +} + +// -------------------------------------------------------------------------- +// use_snapshot_schema: UseSnapshot() sets it to true in the builder context. +// RestTableScanBuilder propagates context from DataTableScanBuilder. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, UseSnapshotPropagatesUseSnapshotSchemaInContext) { + constexpr int64_t kSnapshotId = 1000L; + RestTableScanBuilder builder(metadata_, file_io_, "test.my_table", nullptr, + MakeContext(std::nullopt)); + builder.UseSnapshot(kSnapshotId); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder.Build()); + EXPECT_TRUE(scan->context().use_snapshot_schema); +} + +// -------------------------------------------------------------------------- +// use_snapshot_schema: default scan does not set use_snapshot_schema. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, DefaultScanDoesNotSetUseSnapshotSchema) { + RestTableScanBuilder builder(metadata_, file_io_, "test.my_table", nullptr, + MakeContext(std::nullopt)); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder.Build()); + EXPECT_FALSE(scan->context().use_snapshot_schema); +} + +// -------------------------------------------------------------------------- +// Storage credentials in COMPLETED response: effective_io() returns a +// credential-scoped IO, not the original table IO. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, StorageCredentialsInPlanResponseUpdatesEffectiveIO) { + constexpr std::string_view kResponseBody = R"({ + "status": "completed", + "storage-credentials": [ + {"prefix": "s3://bucket/prefix", "config": {"key": "value"}} + ] + })"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); + + auto* rest_scan = dynamic_cast<RestTableScan*>(scan.get()); + ASSERT_NE(rest_scan, nullptr); + // effective_io() must return a credential-scoped IO, not the original file_io_. + EXPECT_NE(rest_scan->effective_io().get(), file_io_.get()); +} + +// -------------------------------------------------------------------------- +// No storage credentials: effective_io() falls back to the table's FileIO. +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, NoStorageCredentialsEffectiveIoFallsBackToTableIO) { + constexpr std::string_view kResponseBody = R"({"status":"completed"})"; + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kResponseBody)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); + + auto* rest_scan = dynamic_cast<RestTableScan*>(scan.get()); + ASSERT_NE(rest_scan, nullptr); + EXPECT_EQ(rest_scan->effective_io().get(), file_io_.get()); +} + +// -------------------------------------------------------------------------- +// Storage credentials returned in FetchScanTasksResponse also update +// effective_io(). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, StorageCredentialsInFetchScanTasksResponseUpdatesEffectiveIO) { + constexpr std::string_view kPlanResponse = + R"({"status":"completed","plan-id":"plan-cred","plan-tasks":["tok-cred"]})"; + constexpr std::string_view kTasksResponse = R"({ + "file-scan-tasks": [], + "storage-credentials": [ + {"prefix": "s3://bucket/prefix", "config": {"key": "value"}} + ] + })"; + + EXPECT_CALL(*mock_client_, Post(_, _, _, _, _)) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kPlanResponse)))) + .WillOnce(Return(HttpResponse::MakeForTesting(200, std::string(kTasksResponse)))); + + ICEBERG_UNWRAP_OR_FAIL(auto scan, MakeScan(MakeContext())); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + EXPECT_TRUE(tasks.empty()); + + auto* rest_scan = dynamic_cast<RestTableScan*>(scan.get()); + ASSERT_NE(rest_scan, nullptr); + EXPECT_NE(rest_scan->effective_io().get(), file_io_.get()); +} + +// -------------------------------------------------------------------------- +// RestTable::NewScan returns a RestTableScanBuilder (not a plain builder). +// -------------------------------------------------------------------------- +TEST_F(RestTableScanTest, RestTableNewScanReturnsRestTableScanBuilder) { Review Comment: This builds `RestTable` directly. It does not cover `RestCatalog::LoadTable()` choosing the scan type from client and table `scan-planning-mode`, or the missing-endpoint error. -- 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]
