From d161e78677016609138d55d0d00ba2f31afa33f5 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Wed, 8 Jul 2026 20:10:02 +0900
Subject: [PATCH 1/6] Support writing V3 manifests and manifest lists

Add ManifestWriterV3 and ManifestListWriterV3. The manifest writer
uses the V3 record layout so the V3-only data file fields
(first_row_id, referenced_data_file, content_offset,
content_size_in_bytes) are written, rebinding V2-layout records when
needed. The manifest list writer implements the spec's First Row ID
Assignment: preserving existing first_row_id values, never assigning
one to delete manifests, and assigning a running value advanced by
existing and added row counts to unassigned data manifests. It also
exposes next_row_id for the commit path to update the table's
next-row-id.

Closes #3620
---
 pyiceberg/manifest.py        | 130 +++++++++++++++++++++++++++++++++++
 tests/utils/test_manifest.py | 129 ++++++++++++++++++++++++++++++++++
 2 files changed, 259 insertions(+)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index 9842f79d8e..04c6b29b59 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -853,6 +853,17 @@ def partitions(self) -> list[PartitionFieldSummary] | None:
     def key_metadata(self) -> bytes | None:
         return self._data[14]
 
+    @property
+    def first_row_id(self) -> int | None:
+        # Only present when the record is bound to the V3 manifest list schema
+        return self._data[15] if len(self._data) > 15 else None
+
+    @first_row_id.setter
+    def first_row_id(self, value: int | None) -> None:
+        if len(self._data) <= 15:
+            raise ValueError("Cannot set first_row_id on a manifest file not bound to the V3 schema")
+        self._data[15] = value
+
     def has_added_files(self) -> bool:
         return self.added_files_count is None or self.added_files_count > 0
 
@@ -1284,6 +1295,51 @@ def prepare_entry(self, entry: ManifestEntry) -> ManifestEntry:
         return entry
 
 
