refactor(rest): move scan plan catalog impl out of this branch

Remove PlanTableScan, FetchPlanningResult, CancelPlanning, and
FetchScanTasks from Catalog, RestCatalog, and their tests. This branch
now contains only the REST models (types.h), ser/de, and error handlers
for scan planning. The catalog-layer wiring will live in a separate branch.

Author: Sandeep Gottimukkala

src/iceberg/catalog.h

@@ -26,10 +26,8 @@
 #include <unordered_set>
 #include <vector>
 
-#include "iceberg/catalog/rest/types.h"
 #include "iceberg/result.h"
 #include "iceberg/table_identifier.h"
-#include "iceberg/table_scan.h"
 #include "iceberg/type_fwd.h"
 
 namespace iceberg {
@@ -190,44 +188,6 @@ class ICEBERG_EXPORT Catalog {
   /// \return a Table instance or ErrorKind::kAlreadyExists if the table already exists
   virtual Result<std::shared_ptr<Table>> RegisterTable(
       const TableIdentifier& identifier, const std::string& metadata_file_location) = 0;
-
-  /// \brief Initiate a scan planning operation for the given table.
-  ///
-  /// \param table The table to scan.
-  /// \param context The scan context containing snapshot, filter, and other options.
-  /// \return A PlanTableScanResponse with the plan status and initial scan tasks.
-  virtual Result<rest::PlanTableScanResponse> PlanTableScan(
-      const Table& table, const internal::TableScanContext& context) {
-    return NotImplemented("PlanTableScan is not supported by this catalog");
-  }
-
-  /// \brief Fetch the current status and results of an asynchronous scan plan.
-  ///
-  /// \param table The table being scanned.
-  /// \param plan_id The plan ID returned by PlanTableScan.
-  /// \return A FetchPlanningResultResponse with the current plan status and tasks.
-  virtual Result<rest::FetchPlanningResultResponse> FetchPlanningResult(
-      const Table& table, const std::string& plan_id) {
-    return NotImplemented("FetchPlanningResult is not supported by this catalog");
-  }
-
-  /// \brief Cancel an in-progress scan planning operation.
-  ///
-  /// \param table The table being scanned.
-  /// \param plan_id The plan ID returned by PlanTableScan.
-  virtual Status CancelPlanning(const Table& table, const std::string& plan_id) {
-    return NotImplemented("CancelPlanning is not supported by this catalog");
-  }
-
-  /// \brief Fetch the scan tasks for a given plan task token.
-  ///
-  /// \param table The table being scanned.
-  /// \param plan_task The plan task token returned in a scan plan response.
-  /// \return A FetchScanTasksResponse with the file scan tasks.
-  virtual Result<rest::FetchScanTasksResponse> FetchScanTasks(
-      const Table& table, const std::string& plan_task) {
-    return NotImplemented("FetchScanTasks is not supported by this catalog");
-  }
 };
 
 }  // namespace iceberg

src/iceberg/catalog/rest/rest_catalog.cc

@@ -43,7 +43,6 @@
 #include "iceberg/sort_order.h"
 #include "iceberg/table.h"
 #include "iceberg/table_requirement.h"
-#include "iceberg/table_scan.h"
 #include "iceberg/table_update.h"
 #include "iceberg/transaction.h"
 #include "iceberg/util/macros.h"
@@ -63,8 +62,6 @@ std::unordered_set<Endpoint> GetDefaultEndpoints() {
       Endpoint::UpdateTable(),     Endpoint::DeleteTable(),
       Endpoint::RenameTable(),     Endpoint::RegisterTable(),
       Endpoint::ReportMetrics(),   Endpoint::CommitTransaction(),
-      Endpoint::PlanTableScan(),   Endpoint::FetchPlanningResult(),
-      Endpoint::CancelPlanning(),  Endpoint::FetchScanTasks(),
   };
 }
 
@@ -498,85 +495,4 @@ Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
                      shared_from_this());
 }
 
