fix(serde): address FileScanTask JSON review feedback

- reuse core strict integer parsing (GetJsonIntegerOrDefault) in the REST
  scan-task reader instead of a local copy
- omit 'delete-files' when the task has no delete files, matching Java
- look up the partition spec explicitly in ToJson(FileScanTask) instead of
  relying on std::map::at() after an earlier validation

Author: wgtmac

src/iceberg/catalog/rest/json_serde.cc

@@ -18,7 +18,6 @@
  */
 
 #include <iterator>
-#include <limits>
 #include <map>
 #include <memory>
 #include <optional>
@@ -112,23 +111,6 @@ constexpr std::string_view kResidualFilter = "residual-filter";
 constexpr std::string_view kStart = "start";
 constexpr std::string_view kLength = "length";
 
-Result<int64_t> GetRangeValueOrDefault(const nlohmann::json& json, std::string_view key,
-                                       int64_t default_value) {
-  if (!json.contains(key) || json.at(key).is_null()) {
-    return default_value;
-  }
-  const auto& value = json.at(key);
-  if (!value.is_number_integer()) {
-    return JsonParseError("'{}' must be an integer, but is {}", key, value.type_name());
-  }
-  if (value.is_number_unsigned() &&
-      value.get<uint64_t>() >
-          static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
-    return JsonParseError("'{}' integer is out of range: {}", key, SafeDumpJson(value));
-  }
-  return value.get<int64_t>();
-}
-
 Result<nlohmann::json> StorageCredentialToJson(const StorageCredential& credential) {
   ICEBERG_RETURN_UNEXPECTED(credential.Validate());
   nlohmann::json json;
@@ -188,10 +170,10 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
     ICEBERG_ASSIGN_OR_RAISE(
         auto data_file,
         iceberg::DataFileFromJson(data_file_json, partition_spec_by_id, schema));
-    ICEBERG_ASSIGN_OR_RAISE(auto start, GetRangeValueOrDefault(task_json, kStart, 0));
+    ICEBERG_ASSIGN_OR_RAISE(auto start, GetJsonIntegerOrDefault(task_json, kStart, 0));
     ICEBERG_ASSIGN_OR_RAISE(
         auto length,
-        GetRangeValueOrDefault(task_json, kLength, data_file.file_size_in_bytes));
+        GetJsonIntegerOrDefault(task_json, kLength, data_file.file_size_in_bytes));
     ICEBERG_RETURN_UNEXPECTED(
         CheckFileScanTaskNotSplit(start, length, data_file.file_size_in_bytes));
     // FIXME: REST scan-task DataFile JSON currently carries first-row-id,

src/iceberg/json_serde.cc

@@ -574,6 +574,14 @@ nlohmann::json ToJson(const SortField& sort_field) {
   return json;
 }
 
+Result<int64_t> GetJsonIntegerOrDefault(const nlohmann::json& json, std::string_view key,
+                                        int64_t default_value) {
+  if (!json.contains(key) || json.at(key).is_null()) {
+    return default_value;
+  }
+  return IntegerFromJson<int64_t>(json.at(key), std::format("'{}'", key));
+}
+
 nlohmann::json ToJson(const SortOrder& sort_order) {
   nlohmann::json json;
   json[kOrderId] = sort_order.order_id();
@@ -2466,7 +2474,11 @@ Result<nlohmann::json> ToJson(
   ICEBERG_ASSIGN_OR_RAISE(auto data_file_json,
                           ToJson(*task.data_file(), partition_specs_by_id, schema));
   const auto spec_id = task.data_file()->partition_spec_id.value();
-  json[kSpec] = ToJson(*partition_specs_by_id.at(spec_id));
+  const auto spec_it = partition_specs_by_id.find(spec_id);
+  if (spec_it == partition_specs_by_id.end() || !spec_it->second) {
+    return ValidationFailed("Invalid partition spec: null");
+  }
+  json[kSpec] = ToJson(*spec_it->second);
   json[kDataFile] = std::move(data_file_json);
   json[kStart] = 0;
   json[kLength] = task.data_file()->file_size_in_bytes;
@@ -2491,7 +2503,9 @@ Result<nlohmann::json> ToJson(
                             ToJson(*delete_file, partition_specs_by_id, schema));
     delete_files_json.push_back(std::move(delete_file_json));
   }
-  json[kDeleteFiles] = std::move(delete_files_json);
+  if (!delete_files_json.empty()) {
+    json[kDeleteFiles] = std::move(delete_files_json);
+  }
 
   if (task.residual_filter()) {
     ICEBERG_ASSIGN_OR_RAISE(auto residual_json, ToJson(*task.residual_filter()));

src/iceberg/json_serde_internal.h

@@ -445,6 +445,15 @@ ICEBERG_EXPORT Result<DataFile> DataFileFromJson(
         partition_spec_by_id,
     const Schema& schema);
 
+/// \brief Parses an optional signed integer field, returning `default_value` when the
+/// key is missing or null.
+///
+/// The value must be a JSON integer that fits `int64_t`, with the same strictness as
+/// the integer parsing used by `DataFile`/`FileScanTask`.
+ICEBERG_EXPORT Result<int64_t> GetJsonIntegerOrDefault(const nlohmann::json& json,
+                                                       std::string_view key,
+                                                       int64_t default_value);
+
 /// \brief Serializes a `FileScanTask` to a self-contained JSON object.
 ///
 /// Unlike REST scan responses, delete files are inlined under `delete-files` rather
@@ -457,7 +466,8 @@ ICEBERG_EXPORT Result<DataFile> DataFileFromJson(
 /// - `data-file` (required): ContentFile JSON
 /// - `start` (required): always 0 because split tasks are unsupported
 /// - `length` (required): always the data file size
-/// - `delete-files` (optional on input; always emitted): array of ContentFile JSON
+/// - `delete-files` (optional on input; omitted when the task has no delete files):
+///   array of ContentFile JSON
 /// - `residual-filter` (optional): Expression JSON
 ICEBERG_EXPORT Result<nlohmann::json> ToJson(
     const FileScanTask& task,

src/iceberg/test/json_serde_test.cc

@@ -1628,11 +1628,11 @@ TEST_F(FileScanTaskJsonTest, ParsesJavaGoldenWithoutExternalSchemaOrSpec) {
   EXPECT_EQ(residual, true);
 }
 
-TEST_F(FileScanTaskJsonTest, SerializesEmptyDeleteFilesArray) {
+TEST_F(FileScanTaskJsonTest, OmitsEmptyDeleteFilesLikeJava) {
   FileScanTask task(std::make_shared<DataFile>(DataFileForTask()));
 
   ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(task, specs_, schema_));
-  EXPECT_EQ(json["delete-files"], nlohmann::json::array());
+  EXPECT_FALSE(json.contains("delete-files"));
   EXPECT_EQ(json["start"], 0);
   EXPECT_EQ(json["length"], 10);
 }