+class ManifestWriterV3(ManifestWriterV2):
+    """Writes V3 manifest files.
+
+    The writer inherits the V2 sequence-number semantics; the V3 manifest entry
+    schema additionally carries `first_row_id`, `referenced_data_file`,
+    `content_offset` and `content_size_in_bytes` on the data file struct.
+    """
+
+    @property
+    def version(self) -> TableVersion:
+        return 3
+
+    def new_writer(self) -> AvroOutputFile[ManifestEntry]:
+        # Use the V3 record layout so the V3-only data file fields are written
+        return AvroOutputFile[ManifestEntry](
+            output_file=self._output_file,
+            file_schema=self._with_partition(3),
+            record_schema=self._with_partition(3),
+            schema_name="manifest_entry",
+            metadata=self._meta,
+        )
+
+    def _wrap_data_file(self, data_file: DataFile) -> DataFile:
+        """Rebind a data file to the V3 record layout."""
+        if len(data_file._data) >= len(DATA_FILE_TYPE[3].fields):
+            return data_file
+        args = {
+            field.name: value for field, value in zip(DATA_FILE_TYPE[DEFAULT_READ_VERSION].fields, data_file._data, strict=True)
+        }
+        return DataFile.from_args(_table_format_version=3, **args)
+
+    def prepare_entry(self, entry: ManifestEntry) -> ManifestEntry:
+        entry = super().prepare_entry(entry)
+        if len(entry.data_file._data) < len(DATA_FILE_TYPE[3].fields):
+            entry = ManifestEntry.from_args(
+                _table_format_version=3,
+                status=entry.status,
+                snapshot_id=entry.snapshot_id,
+                sequence_number=entry.sequence_number,
+                file_sequence_number=entry.file_sequence_number,
+                data_file=self._wrap_data_file(entry.data_file),
+            )
+        return entry
+
+
 def write_manifest(
     format_version: TableVersion,
     spec: PartitionSpec,
@@ -1296,6 +1352,8 @@ def write_manifest(
         return ManifestWriterV1(spec, schema, output_file, snapshot_id, avro_compression)
     elif format_version == 2:
         return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression)
+    elif format_version == 3:
+        return ManifestWriterV3(spec, schema, output_file, snapshot_id, avro_compression)
     else:
         raise ValueError(f"Cannot write manifest for table version: {format_version}")
 
@@ -1419,6 +1477,71 @@ def prepare_manifest(self, manifest_file: ManifestFile) -> ManifestFile:
         return wrapped_manifest_file
 
 
+class ManifestListWriterV3(ManifestListWriterV2):
+    """Writes V3 manifest lists, assigning `first_row_id` to data manifests.
+
+    Follows the spec's First Row ID Assignment: existing `first_row_id` values are
+    preserved, delete manifests are never assigned one, and data manifests without a
+    `first_row_id` are assigned a running value starting at the snapshot's
+    `first-row-id`, advanced by each assigned manifest's existing and added row counts.
+    """
+
+    _next_row_id: int
+
+    def __init__(
+        self,
+        output_file: OutputFile,
+        snapshot_id: int,
+        parent_snapshot_id: int | None,
+        sequence_number: int,
+        compression: AvroCompressionCodec,
+        first_row_id: int,
+    ):
+        super().__init__(output_file, snapshot_id, parent_snapshot_id, sequence_number, compression)
+        self._format_version = 3
+        self._meta = {
+            **self._meta,
+            "first-row-id": str(first_row_id),
+            "format-version": "3",
+        }
+        self._next_row_id = first_row_id
+
+    def __enter__(self) -> ManifestListWriter:
+        """Open the writer for writing, using the V3 record layout so `first_row_id` is written."""
+        self._writer = AvroOutputFile[ManifestFile](
+            output_file=self._output_file,
+            record_schema=MANIFEST_LIST_FILE_SCHEMAS[3],
+            file_schema=MANIFEST_LIST_FILE_SCHEMAS[3],
+            schema_name="manifest_file",
+            metadata=self._meta,
+        )
+        self._writer.__enter__()
+        return self
+
+    @property
+    def next_row_id(self) -> int:
+        """The row ID after the last assigned one; the table's `next-row-id` after this snapshot."""
+        return self._next_row_id
+
+    def _wrap(self, manifest_file: ManifestFile) -> ManifestFile:
+        """Rebind a manifest file to the V3 record layout."""
+        if len(manifest_file._data) >= len(MANIFEST_LIST_FILE_SCHEMAS[3].fields):
+            return copy(manifest_file)
+        args = {
+            field.name: value
+            for field, value in zip(MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION].fields, manifest_file._data, strict=True)
+        }
+        return ManifestFile.from_args(_table_format_version=3, **args)
+
+    def prepare_manifest(self, manifest_file: ManifestFile) -> ManifestFile:
+        wrapped_manifest_file = super().prepare_manifest(self._wrap(manifest_file))
+
+        if wrapped_manifest_file.content == ManifestContent.DATA and wrapped_manifest_file.first_row_id is None:
+            wrapped_manifest_file.first_row_id = self._next_row_id
+            self._next_row_id += (wrapped_manifest_file.existing_rows_count or 0) + (wrapped_manifest_file.added_rows_count or 0)
+        return wrapped_manifest_file
+
+
 def write_manifest_list(
     format_version: TableVersion,
     output_file: OutputFile,
@@ -1426,6 +1549,7 @@ def write_manifest_list(
     parent_snapshot_id: int | None,
     sequence_number: int | None,
     avro_compression: AvroCompressionCodec,
+    first_row_id: int | None = None,
 ) -> ManifestListWriter:
     if format_version == 1:
         return ManifestListWriterV1(output_file, snapshot_id, parent_snapshot_id, avro_compression)
@@ -1433,5 +1557,11 @@ def write_manifest_list(
         if sequence_number is None:
             raise ValueError(f"Sequence-number is required for V2 tables: {sequence_number}")
         return ManifestListWriterV2(output_file, snapshot_id, parent_snapshot_id, sequence_number, avro_compression)
+    elif format_version == 3:
+        if sequence_number is None:
+            raise ValueError(f"Sequence-number is required for V3 tables: {sequence_number}")
+        if first_row_id is None:
+            raise ValueError(f"First-row-id is required for V3 tables: {first_row_id}")
+        return ManifestListWriterV3(output_file, snapshot_id, parent_snapshot_id, sequence_number, avro_compression, first_row_id)
     else:
         raise ValueError(f"Cannot write manifest list for table version: {format_version}")
diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py
index f2ae1e05ad..b3e37cf8aa 100644
--- a/tests/utils/test_manifest.py
+++ b/tests/utils/test_manifest.py
@@ -1146,3 +1146,132 @@ def test_negative_manifest_cache_size_raises_value_error(monkeypatch: pytest.Mon
     finally:
         monkeypatch.delenv("PYICEBERG_MANIFEST_CACHE_SIZE", raising=False)
         importlib.reload(manifest_module)
+
+
+@pytest.mark.parametrize("compression", ["null", "deflate"])
+def test_write_manifest_v3(compression: AvroCompressionCodec) -> None:
+    io = load_file_io()
+    test_schema = Schema(NestedField(1, "foo", IntegerType(), False))
+
+    v3_data_file = DataFile.from_args(
+        _table_format_version=3,
+        content=DataFileContent.DATA,
+        file_path="/data/file-v3.parquet",
+        file_format=FileFormat.PARQUET,
+        partition=Record(),
+        record_count=100,
+        file_size_in_bytes=1024,
+        first_row_id=1000,
+    )
+    v2_data_file = DataFile.from_args(
+        content=DataFileContent.DATA,
+        file_path="/data/file-v2.parquet",
+        file_format=FileFormat.PARQUET,
+        partition=Record(),
+        record_count=50,
+        file_size_in_bytes=512,
+    )
+
+    with TemporaryDirectory() as tmp_dir:
+        path = tmp_dir + "/manifest-v3.avro"
+        with write_manifest(
+            format_version=3,
+            spec=UNPARTITIONED_PARTITION_SPEC,
+            schema=test_schema,
+            output_file=io.new_output(path),
+            snapshot_id=25,
+            avro_compression=compression,
+        ) as writer:
+            writer.add(ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, snapshot_id=25, data_file=v3_data_file))
+            # a data file bound to the default (V2) layout is rebound to the V3 layout
+            writer.add(ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, snapshot_id=25, data_file=v2_data_file))
+
+        _verify_metadata_with_fastavro(path, {"format-version": "3", "content": "data"})
+
+        with open(path, "rb") as f:
+            entries = list(fastavro.reader(f))
+
+        assert len(entries) == 2
+        assert entries[0]["data_file"]["first_row_id"] == 1000
+        assert entries[1]["data_file"]["first_row_id"] is None
+        for entry in entries:
+            for v3_field in ("first_row_id", "referenced_data_file", "content_offset", "content_size_in_bytes"):
+                assert v3_field in entry["data_file"]
+
+
+@pytest.mark.parametrize("compression", ["null", "deflate"])
+def test_write_manifest_list_v3_assigns_first_row_id(compression: AvroCompressionCodec) -> None:
+    io = load_file_io()
+
+    def manifest(path: str, content: ManifestContent, first_row_id: int | None = None, **counts: int) -> ManifestFile:
+        args: dict[str, Any] = {
+            "manifest_path": path,
+            "manifest_length": 100,
+            "partition_spec_id": 0,
+            "content": content,
+            "sequence_number": 1,
+            "min_sequence_number": 1,
+            "added_snapshot_id": 25,
+            "added_files_count": 1,
+            "existing_files_count": 1,
+            "deleted_files_count": 0,
+            "added_rows_count": counts.get("added", 0),
+            "existing_rows_count": counts.get("existing", 0),
+            "deleted_rows_count": 0,
+        }
+        if first_row_id is not None:
+            return ManifestFile.from_args(_table_format_version=3, first_row_id=first_row_id, **args)
+        # bound to the default (V2) layout to exercise rebinding in the writer
+        return ManifestFile.from_args(**args)
+
+    unassigned_data = manifest("/m1.avro", ManifestContent.DATA, added=100, existing=25)
+    preserved_data = manifest("/m2.avro", ManifestContent.DATA, first_row_id=77, added=10)
+    deletes = manifest("/m3.avro", ManifestContent.DELETES, added=10)
+    second_unassigned_data = manifest("/m4.avro", ManifestContent.DATA, added=5)
+
+    with TemporaryDirectory() as tmp_dir:
+        path = tmp_dir + "/manifest-list-v3.avro"
+        with write_manifest_list(
+            format_version=3,
+            output_file=io.new_output(path),
+            snapshot_id=25,
+            parent_snapshot_id=19,
+            sequence_number=2,
+            avro_compression=compression,
+            first_row_id=1000,
+        ) as writer:
+            writer.add_manifests([unassigned_data, preserved_data, deletes, second_unassigned_data])
+
+        # 1000 + (100 + 25) + (5): assigned manifests advance by added + existing rows
+        assert writer.next_row_id == 1130  # type: ignore[attr-defined]
+
+        _verify_metadata_with_fastavro(
+            path,
+            {
+                "snapshot-id": "25",
+                "parent-snapshot-id": "19",
+                "sequence-number": "2",
+                "first-row-id": "1000",
+                "format-version": "3",
+            },
+        )
+
+        with open(path, "rb") as f:
+            records = list(fastavro.reader(f))
+
+        assert [r["first_row_id"] for r in records] == [1000, 77, None, 1125]
+
+
+def test_write_manifest_list_v3_requires_first_row_id() -> None:
+    io = load_file_io()
+    with TemporaryDirectory() as tmp_dir:
+        with pytest.raises(ValueError, match="First-row-id is required for V3 tables"):
+            write_manifest_list(
+                format_version=3,
+                output_file=io.new_output(tmp_dir + "/manifest-list.avro"),
+                snapshot_id=25,
+                parent_snapshot_id=19,
+                sequence_number=2,
+                avro_compression="null",
+                first_row_id=None,
+            )

From a340d930a612b29fd44542a4cd45f542d256835f Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Wed, 8 Jul 2026 21:13:38 +0900
Subject: [PATCH 2/6] Align V3 manifest writers with the Java implementation

Carry an optional first_row_id through ManifestWriterV3 into the
produced manifest file (used when rewriting manifests whose
first_row_id is known), matching ManifestFiles.newWriter in Java.
Expand tests for parity with TestManifestWriterVersions and
TestManifestListVersions: reader compatibility of V3-written files,
metrics field round-trips, null first_row_id when reading V2 manifest
lists with the V3 layout, and preservation of writer-carried
first_row_id without advancing next_row_id.
---
 pyiceberg/manifest.py        |  33 ++++++++-
 tests/utils/test_manifest.py | 134 +++++++++++++++++++++++++++++++++++
 2 files changed, 166 insertions(+), 1 deletion(-)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index 04c6b29b59..fada97f0c5 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -1301,12 +1301,40 @@ class ManifestWriterV3(ManifestWriterV2):
     The writer inherits the V2 sequence-number semantics; the V3 manifest entry
     schema additionally carries `first_row_id`, `referenced_data_file`,
     `content_offset` and `content_size_in_bytes` on the data file struct.
+
+    An optional `first_row_id` can be provided when rewriting a manifest whose
+    `first_row_id` is already known; it is carried into the produced manifest file
+    so the manifest list writer preserves it instead of assigning a new one. For
+    new manifests it is None and assigned when writing the manifest list.
     """
 
+    _first_row_id: int | None
+
+    def __init__(
+        self,
+        spec: PartitionSpec,
+        schema: Schema,
+        output_file: OutputFile,
+        snapshot_id: int,
+        avro_compression: AvroCompressionCodec,
+        first_row_id: int | None = None,
+    ):
+        super().__init__(spec, schema, output_file, snapshot_id, avro_compression)
+        self._first_row_id = first_row_id
+
     @property
     def version(self) -> TableVersion:
         return 3
 
+    def to_manifest_file(self) -> ManifestFile:
+        """Return the manifest file, bound to the V3 layout and carrying `first_row_id`."""
+        manifest_file = super().to_manifest_file()
+        args = {
+            field.name: value
+            for field, value in zip(MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION].fields, manifest_file._data, strict=True)
+        }
+        return ManifestFile.from_args(_table_format_version=3, first_row_id=self._first_row_id, **args)
+
     def new_writer(self) -> AvroOutputFile[ManifestEntry]:
         # Use the V3 record layout so the V3-only data file fields are written
         return AvroOutputFile[ManifestEntry](
@@ -1347,13 +1375,16 @@ def write_manifest(
     output_file: OutputFile,
     snapshot_id: int,
     avro_compression: AvroCompressionCodec,
+    first_row_id: int | None = None,
 ) -> ManifestWriter:
+    if first_row_id is not None and format_version < 3:
+        raise ValueError(f"First-row-id is only supported for V3 tables: {format_version}")
     if format_version == 1:
         return ManifestWriterV1(spec, schema, output_file, snapshot_id, avro_compression)
     elif format_version == 2:
         return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression)
     elif format_version == 3:
-        return ManifestWriterV3(spec, schema, output_file, snapshot_id, avro_compression)
+        return ManifestWriterV3(spec, schema, output_file, snapshot_id, avro_compression, first_row_id)
     else:
         raise ValueError(f"Cannot write manifest for table version: {format_version}")
 
diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py
index b3e37cf8aa..bbebe0882c 100644
--- a/tests/utils/test_manifest.py
+++ b/tests/utils/test_manifest.py
@@ -1161,6 +1161,11 @@ def test_write_manifest_v3(compression: AvroCompressionCodec) -> None:
         partition=Record(),
         record_count=100,
         file_size_in_bytes=1024,
+        column_sizes={1: 10},
+        value_counts={1: 100},
+        null_value_counts={1: 0},
+        split_offsets=[4],
+        sort_order_id=1,
         first_row_id=1000,
     )
     v2_data_file = DataFile.from_args(
@@ -1197,6 +1202,19 @@ def test_write_manifest_v3(compression: AvroCompressionCodec) -> None:
         for entry in entries:
             for v3_field in ("first_row_id", "referenced_data_file", "content_offset", "content_size_in_bytes"):
                 assert v3_field in entry["data_file"]
+        assert entries[0]["data_file"]["sort_order_id"] == 1
+        assert entries[0]["data_file"]["split_offsets"] == [4]
+
+        # the V3 manifest must remain readable by the current reader
+        read_entries = writer.to_manifest_file().fetch_manifest_entry(io, discard_deleted=False)
+        assert len(read_entries) == 2
+        assert read_entries[0].status == ManifestEntryStatus.ADDED
+        assert read_entries[0].data_file.file_path == "/data/file-v3.parquet"
+        assert read_entries[0].data_file.record_count == 100
+        assert read_entries[0].data_file.column_sizes == {1: 10}
+        assert read_entries[0].data_file.value_counts == {1: 100}
+        assert read_entries[0].data_file.sort_order_id == 1
+        assert read_entries[1].data_file.file_path == "/data/file-v2.parquet"
 
 
 @pytest.mark.parametrize("compression", ["null", "deflate"])
@@ -1261,6 +1279,13 @@ def manifest(path: str, content: ManifestContent, first_row_id: int | None = Non
 
         assert [r["first_row_id"] for r in records] == [1000, 77, None, 1125]
 
+        # the V3 manifest list must remain readable by the current reader
+        read_back = list(read_manifest_list(io.new_input(path)))
+        assert [m.manifest_path for m in read_back] == ["/m1.avro", "/m2.avro", "/m3.avro", "/m4.avro"]
+        assert read_back[0].added_rows_count == 100
+        assert read_back[0].existing_rows_count == 25
+        assert read_back[2].content == ManifestContent.DELETES
+
 
 def test_write_manifest_list_v3_requires_first_row_id() -> None:
     io = load_file_io()
@@ -1275,3 +1300,112 @@ def test_write_manifest_list_v3_requires_first_row_id() -> None:
                 avro_compression="null",
                 first_row_id=None,
             )
+
+
+@pytest.mark.parametrize("compression", ["null", "deflate"])
+def test_write_manifest_v3_carries_first_row_id(compression: AvroCompressionCodec) -> None:
+    io = load_file_io()
+    test_schema = Schema(NestedField(1, "foo", IntegerType(), False))
+    data_file = DataFile.from_args(
+        content=DataFileContent.DATA,
+        file_path="/data/file.parquet",
+        file_format=FileFormat.PARQUET,
+        partition=Record(),
+        record_count=100,
+        file_size_in_bytes=1024,
+    )
+
+    with TemporaryDirectory() as tmp_dir:
+        with write_manifest(
+            format_version=3,
+            spec=UNPARTITIONED_PARTITION_SPEC,
+            schema=test_schema,
+            output_file=io.new_output(tmp_dir + "/manifest-rewrite.avro"),
+            snapshot_id=25,
+            avro_compression=compression,
+            first_row_id=100,
+        ) as writer:
+            writer.add(ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, snapshot_id=25, data_file=data_file))
+
+        manifest_file = writer.to_manifest_file()
+        assert manifest_file.first_row_id == 100
+
+        # a manifest list writer must preserve the carried first_row_id and not advance next_row_id for it
+        path = tmp_dir + "/manifest-list.avro"
+        with write_manifest_list(
+            format_version=3,
+            output_file=io.new_output(path),
+            snapshot_id=25,
+            parent_snapshot_id=19,
+            sequence_number=2,
+            avro_compression=compression,
+            first_row_id=1000,
+        ) as list_writer:
+            list_writer.add_manifests([manifest_file])
+        assert list_writer.next_row_id == 1000  # type: ignore[attr-defined]
+
+        with open(path, "rb") as f:
+            records = list(fastavro.reader(f))
+        assert [r["first_row_id"] for r in records] == [100]
+
+
+def test_write_manifest_first_row_id_requires_v3() -> None:
+    io = load_file_io()
+    test_schema = Schema(NestedField(1, "foo", IntegerType(), False))
+    with TemporaryDirectory() as tmp_dir:
+        with pytest.raises(ValueError, match="First-row-id is only supported for V3 tables"):
+            write_manifest(
+                format_version=2,
+                spec=UNPARTITIONED_PARTITION_SPEC,
+                schema=test_schema,
+                output_file=io.new_output(tmp_dir + "/manifest.avro"),
+                snapshot_id=25,
+                avro_compression="null",
+                first_row_id=100,
+            )
+
+
+def test_read_v2_manifest_list_with_v3_layout() -> None:
+    from pyiceberg.avro.file import AvroFile
+    from pyiceberg.manifest import MANIFEST_LIST_FILE_SCHEMAS
+
+    io = load_file_io()
+    manifest = ManifestFile.from_args(
+        manifest_path="/m1.avro",
+        manifest_length=100,
+        partition_spec_id=0,
+        content=ManifestContent.DATA,
+        sequence_number=1,
+        min_sequence_number=1,
+        added_snapshot_id=25,
+        added_files_count=1,
+        existing_files_count=0,
+        deleted_files_count=0,
+        added_rows_count=100,
+        existing_rows_count=0,
+        deleted_rows_count=0,
+    )
+
+    with TemporaryDirectory() as tmp_dir:
+        path = tmp_dir + "/manifest-list-v2.avro"
+        with write_manifest_list(
+            format_version=2,
+            output_file=io.new_output(path),
+            snapshot_id=25,
+            parent_snapshot_id=19,
+            sequence_number=2,
+            avro_compression="null",
+        ) as writer:
+            writer.add_manifests([manifest])
+
+        # reading a V2 manifest list with the V3 layout yields a null first_row_id
+        with AvroFile[ManifestFile](
+            io.new_input(path),
+            MANIFEST_LIST_FILE_SCHEMAS[3],
+            read_types={-1: ManifestFile},
+            read_enums={517: ManifestContent},
+        ) as reader:
+            entries = list(reader)
+        assert len(entries) == 1
+        assert entries[0].first_row_id is None
+        assert entries[0].manifest_path == "/m1.avro"

From 75f96f415a0122809890c5c79f1b791aee02f1d7 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Thu, 9 Jul 2026 09:07:22 +0900
Subject: [PATCH 3/6] Address review: reject unknown row counts in first-row-id
 assignment

Assigning a first-row-id while treating unknown existing/added row
counts as zero would let subsequent manifests overlap row ID ranges.
Raise a clear error instead, matching the reference implementation
which requires the counts. Also drop the always-None value from the
first-row-id-required error message and document the record layout
invariant behind the V3 rebinding.
---
 pyiceberg/manifest.py        | 16 +++++++++++++---
 tests/utils/test_manifest.py | 28 ++++++++++++++++++++++++++++
 2 files changed, 41 insertions(+), 3 deletions(-)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index fada97f0c5..c5f3c21c52 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -1555,7 +1555,12 @@ def next_row_id(self) -> int:
         return self._next_row_id
 
     def _wrap(self, manifest_file: ManifestFile) -> ManifestFile:
-        """Rebind a manifest file to the V3 record layout."""
+        """Rebind a manifest file to the V3 record layout.
+
+        Records not created with an explicit layout are bound to
+        MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION] (the from_args default),
+        so that is the layout to zip the positional data against.
+        """
         if len(manifest_file._data) >= len(MANIFEST_LIST_FILE_SCHEMAS[3].fields):
             return copy(manifest_file)
         args = {
@@ -1568,8 +1573,13 @@ def prepare_manifest(self, manifest_file: ManifestFile) -> ManifestFile:
         wrapped_manifest_file = super().prepare_manifest(self._wrap(manifest_file))
 
         if wrapped_manifest_file.content == ManifestContent.DATA and wrapped_manifest_file.first_row_id is None:
+            if wrapped_manifest_file.existing_rows_count is None or wrapped_manifest_file.added_rows_count is None:
+                # assigning first-row-id with unknown row counts would overlap row ID ranges
+                raise ValueError(
+                    f"Cannot assign first-row-id to manifest with unknown row counts: {wrapped_manifest_file.manifest_path}"
+                )
             wrapped_manifest_file.first_row_id = self._next_row_id
-            self._next_row_id += (wrapped_manifest_file.existing_rows_count or 0) + (wrapped_manifest_file.added_rows_count or 0)
+            self._next_row_id += wrapped_manifest_file.existing_rows_count + wrapped_manifest_file.added_rows_count
         return wrapped_manifest_file
 
 
@@ -1592,7 +1602,7 @@ def write_manifest_list(
         if sequence_number is None:
             raise ValueError(f"Sequence-number is required for V3 tables: {sequence_number}")
         if first_row_id is None:
-            raise ValueError(f"First-row-id is required for V3 tables: {first_row_id}")
+            raise ValueError("First-row-id is required for V3 tables")
         return ManifestListWriterV3(output_file, snapshot_id, parent_snapshot_id, sequence_number, avro_compression, first_row_id)
     else:
         raise ValueError(f"Cannot write manifest list for table version: {format_version}")
diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py
index bbebe0882c..a000c9d23e 100644
--- a/tests/utils/test_manifest.py
+++ b/tests/utils/test_manifest.py
@@ -1409,3 +1409,31 @@ def test_read_v2_manifest_list_with_v3_layout() -> None:
         assert len(entries) == 1
         assert entries[0].first_row_id is None
         assert entries[0].manifest_path == "/m1.avro"
+
+
+def test_write_manifest_list_v3_rejects_unknown_row_counts() -> None:
+    io = load_file_io()
+    manifest = ManifestFile.from_args(
+        manifest_path="/m1.avro",
+        manifest_length=100,
+        partition_spec_id=0,
+        content=ManifestContent.DATA,
+        sequence_number=1,
+        min_sequence_number=1,
+        added_snapshot_id=25,
+        added_rows_count=None,
+        existing_rows_count=None,
+    )
+
+    with TemporaryDirectory() as tmp_dir:
+        with pytest.raises(ValueError, match="unknown row counts"):
+            with write_manifest_list(
+                format_version=3,
+                output_file=io.new_output(tmp_dir + "/manifest-list.avro"),
+                snapshot_id=25,
+                parent_snapshot_id=19,
+                sequence_number=2,
+                avro_compression="null",
+                first_row_id=1000,
+            ) as writer:
+                writer.add_manifests([manifest])

From de695ba099776ebaea395386636962df7db62d58 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Wed, 22 Jul 2026 14:42:18 +0900
Subject: [PATCH 4/6] Address review: preserve V3-only fields when reading
 manifests

Reading V2/V3 manifests and manifest lists used a fixed V2 read schema,
so V3-only fields (first_row_id, referenced_data_file, content_offset,
content_size_in_bytes) were silently dropped by Avro schema resolution,
breaking V3 round trips. Read with the latest known schema instead,
keeping the V2 default for constructing and writing records unchanged.

Also fixes ManifestListWriterV3._wrap sharing _data with the source
record via a shallow copy, and rebinding a record to the V3 layout
against a fixed V2 field count, which raised on V1-bound records.
---
 pyiceberg/manifest.py        |  74 ++++++++++++++++----
 tests/avro/test_file.py      |  71 ++++++++++++++++++-
 tests/utils/test_manifest.py | 130 +++++++++++++++++++++++++++++++++++
 3 files changed, 262 insertions(+), 13 deletions(-)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index c5f3c21c52..348431296d 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -19,7 +19,7 @@
 import math
 import threading
 from abc import ABC, abstractmethod
-from collections.abc import Iterator
+from collections.abc import Iterator, Mapping
 from copy import copy
 from enum import Enum
 from types import TracebackType
@@ -532,6 +532,26 @@ def equality_ids(self) -> list[int] | None:
     def sort_order_id(self) -> int | None:
         return self._data[15]
 
+    @property
+    def first_row_id(self) -> int | None:
+        # V3-only field; absent on records bound to an older layout
+        return self._data[16] if len(self._data) > 16 else None
+
+    @property
+    def referenced_data_file(self) -> str | None:
+        # V3-only field; absent on records bound to an older layout
+        return self._data[17] if len(self._data) > 17 else None
+
+    @property
+    def content_offset(self) -> int | None:
+        # V3-only field; absent on records bound to an older layout
+        return self._data[18] if len(self._data) > 18 else None
+
+    @property
+    def content_size_in_bytes(self) -> int | None:
+        # V3-only field; absent on records bound to an older layout
+        return self._data[19] if len(self._data) > 19 else None
+
     # Spec ID should not be stored in the file
     _spec_id: int
 
@@ -774,6 +794,37 @@ def construct_partition_summaries(spec: PartitionSpec, schema: Schema, partition
 
 MANIFEST_LIST_FILE_STRUCTS = {format_version: schema.as_struct() for format_version, schema in MANIFEST_LIST_FILE_SCHEMAS.items()}
 
+# Manifests and manifest lists are read with the schema of the latest known format version so that
+# fields added in newer versions (e.g. `first_row_id`) survive a read/write round trip. Reading an
+# older file with this schema is safe: the added fields are all optional and are filled with their
+# defaults (None) by Avro schema resolution. Deriving these from the schema dicts keeps them in sync
+# as new format versions are added. Manifest entries and manifest lists are versioned by separate
+# schema dicts, so each has its own latest-version constant rather than sharing one. Note these are
+# intentionally separate from DEFAULT_READ_VERSION, which remains the default layout for constructing
+# and writing records.
+LATEST_MANIFEST_ENTRY_READ_VERSION: TableVersion = max(MANIFEST_ENTRY_SCHEMAS)
+LATEST_MANIFEST_LIST_READ_VERSION: TableVersion = max(MANIFEST_LIST_FILE_SCHEMAS)
+
+
+def _layout_version_from_field_count(layouts: Mapping[int, Schema | StructType], field_count: int) -> TableVersion:
+    """Return the format version whose layout has exactly `field_count` fields.
+
+    A positional record (`DataFile`, `ManifestFile`, ...) does not carry the format version it was
+    bound to, but each version's layout has a distinct number of fields, so the field count uniquely
+    identifies it. This is used when rebinding a record to a newer layout to zip its `_data` against
+    the fields of the version it was originally bound to, rather than assuming a fixed version.
+
+    Identifying the version by field count is only valid while each version's layout has a distinct
+    field count, so this fails loudly if two versions share one rather than silently binding to the
+    wrong layout.
+    """
+    matches = [version for version, layout in layouts.items() if len(layout.fields) == field_count]
+    if len(matches) > 1:
+        raise ValueError(f"Ambiguous layout: versions {matches} all have {field_count} fields")
+    if not matches:
+        raise ValueError(f"Cannot determine layout version for record with {field_count} fields")
+    return matches[0]
+
 
 POSITIONAL_DELETE_SCHEMA = Schema(
     NestedField(2147483546, "file_path", StringType()), NestedField(2147483545, "pos", IntegerType())
@@ -884,7 +935,7 @@ def fetch_manifest_entry(self, io: FileIO, discard_deleted: bool = True) -> list
         input_file = io.new_input(self.manifest_path)
         with AvroFile[ManifestEntry](
             input_file,
-            MANIFEST_ENTRY_SCHEMAS[DEFAULT_READ_VERSION],
+            MANIFEST_ENTRY_SCHEMAS[LATEST_MANIFEST_ENTRY_READ_VERSION],
             read_types={-1: ManifestEntry, 2: DataFile},
             read_enums={0: ManifestEntryStatus, 101: FileFormat, 134: DataFileContent},
         ) as reader:
@@ -1007,7 +1058,7 @@ def read_manifest_list(input_file: InputFile) -> Iterator[ManifestFile]:
     """
     with AvroFile[ManifestFile](
         input_file,
-        MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION],
+        MANIFEST_LIST_FILE_SCHEMAS[LATEST_MANIFEST_LIST_READ_VERSION],
         read_types={-1: ManifestFile, 508: PartitionFieldSummary},
         read_enums={517: ManifestContent},
     ) as reader:
@@ -1349,9 +1400,8 @@ def _wrap_data_file(self, data_file: DataFile) -> DataFile:
         """Rebind a data file to the V3 record layout."""
         if len(data_file._data) >= len(DATA_FILE_TYPE[3].fields):
             return data_file
-        args = {
-            field.name: value for field, value in zip(DATA_FILE_TYPE[DEFAULT_READ_VERSION].fields, data_file._data, strict=True)
-        }
+        source_version = _layout_version_from_field_count(DATA_FILE_TYPE, len(data_file._data))
+        args = {field.name: value for field, value in zip(DATA_FILE_TYPE[source_version].fields, data_file._data, strict=True)}
         return DataFile.from_args(_table_format_version=3, **args)
 
     def prepare_entry(self, entry: ManifestEntry) -> ManifestEntry:
@@ -1557,15 +1607,15 @@ def next_row_id(self) -> int:
     def _wrap(self, manifest_file: ManifestFile) -> ManifestFile:
         """Rebind a manifest file to the V3 record layout.
 
-        Records not created with an explicit layout are bound to
-        MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION] (the from_args default),
-        so that is the layout to zip the positional data against.
+        The manifest file's original layout version is recovered from its field count (each version's
+        layout has a distinct number of fields) so its positional data is zipped against the fields it
+        was actually bound to. Rebinding always produces a fresh record, so mutating the returned
+        manifest file (e.g. assigning `first_row_id`) never leaks back to the caller's record.
         """