-Result<PlanTableScanResponse> RestCatalog::PlanTableScan(
-    const Table& table, const internal::TableScanContext& context) {
-  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::PlanTableScan());
-  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Plan(table.name()));
-  ICEBERG_ASSIGN_OR_RAISE(auto schema_ptr, table.schema());
-  ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, table.specs());
-
-  PlanTableScanRequest request;
-  if (context.snapshot_id.has_value()) {
-    request.snapshot_id = context.snapshot_id;
-  }
-  request.select = context.selected_columns;
-  request.filter = context.filter;
-  request.case_sensitive = context.case_sensitive;
-  request.use_snapshot_schema = false;
-  if (context.from_snapshot_id.has_value() && context.to_snapshot_id.has_value()) {
-    request.start_snapshot_id = context.from_snapshot_id.value();
-    request.end_snapshot_id = context.to_snapshot_id.value();
-    request.use_snapshot_schema = true;
-  }
-  if (context.min_rows_requested.has_value()) {
-    request.min_rows_requested = context.min_rows_requested.value();
-  }
-
-  ICEBERG_RETURN_UNEXPECTED(request.Validate());
-  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,
-      client_->Post(path, json_request, /*headers=*/{}, *PlanErrorHandler::Instance(),
-                    *catalog_session_));
-
-  ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
-  return PlanTableScanResponseFromJson(json, specs_ref.get(), *schema_ptr);
-}
-
-Result<FetchPlanningResultResponse> RestCatalog::FetchPlanningResult(
-    const Table& table, const std::string& plan_id) {
-  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::FetchPlanningResult());
-  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Plan(table.name(), plan_id));
-  ICEBERG_ASSIGN_OR_RAISE(auto schema_ptr, table.schema());
-  ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, table.specs());
-
-  ICEBERG_ASSIGN_OR_RAISE(
-      const auto response,
-      client_->Get(path, /*params=*/{}, /*headers=*/{}, *PlanErrorHandler::Instance(),
-                   *catalog_session_));
-  ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
-  return FetchPlanningResultResponseFromJson(json, specs_ref.get(), *schema_ptr);
-}
-
-Status RestCatalog::CancelPlanning(const Table& table, const std::string& plan_id) {
-  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::CancelPlanning());
-  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Plan(table.name(), plan_id));
-
-  ICEBERG_ASSIGN_OR_RAISE(
-      const auto response,
-      client_->Delete(path, /*params=*/{}, /*headers=*/{},
-                      *PlanErrorHandler::Instance(), *catalog_session_));
-  return {};
-}
-
-Result<FetchScanTasksResponse> RestCatalog::FetchScanTasks(const Table& table,
-                                                           const std::string& plan_task) {
-  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::FetchScanTasks());
-  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->FetchScanTasks(table.name()));
-  ICEBERG_ASSIGN_OR_RAISE(auto schema_ptr, table.schema());
-  ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, table.specs());
-
-  FetchScanTasksRequest request{.planTask = plan_task};
-  ICEBERG_RETURN_UNEXPECTED(request.Validate());
-  ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
-  ICEBERG_ASSIGN_OR_RAISE(
-      const auto response,
-      client_->Post(path, json_request, /*headers=*/{}, *PlanTaskErrorHandler::Instance(),
-                    *catalog_session_));
-
-  ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
-  return FetchScanTasksResponseFromJson(json, specs_ref.get(), *schema_ptr);
-}
-
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_catalog.h

@@ -104,14 +104,6 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
       const TableIdentifier& identifier,
       const std::string& metadata_file_location) override;
 
-  Result<PlanTableScanResponse> PlanTableScan(
-      const Table& table, const internal::TableScanContext& context) override;
-  Result<FetchPlanningResultResponse> FetchPlanningResult(
-      const Table& table, const std::string& plan_id) override;
-  Status CancelPlanning(const Table& table, const std::string& plan_id) override;
-  Result<FetchScanTasksResponse> FetchScanTasks(const Table& table,
-                                                const std::string& plan_task) override;
-
  private:
   RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
               std::unique_ptr<HttpClient> client, std::unique_ptr<ResourcePaths> paths,

src/iceberg/test/endpoint_test.cc

@@ -145,31 +145,6 @@ TEST(EndpointTest, TransactionEndpoints) {
   EXPECT_EQ(commit_transaction.path(), "/v1/{prefix}/transactions/commit");
 }
 
-// Test predefined scan planning endpoints
-TEST(EndpointTest, ScanPlanEndpoints) {
-  auto plan_table_scan = Endpoint::PlanTableScan();
-  EXPECT_EQ(plan_table_scan.method(), HttpMethod::kPost);
-  EXPECT_EQ(plan_table_scan.path(),
-            "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan");
-  EXPECT_EQ(plan_table_scan.ToString(),
-            "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan");
-
-  auto fetch_planning_result = Endpoint::FetchPlanningResult();
-  EXPECT_EQ(fetch_planning_result.method(), HttpMethod::kGet);
-  EXPECT_EQ(fetch_planning_result.path(),
-            "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}");
-
-  auto cancel_planning = Endpoint::CancelPlanning();
-  EXPECT_EQ(cancel_planning.method(), HttpMethod::kDelete);
-  EXPECT_EQ(cancel_planning.path(),
-            "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}");
-
-  auto fetch_scan_tasks = Endpoint::FetchScanTasks();
-  EXPECT_EQ(fetch_scan_tasks.method(), HttpMethod::kPost);
-  EXPECT_EQ(fetch_scan_tasks.path(),
-            "/v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks");
-}
-
 // Test endpoint equality
 TEST(EndpointTest, Equality) {
   auto endpoint1 = Endpoint::Make(HttpMethod::kGet, "/path");
@@ -267,8 +242,6 @@ TEST(EndpointTest, StringRoundTrip) {
       Endpoint::ListNamespaces(),  Endpoint::GetNamespaceProperties(),
       Endpoint::CreateNamespace(), Endpoint::LoadTable(),
       Endpoint::CreateTable(),     Endpoint::DeleteTable(),
-      Endpoint::PlanTableScan(),   Endpoint::FetchPlanningResult(),
-      Endpoint::CancelPlanning(),  Endpoint::FetchScanTasks(),
   };
 
   for (const auto& original : endpoints) {

src/iceberg/test/rest_catalog_integration_test.cc

@@ -46,7 +46,6 @@
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_requirement.h"
-#include "iceberg/table_scan.h"
 #include "iceberg/table_update.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/std_io.h"
@@ -449,83 +448,6 @@ TEST_F(RestCatalogIntegrationTest, StageCreateTable) {
   EXPECT_EQ(committed->metadata()->properties.configs().at("key1"), "value1");
 }
 
-// -- Scan plan operations --
-
-TEST_F(RestCatalogIntegrationTest, PlanTableScan) {
-  Namespace ns{.levels = {"test_plan_table_scan"}};
-  ICEBERG_UNWRAP_OR_FAIL(auto catalog, CreateCatalogAndNamespace(ns));
-  TableIdentifier table_id{.ns = ns, .name = "scan_table"};
-  ASSERT_THAT(CreateDefaultTable(catalog, table_id), IsOk());
-  ICEBERG_UNWRAP_OR_FAIL(auto table, catalog->LoadTable(table_id));
-
-  internal::TableScanContext context;
-  ICEBERG_UNWRAP_OR_FAIL(auto response, catalog->PlanTableScan(*table, context));
-  // Empty table: no file scan tasks returned
-  EXPECT_TRUE(response.file_scan_tasks.empty());
-}
-
-TEST_F(RestCatalogIntegrationTest, PlanTableScanWithContext) {
-  Namespace ns{.levels = {"test_plan_scan_context"}};
-  ICEBERG_UNWRAP_OR_FAIL(auto catalog, CreateCatalogAndNamespace(ns));
-  TableIdentifier table_id{.ns = ns, .name = "context_scan_table"};
-  ASSERT_THAT(CreateDefaultTable(catalog, table_id), IsOk());
-  ICEBERG_UNWRAP_OR_FAIL(auto table, catalog->LoadTable(table_id));
-
-  internal::TableScanContext context;
-  context.selected_columns = {"id", "data"};
-  context.case_sensitive = true;
-  ICEBERG_UNWRAP_OR_FAIL(auto response, catalog->PlanTableScan(*table, context));
-  EXPECT_TRUE(response.file_scan_tasks.empty());
-}
-
-TEST_F(RestCatalogIntegrationTest, FetchPlanningResult) {
-  Namespace ns{.levels = {"test_fetch_planning_result"}};
-  ICEBERG_UNWRAP_OR_FAIL(auto catalog, CreateCatalogAndNamespace(ns));
-  TableIdentifier table_id{.ns = ns, .name = "fetch_plan_table"};
-  ASSERT_THAT(CreateDefaultTable(catalog, table_id), IsOk());
-  ICEBERG_UNWRAP_OR_FAIL(auto table, catalog->LoadTable(table_id));
-
-  internal::TableScanContext context;
-  ICEBERG_UNWRAP_OR_FAIL(auto plan_response, catalog->PlanTableScan(*table, context));
-
-  // NOTE: apache/iceberg-rest-fixture always responds synchronously (status="completed")
-  // with a non-empty plan_id (e.g. "sync-<uuid>"). Sync plans are immediately discarded
-  // server-side. Async paths are covered by unit tests.
-  EXPECT_FALSE(plan_response.plan_id.empty());
-}
-
-TEST_F(RestCatalogIntegrationTest, CancelPlanning) {
-  Namespace ns{.levels = {"test_cancel_planning"}};
-  ICEBERG_UNWRAP_OR_FAIL(auto catalog, CreateCatalogAndNamespace(ns));
-  TableIdentifier table_id{.ns = ns, .name = "cancel_plan_table"};
-  ASSERT_THAT(CreateDefaultTable(catalog, table_id), IsOk());
-  ICEBERG_UNWRAP_OR_FAIL(auto table, catalog->LoadTable(table_id));
-
-  internal::TableScanContext context;
-  ICEBERG_UNWRAP_OR_FAIL(auto plan_response, catalog->PlanTableScan(*table, context));
-
-  // NOTE: apache/iceberg-rest-fixture always responds synchronously — sync plan_id is
-  // accepted for cancellation (idempotent). Async paths are covered by unit tests.
-  ASSERT_FALSE(plan_response.plan_id.empty());
-  ASSERT_THAT(catalog->CancelPlanning(*table, plan_response.plan_id), IsOk());
-}
-
-TEST_F(RestCatalogIntegrationTest, FetchScanTasks) {
-  Namespace ns{.levels = {"test_fetch_scan_tasks"}};
-  ICEBERG_UNWRAP_OR_FAIL(auto catalog, CreateCatalogAndNamespace(ns));
-  TableIdentifier table_id{.ns = ns, .name = "fetch_tasks_table"};
-  ASSERT_THAT(CreateDefaultTable(catalog, table_id), IsOk());
-  ICEBERG_UNWRAP_OR_FAIL(auto table, catalog->LoadTable(table_id));
-
-  internal::TableScanContext context;
-  ICEBERG_UNWRAP_OR_FAIL(auto plan_response, catalog->PlanTableScan(*table, context));
-
-  // NOTE: apache/iceberg-rest-fixture always responds synchronously — plan_tasks is
-  // always empty. FetchScanTasks is only relevant for async plans; async paths are
-  // covered by unit tests.
-  EXPECT_TRUE(plan_response.plan_tasks.empty());
-}
-
 // -- Snapshot loading mode --
 
 TEST_F(RestCatalogIntegrationTest, LoadTableWithSnapshotModeAll) {