Address comments

Author: Sandeep Gottimukkala

src/iceberg/catalog/rest/error_handlers.cc

@@ -214,4 +214,24 @@ Status ScanPlanErrorHandler::Accept(const ErrorResponse& error) const {
   return DefaultErrorHandler::Accept(error);
 }
 
+const std::shared_ptr<PlanTaskErrorHandler>& PlanTaskErrorHandler::Instance() {
+  static const std::shared_ptr<PlanTaskErrorHandler> instance{new PlanTaskErrorHandler()};
+  return instance;
+}
+
+Status PlanTaskErrorHandler::Accept(const ErrorResponse& error) const {
+  switch (error.code) {
+    case 404:
+      if (error.type == kNoSuchNamespaceException) {
+        return NoSuchNamespace(error.message);
+      }
+      if (error.type == kNoSuchTableException) {
+        return NoSuchTable(error.message);
+      }
+      return NoSuchPlanTask(error.message);
+  }
+
+  return DefaultErrorHandler::Accept(error);
+}
+
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/error_handlers.h

@@ -138,4 +138,15 @@ class ICEBERG_REST_EXPORT ScanPlanErrorHandler final : public DefaultErrorHandle
   constexpr ScanPlanErrorHandler() = default;
 };
 
+/// \brief Fetch scan tasks operation error handler.
+class ICEBERG_REST_EXPORT PlanTaskErrorHandler final : public DefaultErrorHandler {
+ public:
+  static const std::shared_ptr<PlanTaskErrorHandler>& Instance();
+
+  Status Accept(const ErrorResponse& error) const override;
+
+ private:
+  constexpr PlanTaskErrorHandler() = default;
+};
+
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/json_serde.cc

@@ -93,9 +93,40 @@ constexpr std::string_view kUseSnapshotSchema = "use-snapshot-schema";
 constexpr std::string_view kStartSnapshotId = "start-snapshot-id";
 constexpr std::string_view kEndSnapshotId = "end-snapshot-id";
 constexpr std::string_view kStatsFields = "stats-fields";
-constexpr std::string_view kMinRowsRequired = "min-rows-required";
+constexpr std::string_view kMinRowsRequested = "min-rows-requested";
 constexpr std::string_view kPlanTask = "plan-task";
 
+Status BaseScanTaskResponseFromJson(
+    const nlohmann::json& json, BaseScanTaskResponse* response,
+    const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
+        partition_specs_by_id,
+    const Schema& schema) {
+  // 1. plan_tasks
+  ICEBERG_ASSIGN_OR_RAISE(
+      response->plan_tasks,
+      GetJsonValueOrDefault<std::vector<std::string>>(json, kPlanTasks));
+
+  // 2. delete_files
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto delete_files_json,
+      GetJsonValueOrDefault<nlohmann::json>(json, kDeleteFiles, nlohmann::json::array()));
+  for (const auto& entry_json : delete_files_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto delete_file,
+                            DataFileFromJson(entry_json, partition_specs_by_id, schema));
+    response->delete_files.push_back(std::move(delete_file));
+  }
+
+  // 3. file_scan_tasks
+  ICEBERG_ASSIGN_OR_RAISE(auto file_scan_tasks_json,
+                          GetJsonValueOrDefault<nlohmann::json>(json, kFileScanTasks,
+                                                                nlohmann::json::array()));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response->file_scan_tasks,
+      FileScanTasksFromJson(file_scan_tasks_json, response->delete_files,
+                            partition_specs_by_id, schema));
+  return {};
+}
+
 }  // namespace
 
 nlohmann::json ToJson(const CatalogConfig& config) {
@@ -547,7 +578,7 @@ Result<nlohmann::json> ToJson(const PlanTableScanRequest& request) {
     json[kStatsFields] = request.statsFields;
   }
   if (request.min_rows_required.has_value()) {
-    json[kMinRowsRequired] = request.min_rows_required.value();
+    json[kMinRowsRequested] = request.min_rows_required.value();
   }
   return json;
 }
@@ -558,37 +589,6 @@ nlohmann::json ToJson(const FetchScanTasksRequest& request) {
   return json;
 }
 
-Status BaseScanTaskResponseFromJson(
-    const nlohmann::json& json, BaseScanTaskResponse* response,
-    const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
-        partition_specs_by_id,
-    const Schema& schema) {
-  // 1. plan_tasks
-  ICEBERG_ASSIGN_OR_RAISE(
-      response->plan_tasks,
-      GetJsonValueOrDefault<std::vector<std::string>>(json, kPlanTasks));
-
-  // 2. delete_files
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto delete_files_json,
-      GetJsonValueOrDefault<nlohmann::json>(json, kDeleteFiles, nlohmann::json::array()));
-  for (const auto& entry_json : delete_files_json) {
-    ICEBERG_ASSIGN_OR_RAISE(auto delete_file,
-                            DataFileFromJson(entry_json, partition_specs_by_id, schema));
-    response->delete_files.push_back(std::move(delete_file));
-  }
-
-  // 3. file_scan_tasks
-  ICEBERG_ASSIGN_OR_RAISE(auto file_scan_tasks_json,
-                          GetJsonValueOrDefault<nlohmann::json>(json, kFileScanTasks,
-                                                                nlohmann::json::array()));
-  ICEBERG_ASSIGN_OR_RAISE(
-      response->file_scan_tasks,
-      FileScanTasksFromJson(file_scan_tasks_json, response->delete_files,
-                            partition_specs_by_id, schema));
-  return {};
-}
-
 Result<PlanTableScanResponse> PlanTableScanResponseFromJson(
     const nlohmann::json& json,
     const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&

src/iceberg/catalog/rest/resource_paths.cc

@@ -113,8 +113,9 @@ Result<std::string> ResourcePaths::ScanPlan(const TableIdentifier& ident,
                                             const std::string& plan_id) const {
   ICEBERG_ASSIGN_OR_RAISE(std::string encoded_namespace, EncodeNamespace(ident.ns));
   ICEBERG_ASSIGN_OR_RAISE(std::string encoded_table_name, EncodeString(ident.name));
+  ICEBERG_ASSIGN_OR_RAISE(std::string encoded_plan_id, EncodeString(plan_id));
   return std::format("{}/v1/{}namespaces/{}/tables/{}/plan/{}", base_uri_, prefix_,
-                     encoded_namespace, encoded_table_name, plan_id);
+                     encoded_namespace, encoded_table_name, encoded_plan_id);
 }
 
 Result<std::string> ResourcePaths::ScanTask(const TableIdentifier& ident) const {

src/iceberg/catalog/rest/rest_catalog.cc

@@ -572,7 +572,7 @@ Result<FetchScanTasksResponse> RestCatalog::FetchScanTasks(const Table& table,
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
-      client_->Post(path, json_request, /*headers=*/{}, *ScanPlanErrorHandler::Instance(),
+      client_->Post(path, json_request, /*headers=*/{}, *PlanTaskErrorHandler::Instance(),
                     *catalog_session_));
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));