-        if len(manifest_file._data) >= len(MANIFEST_LIST_FILE_SCHEMAS[3].fields):
-            return copy(manifest_file)
+        source_version = _layout_version_from_field_count(MANIFEST_LIST_FILE_SCHEMAS, len(manifest_file._data))
         args = {
             field.name: value
-            for field, value in zip(MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION].fields, manifest_file._data, strict=True)
+            for field, value in zip(MANIFEST_LIST_FILE_SCHEMAS[source_version].fields, manifest_file._data, strict=True)
         }
         return ManifestFile.from_args(_table_format_version=3, **args)
 
diff --git a/tests/avro/test_file.py b/tests/avro/test_file.py
index 137215ebc8..dfc167d68f 100644
--- a/tests/avro/test_file.py
+++ b/tests/avro/test_file.py
@@ -32,11 +32,14 @@
 from pyiceberg.manifest import (
     DEFAULT_BLOCK_SIZE,
     MANIFEST_ENTRY_SCHEMAS,
+    MANIFEST_LIST_FILE_SCHEMAS,
     DataFile,
     DataFileContent,
     FileFormat,
+    ManifestContent,
     ManifestEntry,
     ManifestEntryStatus,
+    ManifestFile,
 )
 from pyiceberg.schema import Schema
 from pyiceberg.typedef import Record, TableVersion
