Fix REST scan JSON serde parity

Author: wgtmac

src/iceberg/catalog/rest/json_serde.cc

@@ -27,11 +27,8 @@
 
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/catalog/rest/types.h"
-#include "iceberg/manifest/manifest_entry.h"
-#include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
-#include "iceberg/schema.h"
-#include "iceberg/table_scan.h"
+#include "iceberg/type_fwd.h"
 
 /// \file iceberg/catalog/rest/json_serde_internal.h
 /// JSON serialization and deserialization for Iceberg REST Catalog API types.
@@ -75,24 +72,52 @@ ICEBERG_REST_EXPORT Result<PlanTableScanResponse> PlanTableScanResponseFromJson(
     const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
         partition_specs_by_id,
     const Schema& schema);
+ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(
+    const PlanTableScanResponse& response,
+    const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
+        partition_specs_by_id,
+    const Schema& schema);
 
 ICEBERG_REST_EXPORT Result<FetchPlanningResultResponse>
 FetchPlanningResultResponseFromJson(
     const nlohmann::json& json,
     const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
         partition_specs_by_id,
     const Schema& schema);
+ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(
+    const FetchPlanningResultResponse& response,
+    const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
+        partition_specs_by_id,
+    const Schema& schema);
 
 ICEBERG_REST_EXPORT Result<FetchScanTasksResponse> FetchScanTasksResponseFromJson(
     const nlohmann::json& json,
     const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
         partition_specs_by_id,
     const Schema& schema);
+ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(
+    const FetchScanTasksResponse& response,
+    const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
+        partition_specs_by_id,
+    const Schema& schema);
 
+ICEBERG_REST_EXPORT Result<PlanTableScanRequest> PlanTableScanRequestFromJson(
+    const nlohmann::json& json);
+template <>
+ICEBERG_REST_EXPORT Result<PlanTableScanRequest> FromJson(const nlohmann::json& json);
 ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const PlanTableScanRequest& request);
+
+ICEBERG_REST_EXPORT Result<FetchScanTasksRequest> FetchScanTasksRequestFromJson(
+    const nlohmann::json& json);
+template <>
+ICEBERG_REST_EXPORT Result<FetchScanTasksRequest> FromJson(const nlohmann::json& json);
 ICEBERG_REST_EXPORT nlohmann::json ToJson(const FetchScanTasksRequest& request);
 
-ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const DataFile& df);
+ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(
+    const DataFile& df,
+    const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
+        partition_specs_by_id,
+    const Schema& schema);
 
 ICEBERG_REST_EXPORT Result<DataFile> DataFileFromJson(
     const nlohmann::json& json,
@@ -107,9 +132,4 @@ FileScanTasksFromJson(const nlohmann::json& json,
                           partition_spec_by_id,
                       const Schema& schema);
 
-ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const PlanTableScanResponse& response);
-ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(
-    const FetchPlanningResultResponse& response);
-ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const FetchScanTasksResponse& response);
-
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/json_serde_internal.h

@@ -20,16 +20,42 @@
 #include "iceberg/catalog/rest/types.h"
 
 #include <algorithm>
+#include <optional>
 
