From 854948156db9dc00cff5fce5157c32926a0aa11f Mon Sep 17 00:00:00 2001
From: kid <lucian1412@outlook.com>
Date: Fri, 24 Jul 2026 01:34:49 +0800
Subject: [PATCH] feat(rest): support data sequence numbers in scan tasks

---
 src/iceberg/catalog/rest/json_serde.cc   | 14 ++++++---
 src/iceberg/test/rest_json_serde_test.cc | 37 ++++++++++++++++++++----
 2 files changed, 41 insertions(+), 10 deletions(-)

diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc
index 3ce753f18..314560c72 100644
--- a/src/iceberg/catalog/rest/json_serde.cc
+++ b/src/iceberg/catalog/rest/json_serde.cc
@@ -130,6 +130,7 @@ constexpr std::string_view kReferencedDataFile = "referenced-data-file";
 constexpr std::string_view kContentOffset = "content-offset";
 constexpr std::string_view kContentSizeInBytes = "content-size-in-bytes";
 constexpr std::string_view kDataFile = "data-file";
+constexpr std::string_view kDataSequenceNumber = "data-sequence-number";
 constexpr std::string_view kDeleteFileReferences = "delete-file-references";
 constexpr std::string_view kResidualFilter = "residual-filter";
 constexpr std::string_view kMapKeys = "keys";
@@ -323,10 +324,11 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
                             GetJsonValue<nlohmann::json>(task_json, kDataFile));
     ICEBERG_ASSIGN_OR_RAISE(
         auto data_file, DataFileFromJson(data_file_json, partition_spec_by_id, schema));
-    // FIXME: REST scan-task DataFile JSON currently carries first-row-id,
-    // but not the manifest-entry data sequence number. Until the REST API exposes
-    // it, REST-planned tasks cannot inherit _last_updated_sequence_number.
-    // See https://github.com/apache/iceberg-cpp/issues/834.
+    if (task_json.contains(kDataSequenceNumber) &&
+        !task_json.at(kDataSequenceNumber).is_null()) {
+      ICEBERG_ASSIGN_OR_RAISE(data_file.data_sequence_number,
+                              GetJsonValue<int64_t>(task_json, kDataSequenceNumber));
+    }
 
     std::vector<std::shared_ptr<DataFile>> task_delete_files;
     if (task_json.contains(kDeleteFileReferences) &&
@@ -497,6 +499,10 @@ Result<nlohmann::json> ScanTaskFieldsToJson(
             auto data_file_json,
             ToJson(*task->data_file(), partition_specs_by_id, schema));
         task_json[kDataFile] = std::move(data_file_json);
+        if (task->data_file()->data_sequence_number.has_value()) {
+          task_json[kDataSequenceNumber] =
+              task->data_file()->data_sequence_number.value();
+        }
       }
       if (!task->delete_files().empty()) {
         std::vector<int32_t> refs;
diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc
index ec41e4a66..3208fa894 100644
--- a/src/iceberg/test/rest_json_serde_test.cc
+++ b/src/iceberg/test/rest_json_serde_test.cc
@@ -2280,14 +2280,32 @@ TEST(FileScanTasksFromJsonTest, SingleTaskNoDeleteFiles) {
   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_FALSE(task->data_file()->data_sequence_number.has_value());
   EXPECT_TRUE(task->delete_files().empty());
   EXPECT_EQ(task->residual_filter(), nullptr);
 }
 
-TEST(FileScanTasksFromJsonTest, RowLineageSequence) {
-  GTEST_SKIP() << "REST scan-task JSON does not expose data-sequence-number yet: "
-               << "https://github.com/apache/iceberg-cpp/issues/834";
+TEST(FileScanTasksFromJsonTest, NullDataSequenceNumber) {
+  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
+    },
+    "data-sequence-number": null
+  }])"_json;
 
+  auto result = FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0));
+  ASSERT_THAT(result, IsOk());
+  ASSERT_EQ(result.value().size(), 1U);
+  EXPECT_FALSE(result.value()[0]->data_file()->data_sequence_number.has_value());
+}
+
+TEST(FileScanTasksFromJsonTest, RowLineageSequence) {
   auto json = R"([{
     "data-file": {
       "content": "data",
@@ -2297,9 +2315,9 @@ TEST(FileScanTasksFromJsonTest, RowLineageSequence) {
       "partition": [],
       "file-size-in-bytes": 12345,
       "record-count": 100,
-      "first-row-id": 100,
-      "data-sequence-number": 7
-    }
+      "first-row-id": 100
+    },
+    "data-sequence-number": 7
   }])"_json;
 
   auto result = FileScanTasksFromJson(json, {}, UnpartitionedSpecs(), Schema({}, 0));
@@ -2309,6 +2327,13 @@ TEST(FileScanTasksFromJsonTest, RowLineageSequence) {
   ASSERT_NE(data_file, nullptr);
   EXPECT_EQ(data_file->first_row_id, 100);
   EXPECT_EQ(data_file->data_sequence_number, 7);
+
+  FetchScanTasksResponse response;
+  response.file_scan_tasks = std::move(result.value());
+  ICEBERG_UNWRAP_OR_FAIL(auto roundtrip_json,
+                         ToJson(response, UnpartitionedSpecs(), Schema({}, 0)));
+  ASSERT_EQ(roundtrip_json["file-scan-tasks"].size(), 1);
+  EXPECT_EQ(roundtrip_json["file-scan-tasks"][0]["data-sequence-number"], 7);
 }
 
 TEST(FileScanTasksFromJsonTest, TaskWithDeleteFileReferences) {