@@ -87,6 +90,23 @@ def test_missing_schema() -> None:
     assert "No schema found in Avro file headers" in str(exc_info.value)
 
 
+# DataFile and ManifestFile expose properties for fields added in newer format versions (e.g. the
+# V3-only first_row_id). A record bound to an older layout does not carry those fields in its
+# positional data, so they are excluded from the serialized dict to match the fields fastavro writes
+# for that layout, keyed by the _data length beyond which each field is absent.
+_VERSION_DEPENDENT_FIELDS_BY_MIN_DATA_LEN: dict[type[Record], dict[str, int]] = {
+    DataFile: {
+        "first_row_id": 17,
+        "referenced_data_file": 18,
+        "content_offset": 19,
+        "content_size_in_bytes": 20,
+    },
+    ManifestFile: {
+        "first_row_id": 16,
+    },
+}
+
+
 # helper function to serialize our objects to dicts to enable
 # direct comparison with the dicts returned by fastavro
 def todict(obj: Any) -> Any:
@@ -100,7 +120,16 @@ def todict(obj: Any) -> Any:
     elif hasattr(obj, "__iter__") and not isinstance(obj, str) and not isinstance(obj, bytes):
         return [todict(v) for v in obj]
     elif isinstance(obj, Record):
-        return {key: todict(value) for key, value in inspect.getmembers(obj) if not callable(value) and not key.startswith("_")}
+        min_data_len = next(
+            (fields for record_type, fields in _VERSION_DEPENDENT_FIELDS_BY_MIN_DATA_LEN.items() if isinstance(obj, record_type)),
+            {},
+        )
+        data_len = len(obj._data)
+        return {
+            key: todict(value)
+            for key, value in inspect.getmembers(obj)
+            if not callable(value) and not key.startswith("_") and data_len >= min_data_len.get(key, 0)
+        }
     else:
         return obj
 