+#include "iceberg/expression/expression.h"
+#include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/schema.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_requirement.h"
+#include "iceberg/table_scan.h"
 #include "iceberg/table_update.h"
 
 namespace iceberg::rest {
 
+std::string_view ToString(PlanStatus status) {
+  switch (status) {
+    case PlanStatus::kSubmitted:
+      return "submitted";
+    case PlanStatus::kCompleted:
+      return "completed";
+    case PlanStatus::kCancelled:
+      return "cancelled";
+    case PlanStatus::kFailed:
+      return "failed";
+  }
+  return "unknown";
+}
+
+Result<PlanStatus> PlanStatusFromString(std::string_view status_str) {
+  if (status_str == "submitted") return PlanStatus::kSubmitted;
+  if (status_str == "completed") return PlanStatus::kCompleted;
+  if (status_str == "cancelled") return PlanStatus::kCancelled;
+  if (status_str == "failed") return PlanStatus::kFailed;
+  return JsonParseError("Unknown plan status: {}", status_str);
+}
+
 bool CreateTableRequest::operator==(const CreateTableRequest& other) const {
   if (name != other.name || location != other.location ||
       stage_create != other.stage_create || properties != other.properties) {
@@ -118,9 +144,22 @@ bool CommitTableResponse::operator==(const CommitTableResponse& other) const {
   return true;
 }
 
+namespace {
+
+bool ExpressionPtrEqual(const std::shared_ptr<Expression>& lhs,
+                        const std::shared_ptr<Expression>& rhs) {
+  if (lhs == rhs) {
+    return true;
+  }
+  return lhs && rhs && lhs->Equals(*rhs);
+}
+
+}  // namespace
+
 bool PlanTableScanRequest::operator==(const PlanTableScanRequest& other) const {
   return snapshot_id == other.snapshot_id && select == other.select &&
-         filter == other.filter && case_sensitive == other.case_sensitive &&
+         ExpressionPtrEqual(filter, other.filter) &&
+         case_sensitive == other.case_sensitive &&
          use_snapshot_schema == other.use_snapshot_schema &&
          start_snapshot_id == other.start_snapshot_id &&
          end_snapshot_id == other.end_snapshot_id && stats_fields == other.stats_fields &&
@@ -129,86 +168,81 @@ bool PlanTableScanRequest::operator==(const PlanTableScanRequest& other) const {
 
 namespace {
 
-bool ScanTaskFieldsEqual(
-    const std::vector<std::string>& plan_tasks_a,
-    const std::vector<std::string>& plan_tasks_b,
-    const std::vector<std::shared_ptr<DataFile>>& delete_files_a,
-    const std::vector<std::shared_ptr<DataFile>>& delete_files_b,
-    const std::vector<std::shared_ptr<FileScanTask>>& file_scan_tasks_a,
-    const std::vector<std::shared_ptr<FileScanTask>>& file_scan_tasks_b) {
-  if (plan_tasks_a != plan_tasks_b) {
-    return false;
+template <typename T>
+bool SharedPtrEqual(const std::shared_ptr<T>& lhs, const std::shared_ptr<T>& rhs) {
+  if (lhs == rhs) {
+    return true;
   }
-  if (delete_files_a.size() != delete_files_b.size()) {
-    return false;
-  }
-  for (size_t i = 0; i < delete_files_a.size(); ++i) {
-    if (!delete_files_a[i] != !delete_files_b[i]) {
-      return false;
-    }
-    if (delete_files_a[i] && *delete_files_a[i] != *delete_files_b[i]) {
-      return false;
-    }
+  return lhs && rhs && *lhs == *rhs;
+}
+
+template <typename T>
+bool SharedPtrVectorEqual(const std::vector<std::shared_ptr<T>>& lhs,
+                          const std::vector<std::shared_ptr<T>>& rhs) {
+  return std::ranges::equal(lhs, rhs, SharedPtrEqual<T>);
+}
+
+bool FileScanTaskEqual(const std::shared_ptr<FileScanTask>& lhs,
+                       const std::shared_ptr<FileScanTask>& rhs) {
+  if (lhs == rhs) {
+    return true;
   }
-  if (file_scan_tasks_a.size() != file_scan_tasks_b.size()) {
+  if (!lhs || !rhs) {
     return false;
   }
-  for (size_t i = 0; i < file_scan_tasks_a.size(); ++i) {
-    const auto& a = file_scan_tasks_a[i];
-    const auto& b = file_scan_tasks_b[i];
-    if (!a != !b) {
-      return false;
-    }
-    if (!a) continue;
-    if (!a->data_file() != !b->data_file()) {
-      return false;
-    }
-    if (a->data_file() && *a->data_file() != *b->data_file()) {
-      return false;
-    }
-    if (a->delete_files().size() != b->delete_files().size()) {
-      return false;
-    }
-    for (size_t j = 0; j < a->delete_files().size(); ++j) {
-      if (!a->delete_files()[j] != !b->delete_files()[j]) {
-        return false;
-      }
-      if (a->delete_files()[j] && *a->delete_files()[j] != *b->delete_files()[j]) {
-        return false;
-      }
-    }
-    if (a->residual_filter() != b->residual_filter()) {
-      return false;
-    }
+  return SharedPtrEqual(lhs->data_file(), rhs->data_file()) &&
+         SharedPtrVectorEqual(lhs->delete_files(), rhs->delete_files()) &&
+         ExpressionPtrEqual(lhs->residual_filter(), rhs->residual_filter());
+}
+
+template <typename T>
+bool OptionalSharedPtrVectorEqual(
+    const std::optional<std::vector<std::shared_ptr<T>>>& lhs,
+    const std::optional<std::vector<std::shared_ptr<T>>>& rhs,
+    bool (*eq)(const std::shared_ptr<T>&, const std::shared_ptr<T>&)) {
+  if (lhs.has_value() != rhs.has_value()) {
+    return false;
   }
-  return true;
+  return !lhs.has_value() || std::ranges::equal(*lhs, *rhs, eq);
+}
+
+template <typename Response>
+bool ScanTaskFieldsEqual(const Response& lhs, const Response& rhs) {
+  return lhs.plan_tasks == rhs.plan_tasks &&
+         SharedPtrVectorEqual(lhs.delete_files, rhs.delete_files) &&
+         OptionalSharedPtrVectorEqual(lhs.file_scan_tasks, rhs.file_scan_tasks,
+                                      FileScanTaskEqual);
+}
+
+template <typename Response>
+bool HasTaskFields(const Response& response) {
+  return response.plan_tasks.has_value() || response.file_scan_tasks.has_value();
+}
+
+template <typename Response>
+bool HasNonEmptyFileScanTasks(const Response& response) {
+  return response.file_scan_tasks.has_value() && !response.file_scan_tasks->empty();
 }
 
 }  // namespace
 
 bool PlanTableScanResponse::operator==(const PlanTableScanResponse& other) const {
-  return ScanTaskFieldsEqual(plan_tasks, other.plan_tasks, delete_files,
-                             other.delete_files, file_scan_tasks,
-                             other.file_scan_tasks) &&
-         plan_status == other.plan_status && plan_id == other.plan_id &&
-         error == other.error;
+  return ScanTaskFieldsEqual(*this, other) && plan_status == other.plan_status &&
+         plan_id == other.plan_id && error == other.error;
 }
 
 bool FetchPlanningResultResponse::operator==(
     const FetchPlanningResultResponse& other) const {
-  return ScanTaskFieldsEqual(plan_tasks, other.plan_tasks, delete_files,
-                             other.delete_files, file_scan_tasks,
-                             other.file_scan_tasks) &&
-         plan_status == other.plan_status && error == other.error;
+  return ScanTaskFieldsEqual(*this, other) && plan_status == other.plan_status &&
+         error == other.error;
 }
 
 bool FetchScanTasksRequest::operator==(const FetchScanTasksRequest& other) const {
   return planTask == other.planTask;
 }
 
 bool FetchScanTasksResponse::operator==(const FetchScanTasksResponse& other) const {
-  return ScanTaskFieldsEqual(plan_tasks, other.plan_tasks, delete_files,
-                             other.delete_files, file_scan_tasks, other.file_scan_tasks);
+  return ScanTaskFieldsEqual(*this, other);
 }
 
 Status OAuthTokenResponse::Validate() const {
@@ -257,8 +291,7 @@ Status PlanTableScanResponse::Validate() const {
     return ValidationFailed(
         "Invalid response: 'cancelled' is not a valid status for planTableScan");
   }
-  if (plan_status != PlanStatus::kCompleted &&
-      (!plan_tasks.empty() || !file_scan_tasks.empty())) {
+  if (plan_status != PlanStatus::kCompleted && HasTaskFields(*this)) {
     return ValidationFailed(
         "Invalid response: tasks can only be defined when status is 'completed'");
   }
@@ -268,7 +301,7 @@ Status PlanTableScanResponse::Validate() const {
         "Invalid response: plan id can only be defined when status is 'submitted' or "
         "'completed'");
   }
-  if (file_scan_tasks.empty() && !delete_files.empty()) {
+  if (!HasNonEmptyFileScanTasks(*this) && !delete_files.empty()) {
     return ValidationFailed(
         "Invalid response: deleteFiles should only be returned with fileScanTasks that "
         "reference them");
@@ -285,12 +318,11 @@ Status PlanTableScanResponse::Validate() const {
 }
 
 Status FetchPlanningResultResponse::Validate() const {
-  if (plan_status != PlanStatus::kCompleted &&
-      (!plan_tasks.empty() || !file_scan_tasks.empty())) {
+  if (plan_status != PlanStatus::kCompleted && HasTaskFields(*this)) {
     return ValidationFailed(
         "Invalid response: tasks can only be returned in a 'completed' status");
   }
-  if (file_scan_tasks.empty() && !delete_files.empty()) {
+  if (!HasNonEmptyFileScanTasks(*this) && !delete_files.empty()) {
     return ValidationFailed(
         "Invalid response: deleteFiles should only be returned with fileScanTasks that "
         "reference them");
@@ -314,12 +346,12 @@ Status FetchScanTasksRequest::Validate() const {
 }
 
 Status FetchScanTasksResponse::Validate() const {
-  if (file_scan_tasks.empty() && !delete_files.empty()) {
+  if (!HasNonEmptyFileScanTasks(*this) && !delete_files.empty()) {
     return ValidationFailed(
         "Invalid response: deleteFiles should only be returned with fileScanTasks that "
         "reference them");
   }
-  if (plan_tasks.empty() && file_scan_tasks.empty()) {
+  if (!HasTaskFields(*this)) {
     return ValidationFailed(
         "Invalid response: planTasks and fileScanTask cannot both be null");
   }

src/iceberg/catalog/rest/types.cc

@@ -29,12 +29,8 @@
 
 #include "iceberg/catalog/rest/endpoint.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
-#include "iceberg/expression/expression.h"
-#include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/result.h"
-#include "iceberg/schema.h"
 #include "iceberg/table_identifier.h"
-#include "iceberg/table_scan.h"
 #include "iceberg/type_fwd.h"
 #include "iceberg/util/macros.h"
 
@@ -43,6 +39,17 @@
 
 namespace iceberg::rest {
 
+/// \brief Status of a REST server-side scan planning operation.
+enum class PlanStatus {
+  kSubmitted,
+  kCompleted,
+  kCancelled,
+  kFailed,
+};
+
+ICEBERG_REST_EXPORT std::string_view ToString(PlanStatus status);
+ICEBERG_REST_EXPORT Result<PlanStatus> PlanStatusFromString(std::string_view status_str);
+
 /// \brief Server-provided configuration for the catalog.
 struct ICEBERG_REST_EXPORT CatalogConfig {
   std::unordered_map<std::string, std::string> defaults;   // required
@@ -319,8 +326,8 @@ struct ICEBERG_REST_EXPORT PlanTableScanRequest {
 /// \brief Response from initiating a scan planning operation, including plan status and
 /// initial scan tasks.
 struct ICEBERG_REST_EXPORT PlanTableScanResponse {
-  std::vector<std::string> plan_tasks;
-  std::vector<std::shared_ptr<FileScanTask>> file_scan_tasks;
+  std::optional<std::vector<std::string>> plan_tasks;
+  std::optional<std::vector<std::shared_ptr<FileScanTask>>> file_scan_tasks;
   std::vector<std::shared_ptr<DataFile>> delete_files;
   PlanStatus plan_status = PlanStatus::kCompleted;
   std::string plan_id;
@@ -335,8 +342,8 @@ struct ICEBERG_REST_EXPORT PlanTableScanResponse {
 /// \brief Response from polling an asynchronous scan plan, including current status and
 /// available scan tasks.
 struct ICEBERG_REST_EXPORT FetchPlanningResultResponse {
-  std::vector<std::string> plan_tasks;
-  std::vector<std::shared_ptr<FileScanTask>> file_scan_tasks;
+  std::optional<std::vector<std::string>> plan_tasks;
+  std::optional<std::vector<std::shared_ptr<FileScanTask>>> file_scan_tasks;
   std::vector<std::shared_ptr<DataFile>> delete_files;
   PlanStatus plan_status = PlanStatus::kCompleted;
   std::optional<ErrorResponse> error;
@@ -358,8 +365,8 @@ struct ICEBERG_REST_EXPORT FetchScanTasksRequest {
 
 /// \brief Response containing the file scan tasks for a given plan task token.
 struct ICEBERG_REST_EXPORT FetchScanTasksResponse {
-  std::vector<std::string> plan_tasks;
-  std::vector<std::shared_ptr<FileScanTask>> file_scan_tasks;
+  std::optional<std::vector<std::string>> plan_tasks;
+  std::optional<std::vector<std::shared_ptr<FileScanTask>>> file_scan_tasks;
   std::vector<std::shared_ptr<DataFile>> delete_files;
 
   Status Validate() const;