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

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

- Add RestTable, RestTableScanBuilder, RestTableScan and RestScanContext
- Promote DataTableScan::PlanFilesStream, TableScan::io and TableScanBuilder::Build to virtual
- Convert RestCatalog::client_ and paths_ to shared_ptr so RestScanContext can share ownership with live scans
- Add REST incremental append planning for scans with an exclusive start snapshot; scans without one (including an inclusive root snapshot) return NotSupported
- Use storage credentials returned by planning responses in a scan-scoped FileIO

Author: gsandeep1241

src/iceberg/catalog/rest/CMakeLists.txt

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

src/iceberg/catalog/rest/catalog_properties.cc

@@ -20,6 +20,8 @@
 #include "iceberg/catalog/rest/catalog_properties.h"
 
 #include <algorithm>
+#include <cstdint>
+#include <optional>
 #include <string>
 #include <string_view>
 
@@ -61,4 +63,14 @@ Result<SnapshotMode> RestCatalogProperties::SnapshotLoadingMode() const {
   }
 }
 
+Result<std::optional<ScanPlanningMode>> RestCatalogProperties::ScanPlanningModeFrom(
+    const std::unordered_map<std::string, std::string>& config) {
+  auto it = config.find(kScanPlanningMode.key());
+  if (it == config.end()) return std::nullopt;
+  std::string lower = StringUtils::ToLower(it->second);
+  if (lower == "client") return ScanPlanningMode::kClient;
+  if (lower == "server") return ScanPlanningMode::kServer;
+  return InvalidArgument("Invalid scan planning mode: '{}'.", it->second);
+}
+
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/catalog_properties.h

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <string>
 #include <unordered_map>
 