@@ -225,6 +254,46 @@ def test_write_manifest_entry_with_iceberg_read_with_fastavro_v2() -> None:
         assert todict(entry) == fa_entry
 
 
+def test_todict_manifest_file_v2_layout_matches_fastavro() -> None:
+    """todict must exclude ManifestFile's V3-only first_row_id for a record bound to the V2 layout,
+
+    the same way it does for DataFile, otherwise the exclusion added for DataFile would be
+    inconsistent for other record types with version-dependent trailing fields.
+    """
+    manifest_file = ManifestFile.from_args(
+        manifest_path="/manifests/m1.avro",
+        manifest_length=100,
+        partition_spec_id=0,
+        content=ManifestContent.DATA,
+        sequence_number=1,
+        min_sequence_number=1,
+        added_snapshot_id=25,
+        added_files_count=1,
+        existing_files_count=0,
+        deleted_files_count=0,
+        added_rows_count=10,
+        existing_rows_count=0,
+        deleted_rows_count=0,
+        partitions=None,
+        key_metadata=None,
+    )
+
+    with TemporaryDirectory() as tmpdir:
+        tmp_avro_file = tmpdir + "/manifest-list.avro"
+
+        with avro.AvroOutputFile[ManifestFile](
+            output_file=PyArrowFileIO().new_output(tmp_avro_file),
+            file_schema=MANIFEST_LIST_FILE_SCHEMAS[2],
+            schema_name="manifest_file",
+        ) as out:
+            out.write_block([manifest_file])
+
+        with open(tmp_avro_file, "rb") as fo:
+            fa_manifest_file = next(iter(reader(fo=fo)))
+
+        assert todict(manifest_file) == fa_manifest_file
+
+
 @pytest.mark.parametrize("format_version", [1, 2])
 def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: TableVersion) -> None:
     data_file_dict = {
diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py
index a000c9d23e..fcb12207ef 100644
--- a/tests/utils/test_manifest.py
+++ b/tests/utils/test_manifest.py
@@ -37,6 +37,7 @@
     ManifestFile,
     PartitionFieldSummary,
     _inherit_from_manifest,
+    _layout_version_from_field_count,
     _manifests,
     clear_manifest_cache,
     read_manifest_list,
@@ -1215,6 +1216,114 @@ def test_write_manifest_v3(compression: AvroCompressionCodec) -> None:
         assert read_entries[0].data_file.value_counts == {1: 100}
         assert read_entries[0].data_file.sort_order_id == 1
         assert read_entries[1].data_file.file_path == "/data/file-v2.parquet"
+        # the reader must expose the V3-only data file fields rather than dropping them, otherwise a
+        # V3 manifest cannot round-trip through the reader
+        assert read_entries[0].data_file.first_row_id == 1000
+        assert read_entries[1].data_file.first_row_id is None
+        for read_entry in read_entries:
+            for v3_field in ("first_row_id", "referenced_data_file", "content_offset", "content_size_in_bytes"):
+                assert hasattr(read_entry.data_file, v3_field)
+
+
+def test_write_manifest_v3_round_trips_deletion_vector_fields() -> None:
+    """The V3-only referenced_data_file/content_offset/content_size_in_bytes fields must survive a
+    write/read round trip with their actual values, not just be present-but-empty on the reader.
+    """
+    io = load_file_io()
+    test_schema = Schema(NestedField(1, "foo", IntegerType(), False))
+
+    dv_data_file = DataFile.from_args(
+        _table_format_version=3,
+        content=DataFileContent.POSITION_DELETES,
+        file_path="/data/dv-1.puffin",
+        file_format=FileFormat.PUFFIN,
+        partition=Record(),
+        record_count=5,
+        file_size_in_bytes=256,
+        referenced_data_file="/data/file-v3.parquet",
+        content_offset=128,
+        content_size_in_bytes=64,
+    )
+
+    with TemporaryDirectory() as tmp_dir:
+        path = tmp_dir + "/manifest-v3-dv.avro"
+        with write_manifest(
+            format_version=3,
+            spec=UNPARTITIONED_PARTITION_SPEC,
+            schema=test_schema,
+            output_file=io.new_output(path),
+            snapshot_id=25,
+            avro_compression="null",
+        ) as writer:
+            writer.add(ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, snapshot_id=25, data_file=dv_data_file))
+
+        read_entries = writer.to_manifest_file().fetch_manifest_entry(io, discard_deleted=False)
+        assert len(read_entries) == 1
+        read_data_file = read_entries[0].data_file
+        assert read_data_file.referenced_data_file == "/data/file-v3.parquet"
+        assert read_data_file.content_offset == 128
+        assert read_data_file.content_size_in_bytes == 64
+
+
+def test_layout_version_from_field_count() -> None:
+    layouts = {
+        1: Schema(NestedField(1, "a", IntegerType(), False)),
+        2: Schema(NestedField(1, "a", IntegerType(), False), NestedField(2, "b", IntegerType(), False)),
+    }
+    assert _layout_version_from_field_count(layouts, 1) == 1
+    assert _layout_version_from_field_count(layouts, 2) == 2
+
+    with pytest.raises(ValueError, match="Cannot determine layout version"):
+        _layout_version_from_field_count(layouts, 3)
+
+
+def test_layout_version_from_field_count_rejects_ambiguous_layouts() -> None:
+    # two versions with the same field count cannot be told apart by field count alone
+    layouts = {
+        1: Schema(NestedField(1, "a", IntegerType(), False)),
+        2: Schema(NestedField(1, "a", IntegerType(), False)),
+    }
+    with pytest.raises(ValueError, match="Ambiguous layout"):
+        _layout_version_from_field_count(layouts, 1)
+
+
+def test_write_manifest_v3_rebinds_v1_data_file() -> None:
+    """A V1-bound data file (fewer fields than V2) must be rebindable to the V3 layout.
+
+    The re-wrap logic recovers the source layout from the field count, so a V1 data file is zipped
+    against the V1 fields rather than a fixed V2 layout, which would raise a confusing strict-zip
+    length mismatch.
+    """
+    io = load_file_io()
+    test_schema = Schema(NestedField(1, "foo", IntegerType(), False))
+
+    v1_data_file = DataFile.from_args(
+        _table_format_version=1,
+        file_path="/data/file-v1.parquet",
+        file_format=FileFormat.PARQUET,
+        partition=Record(),
+        record_count=10,
+        file_size_in_bytes=128,
+    )
+
+    with TemporaryDirectory() as tmp_dir:
+        path = tmp_dir + "/manifest-v3-from-v1.avro"
+        with write_manifest(
+            format_version=3,
+            spec=UNPARTITIONED_PARTITION_SPEC,
+            schema=test_schema,
+            output_file=io.new_output(path),
+            snapshot_id=25,
+            avro_compression="null",
+        ) as writer:
+            writer.add(ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, snapshot_id=25, data_file=v1_data_file))
+
+        read_entries = writer.to_manifest_file().fetch_manifest_entry(io, discard_deleted=False)
+        assert len(read_entries) == 1
+        assert read_entries[0].data_file.file_path == "/data/file-v1.parquet"
+        assert read_entries[0].data_file.record_count == 10
+        # V3-only field is filled with its default when rebinding a V1 data file
+        assert read_entries[0].data_file.first_row_id is None
 
 
 @pytest.mark.parametrize("compression", ["null", "deflate"])
