feat(rest): add scan plan endpoint support to REST catalog client

When a table is loaded from a REST catalog that advertises the PlanTableScan
endpoint, NewScan() now returns a RestTableScanBuilder whose Build() produces
a RestTableScan. PlanFiles() on that scan delegates manifest resolution to
the server via POST /plan, GET /plan/{id} (with exponential backoff),
POST /tasks/{id}, and DELETE /plan/{id} (best-effort cancel), instead of
reading manifests locally.

- Add RestTable, RestTableScanBuilder, RestTableScan and RestScanContext
- Promote DataTableScan::PlanFiles and TableScanBuilder::Build to virtual
- Convert RestCatalog::client_ and paths_ to shared_ptr so RestScanContext
  can share ownership with live scans

Author: Sandeep Gottimukkala

src/iceberg/catalog/rest/CMakeLists.txt

@@ -33,6 +33,8 @@ set(ICEBERG_REST_SOURCES
     resource_paths.cc
     rest_catalog.cc
     rest_file_io.cc
+    rest_table.cc
+    rest_table_scan.cc
     rest_util.cc
     types.cc)
 

src/iceberg/catalog/rest/meson.build

@@ -30,6 +30,8 @@ iceberg_rest_sources = files(
     'resource_paths.cc',
     'rest_catalog.cc',
     'rest_file_io.cc',
+    'rest_table.cc',
+    'rest_table_scan.cc',
     'rest_util.cc',
     'types.cc',
 )
@@ -94,6 +96,8 @@ install_headers(
         'resource_paths.h',
         'rest_catalog.h',
         'rest_file_io.h',
+        'rest_table.h',
+        'rest_table_scan.h',
         'rest_util.h',
         'type_fwd.h',
         'types.h',

src/iceberg/catalog/rest/rest_catalog.cc

@@ -39,6 +39,7 @@
 #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/rest_table.h"
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
@@ -451,7 +452,7 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
   // Get snapshot loading mode
   ICEBERG_ASSIGN_OR_RAISE(auto snapshot_mode, final_config.SnapshotLoadingMode());
 
-  auto client = std::make_unique<HttpClient>(final_config.ExtractHeaders());
+  auto client = std::make_shared<HttpClient>(final_config.ExtractHeaders());
   ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
                           auth_manager->CatalogSession(*client, final_config.configs()));
 
@@ -466,8 +467,8 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
 }
 
 RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
-                         std::unique_ptr<HttpClient> client,
-                         std::unique_ptr<ResourcePaths> paths,
+                         std::shared_ptr<HttpClient> client,
+                         std::shared_ptr<ResourcePaths> paths,
                          std::unordered_set<Endpoint> endpoints,
                          std::unique_ptr<auth::AuthManager> auth_manager,
                          std::shared_ptr<auth::AuthSession> catalog_session,
@@ -873,6 +874,20 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, table_session);
+
+  if (supported_endpoints_.contains(Endpoint::PlanTableScan())) {
+    RestScanContext rest_ctx{
+        .client = client_,
+        .paths = paths_,
+        .session = table_session,
+        .supported_endpoints = supported_endpoints_,
+        .identifier = identifier,
+    };
+    return RestTable::Make(identifier, std::move(result.metadata),
+                           std::move(result.metadata_location), std::move(table_io),
+                           std::move(table_catalog), std::move(rest_ctx));
+  }
+
   return Table::Make(identifier, std::move(result.metadata),
                      std::move(result.metadata_location), std::move(table_io),
                      std::move(table_catalog));

src/iceberg/catalog/rest/rest_catalog.h

@@ -63,7 +63,7 @@ class ICEBERG_REST_EXPORT RestCatalog final
   class TableScopedCatalog;
 
   RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
-              std::unique_ptr<HttpClient> client, std::unique_ptr<ResourcePaths> paths,
+              std::shared_ptr<HttpClient> client, std::shared_ptr<ResourcePaths> paths,
               std::unordered_set<Endpoint> endpoints,
               std::unique_ptr<auth::AuthManager> auth_manager,
               std::shared_ptr<auth::AuthSession> catalog_session,
@@ -173,8 +173,8 @@ class ICEBERG_REST_EXPORT RestCatalog final
 
   RestCatalogProperties config_;
   std::shared_ptr<FileIO> file_io_;
-  std::unique_ptr<HttpClient> client_;
-  std::unique_ptr<ResourcePaths> paths_;
+  std::shared_ptr<HttpClient> client_;
+  std::shared_ptr<ResourcePaths> paths_;
   std::string name_;
   std::unordered_set<Endpoint> supported_endpoints_;
   std::unique_ptr<auth::AuthManager> auth_manager_;

src/iceberg/catalog/rest/rest_table.cc

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/rest_table.h"
+
+#include <memory>
+#include <utility>
+
+#include "iceberg/catalog/rest/rest_table_scan.h"
+#include "iceberg/result.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+RestTable::RestTable(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
+                     std::string metadata_location, std::shared_ptr<FileIO> io,
+                     std::shared_ptr<Catalog> catalog, RestScanContext rest_context)
+    : Table(std::move(identifier), std::move(metadata), std::move(metadata_location),
+            std::move(io), std::move(catalog)),
+      rest_context_(std::move(rest_context)) {}
+
+RestTable::~RestTable() = default;
+
+Result<std::shared_ptr<RestTable>> RestTable::Make(TableIdentifier identifier,
+                                                   std::shared_ptr<TableMetadata> metadata,
+                                                   std::string metadata_location,
+                                                   std::shared_ptr<FileIO> io,
+                                                   std::shared_ptr<Catalog> catalog,
+                                                   RestScanContext rest_context) {
+  if (metadata == nullptr) {
+    return InvalidArgument("Metadata cannot be null");
+  }
+  return std::shared_ptr<RestTable>(
+      new RestTable(std::move(identifier), std::move(metadata),
+                    std::move(metadata_location), std::move(io), std::move(catalog),
+                    std::move(rest_context)));
+}
+
+Result<std::unique_ptr<DataTableScanBuilder>> RestTable::NewScan() const {
+  return std::make_unique<RestTableScanBuilder>(metadata_, io_, rest_context_);
+}
+
+}  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_table.h

@@ -0,0 +1,61 @@
+/*
+ * 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/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,
+                                                 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;
+
+ private:
+  RestTable(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
+            std::string metadata_location, std::shared_ptr<FileIO> io,
+            std::shared_ptr<Catalog> catalog, RestScanContext rest_context);
+
+  RestScanContext rest_context_;
+};
+
+}  // namespace iceberg::rest