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

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,7 @@
 #include "iceberg/catalog/rest/catalog_properties.h"
 
 #include <algorithm>
+#include <optional>
 #include <string>
 #include <string_view>
 
@@ -61,4 +62,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

@@ -34,6 +34,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 +61,8 @@ 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 The prefix for HTTP headers.
   inline static constexpr std::string_view kHeaderPrefix = "header.";
 
@@ -80,6 +85,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

@@ -62,6 +62,15 @@ std::unordered_map<std::string, std::string> HttpResponse::headers() const {
   return impl_->headers();
 }
 
+HttpResponse HttpResponse::MakeForTesting(int32_t status_code, std::string body) {
+  cpr::Response cpr_response;
+  cpr_response.status_code = status_code;
+  cpr_response.text = std::move(body);
+  HttpResponse response;
+  response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(cpr_response));
+  return response;
+}
+
 namespace {
 
 /// \brief Default error type for unparseable REST responses.

src/iceberg/catalog/rest/http_client.h

@@ -61,6 +61,9 @@ class ICEBERG_REST_EXPORT HttpResponse {
   /// \brief Get the headers of the response as a map.
   std::unordered_map<std::string, std::string> headers() const;
 
+  /// \brief Create a response for use in unit tests.
+  static HttpResponse MakeForTesting(int32_t status_code, std::string body);
+
  private:
   friend class HttpClient;
   class Impl;
@@ -71,44 +74,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

@@ -531,6 +531,15 @@ Result<nlohmann::json> ScanTaskFieldsToJson(
     json[kFileScanTasks] = std::move(tasks_json);
   }
 
+  if (!response.storage_credentials.empty()) {
+    nlohmann::json creds_json = nlohmann::json::array();
+    for (const auto& cred : response.storage_credentials) {
+      ICEBERG_ASSIGN_OR_RAISE(auto entry, StorageCredentialToJson(cred));
+      creds_json.push_back(std::move(entry));
+    }
+    json[kStorageCredentials] = std::move(creds_json);
+  }
+
   return json;
 }
 
@@ -571,6 +580,21 @@ Status ScanTaskFieldsFromJson(
         FileScanTasksFromJson(file_scan_tasks_json, response.delete_files,
                               partition_specs_by_id, schema));
   }
+
+  // 4. storage_credentials
+  if (json.contains(kStorageCredentials)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto creds_json,
+                            GetJsonValue<nlohmann::json>(json, kStorageCredentials));
+    if (!creds_json.is_array()) {
+      return JsonParseError("Cannot parse storage credentials from non-array: {}",
+                            SafeDumpJson(creds_json));
+    }
+    for (const auto& entry : creds_json) {
+      ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(entry));
+      response.storage_credentials.push_back(std::move(cred));
+    }
+  }
+
   return {};
 }
 

src/iceberg/catalog/rest/rest_catalog.cc

@@ -28,6 +28,7 @@
 
 #include <nlohmann/json.hpp>
 
+#include "iceberg/catalog/rest/auth/auth_manager_internal.h"
 #include "iceberg/catalog/rest/auth/auth_managers.h"
 #include "iceberg/catalog/rest/auth/auth_session.h"
 #include "iceberg/catalog/rest/catalog_properties.h"
@@ -39,9 +40,12 @@
 #include "iceberg/catalog/rest/resource_paths.h"
 #include "iceberg/catalog/rest/rest_file_io.h"
 #include "iceberg/catalog/rest/rest_metrics_reporter_internal.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"
+#include "iceberg/logging/log_level.h"
+#include "iceberg/logging/logger.h"
 #include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
@@ -452,9 +456,25 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
       snapshot_mode, std::move(default_context), std::move(reporter), metrics_executor));
 }
 
+Result<std::shared_ptr<RestCatalog>> RestCatalog::MakeForTesting(
+    RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
+    std::shared_ptr<HttpClient> client, std::shared_ptr<ResourcePaths> paths,
+    std::unordered_set<Endpoint> endpoints) {
+  std::string catalog_name = config.Get(RestCatalogProperties::kName);
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_manager,
+                          auth::MakeNoopAuthManager(catalog_name, config.configs()));
+  auto session = auth::AuthSession::MakeDefault({});
+  ICEBERG_ASSIGN_OR_RAISE(auto snapshot_mode, config.SnapshotLoadingMode());
+  return std::shared_ptr<RestCatalog>(
+      new RestCatalog(std::move(config), std::move(file_io), std::move(client),
+                      std::move(paths), std::move(endpoints), std::move(auth_manager),
+                      std::move(session), snapshot_mode, SessionContext::Empty(),
+                      /*reporter=*/nullptr, /*metrics_executor=*/nullptr));
+}
+
 RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
                          std::shared_ptr<HttpClient> client,