@@ -1285,6 +1394,27 @@ def manifest(path: str, content: ManifestContent, first_row_id: int | None = Non
         assert read_back[0].added_rows_count == 100
         assert read_back[0].existing_rows_count == 25
         assert read_back[2].content == ManifestContent.DELETES
+        # the reader must expose the assigned first_row_id values rather than dropping them, otherwise a
+        # V3 manifest list cannot round-trip through the reader
+        assert [m.first_row_id for m in read_back] == [1000, 77, None, 1125]
+
+        # re-writing the manifests read back from the V3 list must preserve their already-assigned
+        # first_row_id values instead of reassigning them
+        rewritten_path = tmp_dir + "/manifest-list-v3-rewritten.avro"
+        with write_manifest_list(
+            format_version=3,
+            output_file=io.new_output(rewritten_path),
+            snapshot_id=25,
+            parent_snapshot_id=19,
+            sequence_number=2,
+            avro_compression=compression,
+            first_row_id=2000,
+        ) as rewriter:
+            rewriter.add_manifests(read_back)
+
+        rewritten = list(read_manifest_list(io.new_input(rewritten_path)))
+        # data manifests keep their prior first_row_id; the delete manifest stays None
+        assert [m.first_row_id for m in rewritten] == [1000, 77, None, 1125]
 
 
 def test_write_manifest_list_v3_requires_first_row_id() -> None:

From 8d1c4eb737e6a627edcbb84e5438a2acb0ab44c3 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Wed, 22 Jul 2026 23:06:56 +0900
Subject: [PATCH 5/6] Fix mypy: cast dict-key lookups back to TableVersion

max() and list indexing on the version-keyed schema dicts widen the
Literal[1, 2, 3] TableVersion back to int, so mypy flags the results
as incompatible with the declared TableVersion return types.
---
 pyiceberg/manifest.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index 348431296d..04aa134fe8 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -26,6 +26,7 @@
 from typing import (
     Any,
     Literal,
+    cast,
 )
 
 from cachetools import LRUCache
@@ -802,8 +803,8 @@ def construct_partition_summaries(spec: PartitionSpec, schema: Schema, partition
 # schema dicts, so each has its own latest-version constant rather than sharing one. Note these are
 # intentionally separate from DEFAULT_READ_VERSION, which remains the default layout for constructing
 # and writing records.
-LATEST_MANIFEST_ENTRY_READ_VERSION: TableVersion = max(MANIFEST_ENTRY_SCHEMAS)
-LATEST_MANIFEST_LIST_READ_VERSION: TableVersion = max(MANIFEST_LIST_FILE_SCHEMAS)
+LATEST_MANIFEST_ENTRY_READ_VERSION: TableVersion = cast(TableVersion, max(MANIFEST_ENTRY_SCHEMAS))
+LATEST_MANIFEST_LIST_READ_VERSION: TableVersion = cast(TableVersion, max(MANIFEST_LIST_FILE_SCHEMAS))
 
 
 def _layout_version_from_field_count(layouts: Mapping[int, Schema | StructType], field_count: int) -> TableVersion:
@@ -823,7 +824,7 @@ def _layout_version_from_field_count(layouts: Mapping[int, Schema | StructType],
         raise ValueError(f"Ambiguous layout: versions {matches} all have {field_count} fields")
     if not matches:
         raise ValueError(f"Cannot determine layout version for record with {field_count} fields")
-    return matches[0]
+    return cast(TableVersion, matches[0])
 
 
 POSITIONAL_DELETE_SCHEMA = Schema(

From d8b4e4695ad9c54a430d7c267b3299240838ae4e Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Wed, 22 Jul 2026 23:59:35 +0900
Subject: [PATCH 6/6] Fix integration test todict for DataFile's V3-only
 properties

test_rest_manifest.py has its own todict helper (separate from
tests/avro/test_file.py's), which also enumerates every DataFile
property via reflection. Adding V3-only properties to DataFile in
the previous commit caused this helper to include them for a
V2-shaped record too, diverging from the V2 dict fastavro writes and
failing the REST integration round-trip test.
---
 tests/integration/test_rest_manifest.py | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/tests/integration/test_rest_manifest.py b/tests/integration/test_rest_manifest.py
index 6c2bf7baed..ce8f6926c1 100644
--- a/tests/integration/test_rest_manifest.py
+++ b/tests/integration/test_rest_manifest.py
@@ -33,6 +33,17 @@
 from pyiceberg.typedef import Record
 from pyiceberg.utils.lazydict import LazyDict
 
+# DataFile exposes properties for fields added in newer format versions (e.g. the V3-only
+# first_row_id). A record bound to an older layout does not carry those fields in its positional
+# data, so they are excluded from the serialized dict to match the fields fastavro writes for that
+# layout, keyed by the _data length beyond which each field is absent.
+_DATA_FILE_FIELDS_BY_MIN_DATA_LEN = {
+    "first_row_id": 17,
+    "referenced_data_file": 18,
+    "content_offset": 19,
+    "content_size_in_bytes": 20,
+}
+
 
 # helper function to serialize our objects to dicts to enable
 # direct comparison with the dicts returned by fastavro
@@ -49,10 +60,12 @@ def todict(obj: Any, spec_keys: list[str]) -> Any:
     elif hasattr(obj, "__iter__") and not isinstance(obj, str) and not isinstance(obj, bytes):
         return [todict(v, spec_keys) for v in obj]
     elif hasattr(obj, "__dict__"):
+        min_data_len = _DATA_FILE_FIELDS_BY_MIN_DATA_LEN if isinstance(obj, DataFile) else {}
+        data_len = len(obj._data)
         return {
             key: todict(value, spec_keys)
             for key, value in inspect.getmembers(obj)
-            if not callable(value) and not key.startswith("_")
+            if not callable(value) and not key.startswith("_") and data_len >= min_data_len.get(key, 0)
         }
     else:
         return obj