Address review comments and add tests for scan plan endpoints

- Cancel server-side plan when ResolveScanTasks fails partway through
- Propagate use_snapshot_schema from scan context to PlanTableScanRequest:
  true for UseSnapshot/AsOfTime/tag refs and incremental scans, false for
  branch refs and default scans
- Gate RestTable creation on effective scan-planning-mode config (table
  config overrides client config, default is client); error if server mode
  is requested but endpoint is not advertised
- Add ScanPlanningMode enum and ScanPlanningModeFrom() parser to
  RestCatalogProperties
- Make HttpClient methods virtual and add HttpResponse::MakeForTesting()
  to support unit test mocking
- Add tests: use_snapshot_schema in table_scan_test, ScanPlanningModeFrom
  parsing in catalog_properties_test, and RestTableScan HTTP flow tests
  in rest_table_scan_test

Author: Sandeep Gottimukkala

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> {
@@ -55,6 +58,8 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
   inline static Entry<std::string> kNamespaceSeparator{"namespace-separator", "%1F"};
   /// \brief The snapshot loading mode (ALL or REFS).
   inline static Entry<std::string> kSnapshotLoadingMode{"snapshot-loading-mode", "ALL"};
+  /// \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.";
 
@@ -77,6 +82,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,45 @@ 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/rest_catalog.cc

@@ -43,6 +43,8 @@
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
+#include "iceberg/logging/logger.h"
+#include "iceberg/logging/log_level.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
@@ -875,7 +877,32 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, table_session);
 
-  if (supported_endpoints_.contains(Endpoint::PlanTableScan())) {
+  // 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_,

src/iceberg/catalog/rest/rest_table_scan.cc

@@ -100,8 +100,10 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan(
   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()) {
@@ -129,8 +131,11 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan(
   plan_id = result.plan_id;
 
   switch (result.plan_status) {
-    case PlanStatus::kCompleted:
-      return ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs);
+    case PlanStatus::kCompleted: {
+      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:
@@ -165,8 +170,11 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchPlanningR
     ICEBERG_RETURN_UNEXPECTED(result.Validate());
 
     switch (result.plan_status) {
-      case PlanStatus::kCompleted:
-        return ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs);
+      case PlanStatus::kCompleted: {
+        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)

src/iceberg/table_scan.cc

@@ -301,6 +301,7 @@ TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::UseSnapshot(int64_t snap
                         context_.snapshot_id.value());
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(std::ignore, metadata_->SnapshotById(snapshot_id));
   context_.snapshot_id = snapshot_id;
+  context_.use_snapshot_schema = true;
   return *this;
 }
 
@@ -309,6 +310,7 @@ TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::UseRef(const std::string
   if (ref == SnapshotRef::kMainBranch) {
     snapshot_schema_ = nullptr;
     context_.snapshot_id.reset();
+    context_.use_snapshot_schema = false;
     return *this;
   }
 
@@ -321,6 +323,7 @@ TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::UseRef(const std::string
   const int64_t snapshot_id = iter->second->snapshot_id;
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(std::ignore, metadata_->SnapshotById(snapshot_id));
   context_.snapshot_id = snapshot_id;
+  context_.use_snapshot_schema = (iter->second->type() == SnapshotRefType::kTag);
 
   return *this;
 }

src/iceberg/table_scan.h

@@ -229,6 +229,7 @@ struct TableScanContext {
   std::optional<int64_t> to_snapshot_id;
   std::string branch{};
   std::optional<int64_t> min_rows_requested;
+  bool use_snapshot_schema{false};
   OptionalExecutor plan_executor;
 
   // Validate the context parameters to see if they have conflicts.

src/iceberg/test/CMakeLists.txt

@@ -297,10 +297,12 @@ if(ICEBERG_BUILD_REST)
   add_rest_iceberg_test(rest_catalog_test
                         SOURCES
                         auth_manager_test.cc
+                        catalog_properties_test.cc
                         error_handlers_test.cc
                         endpoint_test.cc
                         rest_file_io_test.cc
                         rest_json_serde_test.cc
+                        rest_table_scan_test.cc
                         rest_util_test.cc)
 
   if(ICEBERG_SIGV4)

src/iceberg/test/catalog_properties_test.cc

@@ -0,0 +1,81 @@
+/*
+ * 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/catalog_properties.h"
+
+#include <gtest/gtest.h>
+
+#include "iceberg/test/matchers.h"
+
+namespace iceberg::rest {
+
+TEST(ScanPlanningModeTest, MissingKeyReturnsNullopt) {
+  std::unordered_map<std::string, std::string> config;
+  auto result = RestCatalogProperties::ScanPlanningModeFrom(config);
+  ASSERT_THAT(result, IsOk());
+  EXPECT_FALSE(result->has_value());
+}
+
+TEST(ScanPlanningModeTest, ClientLowercaseReturnsKClient) {
+  auto result = RestCatalogProperties::ScanPlanningModeFrom(
+      {{"scan-planning-mode", "client"}});
+  ASSERT_THAT(result, IsOk());
+  ASSERT_TRUE(result->has_value());
+  EXPECT_EQ(**result, ScanPlanningMode::kClient);
+}
+
+TEST(ScanPlanningModeTest, ServerLowercaseReturnsKServer) {
+  auto result = RestCatalogProperties::ScanPlanningModeFrom(
+      {{"scan-planning-mode", "server"}});
+  ASSERT_THAT(result, IsOk());
+  ASSERT_TRUE(result->has_value());
+  EXPECT_EQ(**result, ScanPlanningMode::kServer);
+}
+
+TEST(ScanPlanningModeTest, ClientUppercaseReturnsKClient) {
+  auto result = RestCatalogProperties::ScanPlanningModeFrom(
+      {{"scan-planning-mode", "CLIENT"}});
+  ASSERT_THAT(result, IsOk());
+  ASSERT_TRUE(result->has_value());
+  EXPECT_EQ(**result, ScanPlanningMode::kClient);
+}
+
+TEST(ScanPlanningModeTest, ServerUppercaseReturnsKServer) {
+  auto result = RestCatalogProperties::ScanPlanningModeFrom(
+      {{"scan-planning-mode", "SERVER"}});
+  ASSERT_THAT(result, IsOk());
+  ASSERT_TRUE(result->has_value());
+  EXPECT_EQ(**result, ScanPlanningMode::kServer);
+}
+
+TEST(ScanPlanningModeTest, InvalidValueReturnsError) {
+  auto result = RestCatalogProperties::ScanPlanningModeFrom(
+      {{"scan-planning-mode", "invalid"}});
+  EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+}
+
+TEST(ScanPlanningModeTest, OtherKeysAreIgnored) {
+  auto result = RestCatalogProperties::ScanPlanningModeFrom(
+      {{"other-key", "server"}, {"scan-planning-mode", "client"}});
+  ASSERT_THAT(result, IsOk());
+  ASSERT_TRUE(result->has_value());
+  EXPECT_EQ(**result, ScanPlanningMode::kClient);
+}
+
+}  // namespace iceberg::rest