@@ -34,6 +35,9 @@ namespace iceberg::rest {
 /// \brief Snapshot loading mode for REST catalog.
 enum class SnapshotMode : uint8_t { kAll, kRefs };
 
+/// \brief Scan planning mode for REST catalog.
+enum class ScanPlanningMode : uint8_t { kClient, kServer };
+
 /// \brief Configuration class for a REST Catalog.
 class ICEBERG_REST_EXPORT RestCatalogProperties
     : public ConfigBase<RestCatalogProperties> {
@@ -58,6 +62,14 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
   /// \brief Whether to report metrics to the REST catalog server (default: true).
   inline static Entry<std::string> kMetricsReportingEnabled{
       "rest-metrics-reporting-enabled", "true"};
+  /// \brief The scan planning mode (client or server).
+  inline static Entry<std::string> kScanPlanningMode{"scan-planning-mode", "client"};
+  /// \brief Maximum time to wait for a submitted scan plan, in milliseconds.
+  inline static Entry<int64_t> kScanPlanningPollTimeoutMs{
+      "rest-scan-planning.poll-timeout-ms", 300000};
+  /// \brief Extra polls after the first fetch of a submitted scan plan.
+  inline static Entry<int32_t> kScanPlanningPollNumRetries{
+      "rest-scan-planning.poll-num-retries", 10};
   /// \brief The prefix for HTTP headers.
   inline static constexpr std::string_view kHeaderPrefix = "header.";
 
@@ -80,6 +92,11 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
   ///         "REFS", or an error if the value is invalid. Parsing is
   ///         case-insensitive to match Java behavior.
   Result<SnapshotMode> SnapshotLoadingMode() const;
+
+  /// \brief Get the scan planning mode from the given config map, returning
+  ///        std::nullopt if the key is absent.
+  static Result<std::optional<ScanPlanningMode>> ScanPlanningModeFrom(
+      const std::unordered_map<std::string, std::string>& config);
 };
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/http_client.cc

@@ -50,6 +50,14 @@ class HttpResponse::Impl {
 };
 
 HttpResponse::HttpResponse() = default;
+
+HttpResponse::HttpResponse(int32_t status_code, std::string body) {
+  cpr::Response cpr_response;
+  cpr_response.status_code = status_code;
+  cpr_response.text = std::move(body);
+  impl_ = std::make_unique<HttpResponse::Impl>(std::move(cpr_response));
+}
+
 HttpResponse::~HttpResponse() = default;
 HttpResponse::HttpResponse(HttpResponse&&) noexcept = default;
 HttpResponse& HttpResponse::operator=(HttpResponse&&) noexcept = default;

src/iceberg/catalog/rest/http_client.h

@@ -45,6 +45,7 @@ namespace iceberg::rest {
 class ICEBERG_REST_EXPORT HttpResponse {
  public:
   HttpResponse();
+  HttpResponse(int32_t status_code, std::string body);
   ~HttpResponse();
 
   HttpResponse(const HttpResponse&) = delete;
@@ -71,44 +72,43 @@ class ICEBERG_REST_EXPORT HttpResponse {
 class ICEBERG_REST_EXPORT HttpClient {
  public:
   explicit HttpClient(std::unordered_map<std::string, std::string> default_headers = {});
-  ~HttpClient();
+  virtual ~HttpClient();
 
   HttpClient(const HttpClient&) = delete;
   HttpClient& operator=(const HttpClient&) = delete;
   HttpClient(HttpClient&&) = delete;
   HttpClient& operator=(HttpClient&&) = delete;
 
   /// \brief Sends a GET request.
-  Result<HttpResponse> Get(const std::string& path,
-                           const std::unordered_map<std::string, std::string>& params,
-                           const std::unordered_map<std::string, std::string>& headers,
-                           const ErrorHandler& error_handler, auth::AuthSession& session);
+  virtual Result<HttpResponse> Get(
+      const std::string& path, const std::unordered_map<std::string, std::string>& params,
+      const std::unordered_map<std::string, std::string>& headers,
+      const ErrorHandler& error_handler, auth::AuthSession& session);
 
   /// \brief Sends a POST request.
-  Result<HttpResponse> Post(const std::string& path, const std::string& body,
-                            const std::unordered_map<std::string, std::string>& headers,
-                            const ErrorHandler& error_handler,
-                            auth::AuthSession& session);
+  virtual Result<HttpResponse> Post(
+      const std::string& path, const std::string& body,
+      const std::unordered_map<std::string, std::string>& headers,
+      const ErrorHandler& error_handler, auth::AuthSession& session);
 
   /// \brief Sends a POST request with form data.
-  Result<HttpResponse> PostForm(
+  virtual Result<HttpResponse> PostForm(
       const std::string& path,
       const std::unordered_map<std::string, std::string>& form_data,
       const std::unordered_map<std::string, std::string>& headers,
       const ErrorHandler& error_handler, auth::AuthSession& session);
 
   /// \brief Sends a HEAD request.
-  Result<HttpResponse> Head(const std::string& path,
-                            const std::unordered_map<std::string, std::string>& headers,
-                            const ErrorHandler& error_handler,
-                            auth::AuthSession& session);
+  virtual Result<HttpResponse> Head(
+      const std::string& path,
+      const std::unordered_map<std::string, std::string>& headers,
+      const ErrorHandler& error_handler, auth::AuthSession& session);
 
   /// \brief Sends a DELETE request.
-  Result<HttpResponse> Delete(const std::string& path,
-                              const std::unordered_map<std::string, std::string>& params,
-                              const std::unordered_map<std::string, std::string>& headers,
-                              const ErrorHandler& error_handler,
-                              auth::AuthSession& session);
+  virtual Result<HttpResponse> Delete(
+      const std::string& path, const std::unordered_map<std::string, std::string>& params,
+      const std::unordered_map<std::string, std::string>& headers,
+      const ErrorHandler& error_handler, auth::AuthSession& session);
 
  private:
   std::unordered_map<std::string, std::string> default_headers_;

src/iceberg/catalog/rest/json_serde.cc

@@ -128,8 +128,7 @@ Result<StorageCredential> StorageCredentialFromJson(const nlohmann::json& json)
   return credential;
 }
 
-/// \brief Reads the optional `storage-credentials` array shared by the
-/// LoadTable and LoadCredentials responses.
+/// \brief Reads the optional `storage-credentials` array.
 Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
     const nlohmann::json& json) {
   std::vector<StorageCredential> credentials;
@@ -146,6 +145,21 @@ Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
   }
   return credentials;
 }
+
+/// \brief Writes `storage-credentials` to json if any are present.
+Status StorageCredentialsToJson(const std::vector<StorageCredential>& credentials,
+                                nlohmann::json& json) {
+  if (credentials.empty()) {
+    return {};
+  }
+  nlohmann::json creds_json = nlohmann::json::array();
+  for (const auto& cred : credentials) {
+    ICEBERG_ASSIGN_OR_RAISE(auto entry, StorageCredentialToJson(cred));
+    creds_json.push_back(std::move(entry));
+  }
+  json[kStorageCredentials] = std::move(creds_json);
+  return {};
+}
 }  // namespace
 
 Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
@@ -829,6 +843,8 @@ Result<nlohmann::json> ToJson(const PlanTableScanRequest& request) {
   if (request.snapshot_id.has_value()) {
     json[kSnapshotId] = request.snapshot_id.value();
   }
+  // FIXME: Preserve explicit empty projections as select: []; omitting select
+  // means the default projection. Java currently has the same issue.
   if (!request.select.empty()) {
     json[kSelect] = request.select;
   }
@@ -884,6 +900,7 @@ Result<PlanTableScanResponse> PlanTableScanResponseFromJson(
   }
   ICEBERG_RETURN_UNEXPECTED(
       ScanTaskFieldsFromJson(json, response, partition_specs_by_id, schema));
+  ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials, StorageCredentialsFromJson(json));
   ICEBERG_RETURN_UNEXPECTED(response.Validate());
   return response;
 }
@@ -895,6 +912,7 @@ Result<nlohmann::json> ToJson(
     const Schema& schema) {
   ICEBERG_ASSIGN_OR_RAISE(nlohmann::json json,
                           ScanTaskFieldsToJson(response, partition_specs_by_id, schema));
+  ICEBERG_RETURN_UNEXPECTED(StorageCredentialsToJson(response.storage_credentials, json));
   json[kPlanStatus] = ToString(response.plan_status);
   if (!response.plan_id.empty()) {
     json[kPlanId] = response.plan_id;
@@ -921,6 +939,7 @@ Result<FetchPlanningResultResponse> FetchPlanningResultResponseFromJson(
   }
   ICEBERG_RETURN_UNEXPECTED(
       ScanTaskFieldsFromJson(json, response, partition_specs_by_id, schema));
+  ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials, StorageCredentialsFromJson(json));
   ICEBERG_RETURN_UNEXPECTED(response.Validate());
   return response;
 }
@@ -932,6 +951,7 @@ Result<nlohmann::json> ToJson(
     const Schema& schema) {
   ICEBERG_ASSIGN_OR_RAISE(nlohmann::json json,
                           ScanTaskFieldsToJson(response, partition_specs_by_id, schema));
+  ICEBERG_RETURN_UNEXPECTED(StorageCredentialsToJson(response.storage_credentials, json));
   json[kPlanStatus] = ToString(response.plan_status);
   if (response.error.has_value()) {
     json[kError] = ToJson(*response.error)[kError];