-                         std::unique_ptr<ResourcePaths> paths,
+                         std::shared_ptr<ResourcePaths> paths,
                          std::unordered_set<Endpoint> endpoints,
                          std::unique_ptr<auth::AuthManager> auth_manager,
                          std::shared_ptr<auth::AuthSession> catalog_session,
@@ -899,6 +919,47 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, table_session, table_io);
 
+  // Determine effective scan planning mode: table config overrides client config.
+  ICEBERG_ASSIGN_OR_RAISE(auto client_mode,
+                          RestCatalogProperties::ScanPlanningModeFrom(config_.configs()));
+  ICEBERG_ASSIGN_OR_RAISE(auto server_mode,
+                          RestCatalogProperties::ScanPlanningModeFrom(table_config));
+
+  if (client_mode.has_value() && server_mode.has_value() &&
+      *client_mode != *server_mode) {
+    Log(LogLevel::kWarn,
+        "Scan planning mode mismatch for table {}: client config={}, server config={}. "
+        "Server config will take precedence.",
+        identifier.ToString(),
+        *client_mode == ScanPlanningMode::kClient ? "client" : "server",
+        *server_mode == ScanPlanningMode::kClient ? "client" : "server");
+  }
+
+  ScanPlanningMode effective_mode =
+      server_mode.value_or(client_mode.value_or(ScanPlanningMode::kClient));
+
+  if (effective_mode == ScanPlanningMode::kServer) {
+    if (!supported_endpoints_.contains(Endpoint::PlanTableScan())) {
+      return NotSupported(
+          "Server requires server-side scan planning for table {} but does not support "
+          "the PlanTableScan endpoint.",
+          identifier.ToString());
+    }
+    RestScanContext rest_ctx{
+        .client = client_,
+        .paths = paths_,
+        .session = table_session,
+        .supported_endpoints = supported_endpoints_,
+        .identifier = identifier,
+        .catalog_config = config_.configs(),
+        .table_config = table_config,
+    };
+    return RestTable::Make(identifier, std::move(result.metadata),
+                           std::move(result.metadata_location), std::move(table_io),
+                           std::move(table_catalog), RestTableName(name_, identifier),
+                           reporter, 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), RestTableName(name_, identifier),

src/iceberg/catalog/rest/rest_catalog.h

@@ -58,6 +58,13 @@ class ICEBERG_REST_EXPORT RestCatalog final
   static Result<std::shared_ptr<RestCatalog>> Make(const RestCatalogProperties& config,
                                                    Executor* metrics_executor = nullptr);
 
+  /// \brief Test-only factory that constructs a RestCatalog with pre-built dependencies,
+  /// bypassing the FetchServerConfig HTTP exchange performed by Make().
+  static Result<std::shared_ptr<RestCatalog>> MakeForTesting(
+      RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
+      std::shared_ptr<HttpClient> client, std::shared_ptr<ResourcePaths> paths,
+      std::unordered_set<Endpoint> endpoints);
+
   std::string_view name() const override;
 
   Result<std::shared_ptr<Catalog>> AsCatalog() override;
@@ -69,7 +76,7 @@ class ICEBERG_REST_EXPORT RestCatalog final
   class TableScopedCatalog;
 
   RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
-              std::shared_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,
@@ -193,7 +200,7 @@ class ICEBERG_REST_EXPORT RestCatalog final
   RestCatalogProperties config_;
   std::shared_ptr<FileIO> file_io_;
   std::shared_ptr<HttpClient> client_;
-  std::unique_ptr<ResourcePaths> paths_;
+  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,68 @@
+/*
+ * 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, std::string full_name,
+                     std::shared_ptr<MetricsReporter> reporter,
+                     RestScanContext rest_context)
+    : Table(std::move(identifier), std::move(metadata), std::move(metadata_location),
+            std::move(io), std::move(catalog), std::move(full_name), std::move(reporter)),
+      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, std::string full_name,
+    std::shared_ptr<MetricsReporter> reporter, 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(full_name), std::move(reporter), std::move(rest_context)));
+}
+
+Result<std::unique_ptr<DataTableScanBuilder>> RestTable::NewScan() const {
+  return std::make_unique<RestTableScanBuilder>(metadata_, io_, full_name_, reporter_,
+                                                rest_context_);
+}
+
+Result<std::unique_ptr<IncrementalAppendScanBuilder>>
+RestTable::NewIncrementalAppendScan() const {
+  return std::make_unique<RestIncrementalAppendScanBuilder>(metadata_, io_, full_name_,
+                                                            reporter_, rest_context_);
+}
+
+}  // namespace iceberg::rest