test(serde): cover FileScanTask range and spec boundaries

Author: kamcheungting-db

src/iceberg/test/json_serde_test.cc

@@ -18,6 +18,7 @@
  */
 
 #include <memory>
+#include <optional>
 #include <string>
 #include <string_view>
 #include <unordered_map>
@@ -1612,6 +1613,34 @@ TEST_F(FileScanTaskJsonTest, RoundTripsHistoricalSpecWithDroppedSourceField) {
   EXPECT_EQ(serialized, json);
 }
 
+TEST_F(FileScanTaskJsonTest, RejectsNonNullPartitionForDroppedSourceField) {
+  auto json = JavaGolden();
+  json["schema"]["fields"] = nlohmann::json::array();
+  json["data-file"]["partition"] = {7};
+
+  // The embedded schema no longer contains the source type, so neither Java nor
+  // C++ can decode a non-null value for this historical partition field.
+  auto result = FileScanTaskFromJson(json);
+  EXPECT_THAT(result, IsError(ErrorKind::kNotSupported));
+  EXPECT_THAT(result, HasErrorMessage("Unsupported type for literal JSON parsing"));
+}
+
+TEST_F(FileScanTaskJsonTest, RejectsFilesWithMismatchedEmbeddedSpecId) {
+  for (bool mismatch_delete_file : {false, true}) {
+    SCOPED_TRACE(mismatch_delete_file ? "delete-file" : "data-file");
+    auto json = JavaGolden();
+    if (mismatch_delete_file) {
+      json["delete-files"][0]["spec-id"] = 1;
+    } else {
+      json["data-file"]["spec-id"] = 1;
+    }
+
+    auto result = FileScanTaskFromJson(json);
+    EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(result, HasErrorMessage("Invalid partition spec id"));
+  }
+}
+
 TEST_F(FileScanTaskJsonTest, AcceptsMissingPartitionLikeJava) {
   auto json = JavaGolden();
   json["data-file"].erase("partition");
@@ -1684,6 +1713,21 @@ TEST_F(FileScanTaskJsonTest, RejectsInvalidContentRolesWhenSerializing) {
   EXPECT_THAT(delete_result, HasErrorMessage("delete-file"));
 }
 
+TEST_F(FileScanTaskJsonTest, RejectsDeleteFilesWithDifferentSpecIdWhenSerializing) {
+  for (const auto& spec_id : {std::optional<int32_t>(1), std::optional<int32_t>()}) {
+    SCOPED_TRACE(spec_id.value_or(-1));
+    auto delete_file = DeleteFileForTask();
+    delete_file.partition_spec_id = spec_id;
+    FileScanTask task(std::make_shared<DataFile>(DataFileForTask()),
+                      {std::make_shared<DataFile>(std::move(delete_file))});
+
+    auto result = ToJson(task, specs_, schema_);
+    EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+    EXPECT_THAT(result, HasErrorMessage("Invalid partition spec id"));
+    EXPECT_THAT(result, HasErrorMessage(spec_id ? "actual = 1" : "actual = null"));
+  }
+}
+
 TEST_F(FileScanTaskJsonTest, RejectsInvalidPartitionValueTypeWhenSerializing) {
   auto data_file = DataFileForTask();
   data_file.partition = PartitionValues({Literal::String("not-an-int")});

src/iceberg/test/rest_json_serde_test.cc

@@ -2305,6 +2305,58 @@ TEST(FileScanTasksFromJsonTest, AcceptsWholeFileStartAndLength) {
               IsOk());
 }
 
+TEST(FileScanTasksFromJsonTest, NullStartAndLengthUseWholeFileDefaults) {
+  auto json = R"([{
+    "data-file": {
+      "content": "data",
+      "file-path": "s3://bucket/data/file.parquet",
+      "file-format": "PARQUET",
+      "spec-id": 0,
+      "partition": [],
+      "file-size-in-bytes": 12345,
+      "record-count": 100
+    },
+    "start": null,
+    "length": null
+  }])"_json;
+
+  EXPECT_THAT(FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0)),
+              IsOk());
+}
+
+TEST(FileScanTasksFromJsonTest, RejectsInvalidRangeIntegers) {
+  auto valid_json = R"([{
+    "data-file": {
+      "content": "data",
+      "file-path": "s3://bucket/data/file.parquet",
+      "file-format": "PARQUET",
+      "spec-id": 0,
+      "partition": [],
+      "file-size-in-bytes": 12345,
+      "record-count": 100
+    },
+    "start": 0,
+    "length": 12345
+  }])"_json;
+
+  for (std::string_view field : {"start", "length"}) {
+    for (const auto& [invalid_value, expected_error] :
+         {std::pair<nlohmann::json, std::string_view>{0.5, "must be an integer"},
+          {true, "must be an integer"},
+          {"0", "must be an integer"},
+          {18446744073709551615ULL, "out of range"}}) {
+      SCOPED_TRACE(testing::Message() << field << "=" << invalid_value.dump());
+      auto json = valid_json;
+      json[0][field] = invalid_value;
+
+      auto result = FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0));
+      EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+      EXPECT_THAT(result, HasErrorMessage(field));
+      EXPECT_THAT(result, HasErrorMessage(expected_error));
+    }
+  }
+}
+
 TEST(FileScanTasksFromJsonTest, RejectsSplitTaskWhenEitherRangeFieldDiffers) {
   auto json = R"([{
     "data-file": {