Core: Fix REST scan tasks dropping content file sequence numbers

core/src/main/java/org/apache/iceberg/ContentFileParser.java

@@ -48,6 +48,8 @@ public class ContentFileParser {
   private static final String EQUALITY_IDS = "equality-ids";
   private static final String SORT_ORDER_ID = "sort-order-id";
   private static final String FIRST_ROW_ID = "first-row-id";
+  private static final String DATA_SEQUENCE_NUMBER = "data-sequence-number";
+  private static final String FILE_SEQUENCE_NUMBER = "file-sequence-number";
   private static final String REFERENCED_DATA_FILE = "referenced-data-file";
   private static final String CONTENT_OFFSET = "content-offset";
   private static final String CONTENT_SIZE = "content-size-in-bytes";
@@ -122,6 +124,8 @@ public static void toJson(ContentFile<?> contentFile, PartitionSpec spec, JsonGe
     }
 
     JsonUtil.writeLongFieldIfPresent(FIRST_ROW_ID, contentFile.firstRowId(), generator);
+    JsonUtil.writeLongFieldIfPresent(DATA_SEQUENCE_NUMBER, contentFile.dataSequenceNumber(), generator);
+    JsonUtil.writeLongFieldIfPresent(FILE_SEQUENCE_NUMBER, contentFile.fileSequenceNumber(), generator);
 
     if (contentFile instanceof DeleteFile) {
       DeleteFile deleteFile = (DeleteFile) contentFile;
@@ -170,38 +174,48 @@ public static ContentFile<?> fromJson(JsonNode jsonNode, Map<Integer, PartitionS
     int[] equalityFieldIds = JsonUtil.getIntArrayOrNull(EQUALITY_IDS, jsonNode);
     Integer sortOrderId = JsonUtil.getIntOrNull(SORT_ORDER_ID, jsonNode);
     Long firstRowId = JsonUtil.getLongOrNull(FIRST_ROW_ID, jsonNode);
+    Long dataSequenceNumber = JsonUtil.getLongOrNull(DATA_SEQUENCE_NUMBER, jsonNode);
+    Long fileSequenceNumber = JsonUtil.getLongOrNull(FILE_SEQUENCE_NUMBER, jsonNode);
     String referencedDataFile = JsonUtil.getStringOrNull(REFERENCED_DATA_FILE, jsonNode);
     Long contentOffset = JsonUtil.getLongOrNull(CONTENT_OFFSET, jsonNode);
     Long contentSizeInBytes = JsonUtil.getLongOrNull(CONTENT_SIZE, jsonNode);
 
     if (fileContent == FileContent.DATA) {
-      return new GenericDataFile(
-          specId,
-          filePath,
-          fileFormat,
-          partitionData,
-          fileSizeInBytes,
-          metrics,
-          keyMetadata,
-          splitOffsets,
-          sortOrderId,
-          firstRowId);
+      GenericDataFile dataFile =
+          new GenericDataFile(
+              specId,
+              filePath,
+              fileFormat,
+              partitionData,
+              fileSizeInBytes,
+              metrics,
+              keyMetadata,
+              splitOffsets,
+              sortOrderId,
+              firstRowId);
+      dataFile.setDataSequenceNumber(dataSequenceNumber);
+      dataFile.setFileSequenceNumber(fileSequenceNumber);
+      return dataFile;
     } else {
-      return new GenericDeleteFile(
-          specId,
-          fileContent,
-          filePath,
-          fileFormat,
-          partitionData,
-          fileSizeInBytes,
-          metrics,
-          equalityFieldIds,
-          sortOrderId,
-          splitOffsets,
-          keyMetadata,
-          referencedDataFile,
-          contentOffset,
-          contentSizeInBytes);
+      GenericDeleteFile deleteFile =
+          new GenericDeleteFile(
+              specId,
+              fileContent,
+              filePath,
+              fileFormat,
+              partitionData,
+              fileSizeInBytes,
+              metrics,
+              equalityFieldIds,
+              sortOrderId,
+              splitOffsets,
+              keyMetadata,
+              referencedDataFile,
+              contentOffset,
+              contentSizeInBytes);
+      deleteFile.setDataSequenceNumber(dataSequenceNumber);
+      deleteFile.setFileSequenceNumber(fileSequenceNumber);
+      return deleteFile;
     }
   }
 

core/src/test/java/org/apache/iceberg/TestContentFileParser.java

@@ -271,6 +271,69 @@ public void testEnumContentTypeSerialization(FileContent content, String expecte
     assertThat(serializedStr).contains("\"content\":\"" + expectedJsonContent + "\"");
   }
 
+  @Test
+  public void testSequenceNumberRoundTrip() throws Exception {
+    PartitionSpec spec = PartitionSpec.unpartitioned();
+    String jsonStr =
+        "{\"spec-id\":0,\"content\":\"data\",\"file-path\":\"/path/to/data.parquet\","
+            + "\"file-format\":\"parquet\",\"partition\":[],\"file-size-in-bytes\":10,"
+            + "\"record-count\":1,\"data-sequence-number\":5,\"file-sequence-number\":7}";
+
+    JsonNode jsonNode = JsonUtil.mapper().readTree(jsonStr);
+    ContentFile<?> deserialized =
+        ContentFileParser.fromJson(jsonNode, Map.of(0, spec));
+    assertThat(deserialized).isInstanceOf(DataFile.class);
+    assertThat(deserialized.dataSequenceNumber()).isEqualTo(5L);
+    assertThat(deserialized.fileSequenceNumber()).isEqualTo(7L);
+
+    String reSerialized = ContentFileParser.toJson(deserialized, spec);
+    assertThat(reSerialized).contains("\"data-sequence-number\":5");
+    assertThat(reSerialized).contains("\"file-sequence-number\":7");
+
+    // Verify round-trip
+    ContentFile<?> reParsed =
+        ContentFileParser.fromJson(
+            JsonUtil.mapper().readTree(reSerialized), Map.of(0, spec));
+    assertThat(reParsed.dataSequenceNumber()).isEqualTo(5L);
+    assertThat(reParsed.fileSequenceNumber()).isEqualTo(7L);
+  }
+
+  @Test
+  public void testSequenceNumberDefaultsToNull() throws Exception {
+    // JSON without sequence numbers should deserialize to null (not 0)
+    PartitionSpec spec = PartitionSpec.unpartitioned();
+    String jsonStr =
+        "{\"spec-id\":0,\"content\":\"data\",\"file-path\":\"/path/to/data.parquet\","
+            + "\"file-format\":\"parquet\",\"partition\":[],\"file-size-in-bytes\":10,"
+            + "\"record-count\":1}";
+
+    JsonNode jsonNode = JsonUtil.mapper().readTree(jsonStr);
+    ContentFile<?> deserialized =
+        ContentFileParser.fromJson(jsonNode, Map.of(0, spec));
+    assertThat(deserialized.dataSequenceNumber()).isNull();
+    assertThat(deserialized.fileSequenceNumber()).isNull();
+  }
+
+  @Test
+  public void testSequenceNumberForDeleteFile() throws Exception {
+    PartitionSpec spec = PartitionSpec.unpartitioned();
+    String jsonStr =
+        "{\"spec-id\":0,\"content\":\"position-deletes\",\"file-path\":\"/path/to/delete.parquet\","
+            + "\"file-format\":\"parquet\",\"partition\":[],\"file-size-in-bytes\":1234,"
+            + "\"record-count\":10,\"data-sequence-number\":3,\"file-sequence-number\":4}";
+
+    JsonNode jsonNode = JsonUtil.mapper().readTree(jsonStr);
+    ContentFile<?> deserialized =
+        ContentFileParser.fromJson(jsonNode, Map.of(0, spec));
+    assertThat(deserialized).isInstanceOf(DeleteFile.class);
+    assertThat(deserialized.dataSequenceNumber()).isEqualTo(3L);
+    assertThat(deserialized.fileSequenceNumber()).isEqualTo(4L);
+
+    String reSerialized = ContentFileParser.toJson(deserialized, spec);
+    assertThat(reSerialized).contains("\"data-sequence-number\":3");
+    assertThat(reSerialized).contains("\"file-sequence-number\":4");
+  }
+
   private static Stream<Arguments> enumContentTypeCases() {
     return Stream.of(
         Arguments.of(FileContent.DATA, "data"),

open-api/rest-catalog-open-api.yaml

@@ -5222,6 +5222,17 @@ components:
           description: "List of splittable offsets"
         sort-order-id:
           type: integer
+        data-sequence-number:
+          type: integer
+          format: int64
+          description:
+            "The data sequence number of the file, used to determine which delete
+            files apply to it during reads (see the Iceberg spec)"
+        file-sequence-number:
+          type: integer
+          format: int64
+          description:
+            "The sequence number of the snapshot in which the underlying file was added"
 
     DataFile:
       allOf: