feat(rest): use PlanStatus enum and shared_ptr for scan task types

Replace string-based plan_status fields in PlanTableScanResponse and
FetchPlanningResultResponse with a PlanStatus enum defined in table_scan.h.
Change file_scan_tasks and delete_files in BaseScanTaskResponse from value
vectors to vector<shared_ptr<...>> for shared ownership semantics.

Author: Sandeep Gottimukkala

src/iceberg/catalog/rest/json_serde.cc

@@ -113,7 +113,7 @@ Status BaseScanTaskResponseFromJson(
   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));
+    response->delete_files.push_back(std::make_shared<DataFile>(std::move(delete_file)));
   }
 
   // 3. file_scan_tasks
@@ -595,8 +595,9 @@ Result<PlanTableScanResponse> PlanTableScanResponseFromJson(
         partition_specs_by_id,
     const Schema& schema) {
   PlanTableScanResponse response;
-  ICEBERG_ASSIGN_OR_RAISE(response.plan_status,
+  ICEBERG_ASSIGN_OR_RAISE(auto plan_status_str,
                           GetJsonValue<std::string>(json, kPlanStatus));
+  ICEBERG_ASSIGN_OR_RAISE(response.plan_status, PlanStatusFromString(plan_status_str));
   ICEBERG_ASSIGN_OR_RAISE(response.plan_id,
                           GetJsonValueOrDefault<std::string>(json, kPlanId));
   ICEBERG_RETURN_UNEXPECTED(
@@ -611,8 +612,9 @@ Result<FetchPlanningResultResponse> FetchPlanningResultResponseFromJson(
         partition_specs_by_id,
     const Schema& schema) {
   FetchPlanningResultResponse response;
-  ICEBERG_ASSIGN_OR_RAISE(response.plan_status,
+  ICEBERG_ASSIGN_OR_RAISE(auto plan_status_str,
                           GetJsonValue<std::string>(json, kPlanStatus));
+  ICEBERG_ASSIGN_OR_RAISE(response.plan_status, PlanStatusFromString(plan_status_str));
   ICEBERG_RETURN_UNEXPECTED(
       BaseScanTaskResponseFromJson(json, &response, partition_specs_by_id, schema));
   ICEBERG_RETURN_UNEXPECTED(response.Validate());

src/iceberg/catalog/rest/types.cc

@@ -131,33 +131,45 @@ bool BaseScanTaskResponse::operator==(const BaseScanTaskResponse& other) const {
   if (plan_tasks != other.plan_tasks) {
     return false;
   }
-  if (delete_files != other.delete_files) {
+  if (delete_files.size() != other.delete_files.size()) {
     return false;
   }
+  for (size_t i = 0; i < delete_files.size(); ++i) {
+    if (!delete_files[i] != !other.delete_files[i]) {
+      return false;
+    }
+    if (delete_files[i] && *delete_files[i] != *other.delete_files[i]) {
+      return false;
+    }
+  }
   if (file_scan_tasks.size() != other.file_scan_tasks.size()) {
     return false;
   }
   for (size_t i = 0; i < file_scan_tasks.size(); ++i) {
     const auto& a = file_scan_tasks[i];
     const auto& b = other.file_scan_tasks[i];
-    if (!a.data_file() != !b.data_file()) {
+    if (!a != !b) {
       return false;
     }
-    if (a.data_file() && *a.data_file() != *b.data_file()) {
+    if (!a) continue;
+    if (!a->data_file() != !b->data_file()) {
       return false;
     }
-    if (a.delete_files().size() != b.delete_files().size()) {
+    if (a->data_file() && *a->data_file() != *b->data_file()) {
       return false;
     }
-    for (size_t j = 0; j < a.delete_files().size(); ++j) {
-      if (!a.delete_files()[j] != !b.delete_files()[j]) {
+    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]) {
+      if (a->delete_files()[j] && *a->delete_files()[j] != *b->delete_files()[j]) {
         return false;
       }
     }
-    if (a.residual_filter() != b.residual_filter()) {
+    if (a->residual_filter() != b->residual_filter()) {
       return false;
     }
   }
@@ -220,22 +232,20 @@ Status PlanTableScanRequest::Validate() const {
 }
 
 Status PlanTableScanResponse::Validate() const {
-  if (plan_status.empty()) {
-    return ValidationFailed("Invalid response: plan status must be defined");
-  }
-  if (plan_status == "submitted" && plan_id.empty()) {
+  if (plan_status == PlanStatus::kSubmitted && plan_id.empty()) {
     return ValidationFailed(
         "Invalid response: plan id should be defined when status is 'submitted'");
   }
-  if (plan_status == "cancelled") {
+  if (plan_status == PlanStatus::kCancelled) {
     return ValidationFailed(
         "Invalid response: 'cancelled' is not a valid status for planTableScan");
   }
-  if (plan_status != "completed" && (!plan_tasks.empty() || !file_scan_tasks.empty())) {
+  if (plan_status != PlanStatus::kCompleted && (!plan_tasks.empty() || !file_scan_tasks.empty())) {
     return ValidationFailed(
         "Invalid response: tasks can only be defined when status is 'completed'");
   }
-  if (!plan_id.empty() && plan_status != "submitted" && plan_status != "completed") {
+  if (!plan_id.empty() && plan_status != PlanStatus::kSubmitted &&
+      plan_status != PlanStatus::kCompleted) {
     return ValidationFailed(
         "Invalid response: plan id can only be defined when status is 'submitted' or "
         "'completed'");
@@ -249,10 +259,7 @@ Status PlanTableScanResponse::Validate() const {
 }
 
 Status FetchPlanningResultResponse::Validate() const {
-  if (plan_status.empty()) {
-    return ValidationFailed("Invalid status: null");
-  }
-  if (plan_status != "completed" && (!plan_tasks.empty() || !file_scan_tasks.empty())) {
+  if (plan_status != PlanStatus::kCompleted && (!plan_tasks.empty() || !file_scan_tasks.empty())) {
     return ValidationFailed(
         "Invalid response: tasks can only be returned in a 'completed' status");
   }

src/iceberg/catalog/rest/types.h

@@ -23,6 +23,7 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <string_view>
 #include <unordered_map>
 #include <vector>
 
@@ -319,8 +320,8 @@ struct ICEBERG_REST_EXPORT PlanTableScanRequest {
 /// endpoints.
 struct ICEBERG_REST_EXPORT BaseScanTaskResponse {
   std::vector<std::string> plan_tasks;
-  std::vector<FileScanTask> file_scan_tasks;
-  std::vector<DataFile> delete_files;
+  std::vector<std::shared_ptr<FileScanTask>> file_scan_tasks;
+  std::vector<std::shared_ptr<DataFile>> delete_files;
   // std::unordered_map<std::string, PartitionSpec> specsById;
 
   Status Validate() const { return {}; };
@@ -331,7 +332,7 @@ struct ICEBERG_REST_EXPORT BaseScanTaskResponse {
 /// \brief Response from initiating a scan planning operation, including plan status and
 /// initial scan tasks.
 struct ICEBERG_REST_EXPORT PlanTableScanResponse : BaseScanTaskResponse {
-  std::string plan_status;
+  PlanStatus plan_status = PlanStatus::kCompleted;
   std::string plan_id;
   // TODO(sandeepg): Add credentials.
 
@@ -343,7 +344,7 @@ struct ICEBERG_REST_EXPORT PlanTableScanResponse : BaseScanTaskResponse {
 /// \brief Response from polling an asynchronous scan plan, including current status and
 /// available scan tasks.
 struct ICEBERG_REST_EXPORT FetchPlanningResultResponse : BaseScanTaskResponse {
-  std::string plan_status;
+  PlanStatus plan_status = PlanStatus::kCompleted;
   // TODO(sandeepg): Add credentials.
 
   Status Validate() const;

src/iceberg/json_serde.cc

@@ -1894,15 +1894,16 @@ Result<DataFile> DataFileFromJson(
   return df;
 }
 
-Result<std::vector<FileScanTask>> FileScanTasksFromJson(
-    const nlohmann::json& json, const std::vector<DataFile>& delete_files,
+Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
+    const nlohmann::json& json,
+    const std::vector<std::shared_ptr<DataFile>>& delete_files,
     const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& partitionSpecById,
     const Schema& schema) {
   if (!json.is_array()) {
     return JsonParseError("Cannot parse file scan tasks from non-array: {}",
                           SafeDumpJson(json));
   }
-  std::vector<FileScanTask> file_scan_tasks;
+  std::vector<std::shared_ptr<FileScanTask>> file_scan_tasks;
   for (const auto& task_json : json) {
     if (!task_json.is_object()) {
       return JsonParseError("Cannot parse file scan task from a non-object: {}",
@@ -1925,7 +1926,7 @@ Result<std::vector<FileScanTask>> FileScanTasksFromJson(
               "delete-file-references index {} is out of range (delete_files size: {})",
               ref, delete_files.size());
         }
-        task_delete_files.push_back(std::make_shared<DataFile>(delete_files[ref]));
+        task_delete_files.push_back(delete_files[ref]);
       }
     }
 
@@ -1936,9 +1937,10 @@ Result<std::vector<FileScanTask>> FileScanTasksFromJson(
       ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json));
     }
 
-    file_scan_tasks.emplace_back(std::make_shared<DataFile>(std::move(data_file)),
-                                 std::move(task_delete_files),
-                                 std::move(residual_filter));
+    file_scan_tasks.push_back(
+        std::make_shared<FileScanTask>(std::make_shared<DataFile>(std::move(data_file)),
+                                      std::move(task_delete_files),
+                                      std::move(residual_filter)));
   }
   return file_scan_tasks;
 }

src/iceberg/json_serde_internal.h

@@ -447,8 +447,9 @@ ICEBERG_EXPORT Result<DataFile> DataFileFromJson(
 /// \param schema The table schema, used with partitionSpecById to resolve partition
 /// types.
 /// \return A vector of `FileScanTask` objects or an error if the conversion fails.
-ICEBERG_EXPORT Result<std::vector<FileScanTask>> FileScanTasksFromJson(
-    const nlohmann::json& json, const std::vector<DataFile>& delete_files,
+ICEBERG_EXPORT Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
+    const nlohmann::json& json,
+    const std::vector<std::shared_ptr<DataFile>>& delete_files,
     const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& partitionSpecById,
     const Schema& schema);
 

src/iceberg/table_scan.cc

@@ -765,4 +765,22 @@ IncrementalChangelogScan::PlanFiles(std::optional<int64_t> from_snapshot_id_excl
   return NotImplemented("IncrementalChangelogScan::PlanFiles is not implemented");
 }
 
+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);
+}
+
 }  // namespace iceberg

src/iceberg/table_scan.h

@@ -23,6 +23,7 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <string_view>
 #include <unordered_map>
 #include <unordered_set>
 #include <vector>
@@ -35,6 +36,17 @@
 
 namespace iceberg {
 
+/// \brief Status of a server-side scan planning operation.
+enum class PlanStatus {
+  kSubmitted,
+  kCompleted,
+  kCancelled,
+  kFailed,
+};
+
+ICEBERG_EXPORT std::string_view ToString(PlanStatus status);
+ICEBERG_EXPORT Result<PlanStatus> PlanStatusFromString(std::string_view status_str);
+
 /// \brief An abstract scan task.
 class ICEBERG_EXPORT ScanTask {
  public:

src/iceberg/test/json_serde_test.cc

@@ -942,10 +942,10 @@ TEST(FileScanTasksFromJsonTest, SingleTaskNoDeleteFiles) {
   ASSERT_THAT(result, IsOk());
   ASSERT_EQ(result.value().size(), 1U);
   const auto& task = result.value()[0];
-  ASSERT_NE(task.data_file(), nullptr);
-  EXPECT_EQ(task.data_file()->file_path, "s3://bucket/data/file.parquet");
-  EXPECT_TRUE(task.delete_files().empty());
-  EXPECT_EQ(task.residual_filter(), nullptr);
+  ASSERT_NE(task->data_file(), nullptr);
+  EXPECT_EQ(task->data_file()->file_path, "s3://bucket/data/file.parquet");
+  EXPECT_TRUE(task->delete_files().empty());
+  EXPECT_EQ(task->residual_filter(), nullptr);
 }
 
 TEST(FileScanTasksFromJsonTest, TaskWithDeleteFileReferences) {
@@ -967,12 +967,13 @@ TEST(FileScanTasksFromJsonTest, TaskWithDeleteFileReferences) {
     "delete-file-references": [0]
   }])"_json;
 
-  auto result = FileScanTasksFromJson(json, {delete_file}, {}, Schema({}, 0));
+  auto result =
+      FileScanTasksFromJson(json, {std::make_shared<DataFile>(delete_file)}, {}, Schema({}, 0));
   ASSERT_THAT(result, IsOk());
   ASSERT_EQ(result.value().size(), 1U);
   const auto& task = result.value()[0];
-  ASSERT_EQ(task.delete_files().size(), 1U);
-  EXPECT_EQ(task.delete_files()[0]->file_path, "s3://bucket/deletes/pos_delete.parquet");
+  ASSERT_EQ(task->delete_files().size(), 1U);
+  EXPECT_EQ(task->delete_files()[0]->file_path, "s3://bucket/deletes/pos_delete.parquet");
 }
 
 TEST(FileScanTasksFromJsonTest, DeleteFileReferenceOutOfRange) {

src/iceberg/test/rest_json_serde_test.cc

@@ -1395,7 +1395,7 @@ TEST(PlanTableScanResponseFromJsonTest, SubmittedStatusMissingOptionalFields) {
   auto json = nlohmann::json::parse(R"({"status":"submitted","plan-id":"abc-123"})");
   auto result = PlanTableScanResponseFromJson(json, EmptySpecs(), EmptySchema());
   ASSERT_THAT(result, IsOk());
-  EXPECT_EQ(result->plan_status, "submitted");
+  EXPECT_EQ(result->plan_status, PlanStatus::kSubmitted);
   EXPECT_EQ(result->plan_id, "abc-123");
   EXPECT_TRUE(result->plan_tasks.empty());
   EXPECT_TRUE(result->file_scan_tasks.empty());
@@ -1408,7 +1408,7 @@ TEST(PlanTableScanResponseFromJsonTest, CompletedStatusWithPlanTasks) {
       R"({"status":"completed","plan-id":"abc-123","plan-tasks":["task-1","task-2"],"delete-files":[],"file-scan-tasks":[]})");
   auto result = PlanTableScanResponseFromJson(json, EmptySpecs(), EmptySchema());
   ASSERT_THAT(result, IsOk());
-  EXPECT_EQ(result->plan_status, "completed");
+  EXPECT_EQ(result->plan_status, PlanStatus::kCompleted);
   EXPECT_EQ(result->plan_id, "abc-123");
   ASSERT_EQ(result->plan_tasks.size(), 2);
   EXPECT_EQ(result->plan_tasks[0], "task-1");
@@ -1436,7 +1436,7 @@ TEST(FetchPlanningResultResponseFromJsonTest, SubmittedStatusNoTasks) {
   auto json = nlohmann::json::parse(R"({"status":"submitted"})");
   auto result = FetchPlanningResultResponseFromJson(json, EmptySpecs(), EmptySchema());
   ASSERT_THAT(result, IsOk());
-  EXPECT_EQ(result->plan_status, "submitted");
+  EXPECT_EQ(result->plan_status, PlanStatus::kSubmitted);
   EXPECT_TRUE(result->plan_tasks.empty());
   EXPECT_TRUE(result->file_scan_tasks.empty());
   EXPECT_TRUE(result->delete_files.empty());
@@ -1447,7 +1447,7 @@ TEST(FetchPlanningResultResponseFromJsonTest, CompletedStatusWithPlanTasks) {
       R"({"status":"completed","plan-tasks":["task-1"],"delete-files":[],"file-scan-tasks":[]})");
   auto result = FetchPlanningResultResponseFromJson(json, EmptySpecs(), EmptySchema());
   ASSERT_THAT(result, IsOk());
-  EXPECT_EQ(result->plan_status, "completed");
+  EXPECT_EQ(result->plan_status, PlanStatus::kCompleted);
   ASSERT_EQ(result->plan_tasks.size(), 1);
   EXPECT_EQ(result->plan_tasks[0], "task-1");
 }
@@ -1492,10 +1492,10 @@ TEST(FetchScanTasksResponseFromJsonTest, WithFileScanTasks) {
   EXPECT_TRUE(result->plan_tasks.empty());
   ASSERT_EQ(result->delete_files.size(), 1);
   ASSERT_EQ(result->file_scan_tasks.size(), 1);
-  EXPECT_EQ(result->file_scan_tasks[0].data_file()->file_path,
+  EXPECT_EQ(result->file_scan_tasks[0]->data_file()->file_path,
             "s3://bucket/data/file.parquet");
-  ASSERT_EQ(result->file_scan_tasks[0].delete_files().size(), 1);
-  EXPECT_EQ(result->file_scan_tasks[0].delete_files()[0]->file_path,
+  ASSERT_EQ(result->file_scan_tasks[0]->delete_files().size(), 1);
+  EXPECT_EQ(result->file_scan_tasks[0]->delete_files()[0]->file_path,
             "s3://bucket/deletes/delete.parquet");
 }