From 23b426bd55cd670fcf068bbd7d11152557d6df32 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/9] 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        | 128 +++++++++++++++++++++++++++++++++-
 tests/utils/test_manifest.py | 129 +++++++++++++++++++++++++++++++++++
 2 files changed, 256 insertions(+), 1 deletion(-)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index 88ca051015..e0ff708c7f 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -883,7 +883,14 @@ def key_metadata(self) -> bytes | None:
 
     @property
     def first_row_id(self) -> int | None:
-        return self._data[15]
+        # 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
@@ -1330,6 +1337,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,
@@ -1342,6 +1394,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}")
 
@@ -1465,6 +1519,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,
@@ -1472,6 +1591,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)
@@ -1479,5 +1599,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 45e4e894ef..28de0f94c4 100644
--- a/tests/utils/test_manifest.py
+++ b/tests/utils/test_manifest.py
@@ -1477,3 +1477,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 425378268126da260eb7901167ab2596ce3ce112 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/9] 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 e0ff708c7f..0ce832462a 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -1343,12 +1343,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](
@@ -1389,13 +1417,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 28de0f94c4..84948839da 100644
--- a/tests/utils/test_manifest.py
+++ b/tests/utils/test_manifest.py
@@ -1492,6 +1492,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(
@@ -1528,6 +1533,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"])
@@ -1592,6 +1610,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()
@@ -1606,3 +1631,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 fd809b80849b70ac133fc3b1f7b33630f7f52583 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/9] 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 0ce832462a..5d3fd1d097 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -1597,7 +1597,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 = {
@@ -1610,8 +1615,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
 
 
@@ -1634,7 +1644,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 84948839da..dced2c959b 100644
--- a/tests/utils/test_manifest.py
+++ b/tests/utils/test_manifest.py
@@ -1740,3 +1740,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 02deddb7745de2320491e40a399b5dc29dbb8e77 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/9] 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        |  66 +++++++++++++-----
 tests/avro/test_file.py      |  71 ++++++++++++++++++-
 tests/utils/test_manifest.py | 130 +++++++++++++++++++++++++++++++++++
 3 files changed, 250 insertions(+), 17 deletions(-)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index 5d3fd1d097..395916c62f 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 Callable, Iterator
+from collections.abc import Callable, Iterator, Mapping
 from copy import copy
 from enum import Enum
 from types import TracebackType
@@ -546,19 +546,23 @@ def sort_order_id(self) -> int | None:
 
     @property
     def first_row_id(self) -> int | None:
-        return self._data[16]
+        # 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:
-        return self._data[17]
+        # 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:
-        return self._data[18]
+        # 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:
-        return self._data[19]
+        # 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
@@ -802,6 +806,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())
@@ -918,7 +953,7 @@ def fetch_manifest_entry(
         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:
@@ -1047,7 +1082,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:
@@ -1391,9 +1426,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:
@@ -1599,15 +1633,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 175089c30b..12cf26e44c 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
 
@@ -264,6 +293,46 @@ def test_write_v2_referenced_data_file_with_fastavro() -> None:
         assert fa_entry["data_file"]["referenced_data_file"] == referenced_data_file
 
 
+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:
     common_data_file_args = {
diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py
index dced2c959b..ff0ccdf404 100644
--- a/tests/utils/test_manifest.py
+++ b/tests/utils/test_manifest.py
@@ -41,6 +41,7 @@
     ManifestFile,
     PartitionFieldSummary,
     _inherit_from_manifest,
+    _layout_version_from_field_count,
     _manifests,
     clear_manifest_cache,
     read_manifest_list,
@@ -1546,6 +1547,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"])
@@ -1616,6 +1725,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 c004569323411d0c99962d16a56cd7153d3e0141 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/9] 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 395916c62f..1fdb68af54 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -26,6 +26,7 @@
 from typing import (
     Any,
     Literal,
+    cast,
 )
 
 from cachetools import LRUCache
@@ -814,8 +815,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:
@@ -835,7 +836,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 5f1a91cbb811fd20e93e3e3b1c4202764e623c05 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/9] 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 849d1ac2b9..8fd52761a9 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

From cd47f83715dac980cb628c05894b792764aa10cb Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Sun, 6 Sep 2026 08:43:08 +0900
Subject: [PATCH 7/9] Pass first-row-id to the V3 manifest list writer

---
 pyiceberg/table/update/snapshot.py | 11 +++++----
 tests/table/test_snapshots.py      | 37 ++++++++++++++++++++++++++++++
 2 files changed, 43 insertions(+), 5 deletions(-)

diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index 8024e808b2..d5fdca427c 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -334,6 +334,11 @@ def _commit(self) -> UpdatesAndRequirements:
         manifest_list_file_path = location_provider.new_metadata_location(file_name)
         self._written_manifest_lists.append(manifest_list_file_path)
 
+        first_row_id: int | None = None
+
+        if self._transaction.table_metadata.format_version >= 3:
+            first_row_id = self._transaction.table_metadata.next_row_id
+
         with write_manifest_list(
             format_version=self._transaction.table_metadata.format_version,
             output_file=self._io.new_output(manifest_list_file_path),
@@ -341,14 +346,10 @@ def _commit(self) -> UpdatesAndRequirements:
             parent_snapshot_id=self._parent_snapshot_id,
             sequence_number=next_sequence_number,
             avro_compression=self._compression,
+            first_row_id=first_row_id,
         ) as writer:
             writer.add_manifests(new_manifests)
 
-        first_row_id: int | None = None
-
-        if self._transaction.table_metadata.format_version >= 3:
-            first_row_id = self._transaction.table_metadata.next_row_id
-
         snapshot = Snapshot(
             snapshot_id=self._snapshot_id,
             parent_snapshot_id=self._parent_snapshot_id,
diff --git a/tests/table/test_snapshots.py b/tests/table/test_snapshots.py
index 39aeb7b349..8dfba8ca4b 100644
--- a/tests/table/test_snapshots.py
+++ b/tests/table/test_snapshots.py
@@ -821,3 +821,40 @@ def test_overwrite_rejects_explicit_delete_without_parent_snapshot(
         with empty.transaction() as tx:
             with tx.update_snapshot().overwrite() as overwrite:
                 overwrite.delete_data_file(stale_file)
+
+
+def test_manifest_list_receives_first_row_id_for_v3(table_v3: Table) -> None:
+    """The V3 manifest list writer requires first-row-id, so the producer has to supply it."""
+    from unittest import mock
+
+    from pyiceberg.table.update.snapshot import _FastAppendFiles
+
+    txn = table_v3.transaction()
+    append = _FastAppendFiles(operation=Operation.APPEND, transaction=txn, io=table_v3.io)
+
+    with (
+        mock.patch("pyiceberg.table.update.snapshot.write_manifest_list") as writer,
+        mock.patch.object(_FastAppendFiles, "_manifests", return_value=[]),
+    ):
+        append._commit()
+
+    assert writer.call_args.kwargs["format_version"] == 3
+    assert writer.call_args.kwargs["first_row_id"] == table_v3.metadata.next_row_id
+
+
+def test_manifest_list_has_no_first_row_id_for_v2(table_v2: Table) -> None:
+    from unittest import mock
+
+    from pyiceberg.table.update.snapshot import _FastAppendFiles
+
+    txn = table_v2.transaction()
+    append = _FastAppendFiles(operation=Operation.APPEND, transaction=txn, io=table_v2.io)
+
+    with (
+        mock.patch("pyiceberg.table.update.snapshot.write_manifest_list") as writer,
+        mock.patch.object(_FastAppendFiles, "_manifests", return_value=[]),
+    ):
+        append._commit()
+
+    assert writer.call_args.kwargs["format_version"] == 2
+    assert writer.call_args.kwargs["first_row_id"] is None

From 476b5caf76e19f4ea3b86e5313f65b0e23bcb47e Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Sun, 6 Sep 2026 10:38:35 +0900
Subject: [PATCH 8/9] Adapt V3 manifest writers to the V3 manifest list schema
 now on main

---
 pyiceberg/manifest.py   | 3 ++-
 tests/avro/test_file.py | 4 +++-
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index 1fdb68af54..f9dfda86af 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -1411,7 +1411,8 @@ def to_manifest_file(self) -> ManifestFile:
             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)
+        args["first_row_id"] = self._first_row_id
+        return ManifestFile.from_args(_table_format_version=3, **args)
 
     def new_writer(self) -> AvroOutputFile[ManifestEntry]:
         # Use the V3 record layout so the V3-only data file fields are written
diff --git a/tests/avro/test_file.py b/tests/avro/test_file.py
index 12cf26e44c..5ad680d218 100644
--- a/tests/avro/test_file.py
+++ b/tests/avro/test_file.py
@@ -297,9 +297,11 @@ 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.
+    inconsistent for other record types with version-dependent trailing fields. The binding is
+    explicit because the default read version is V3, whose layout does carry first_row_id.
     """
     manifest_file = ManifestFile.from_args(
+        _table_format_version=2,
         manifest_path="/manifests/m1.avro",
         manifest_length=100,
         partition_spec_id=0,

From c152089dabd38d065b8147c0137bfb9a78a80d55 Mon Sep 17 00:00:00 2001
From: Noritaka Sekiyama <noritaka.sekiyama@databricks.com>
Date: Wed, 7 Oct 2026 18:59:26 +0900
Subject: [PATCH 9/9] Drop V1-only fields when rebinding a data file to the V3
 layout

DataFile.from_args now rejects fields that are not in the target
layout, so rebinding a V1 data file to V3 raised on
block_size_in_bytes, which V3 does not have.

Generated-by: Claude Code
---
 pyiceberg/manifest.py | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index f9dfda86af..1e1c238591 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -1429,7 +1429,13 @@ def _wrap_data_file(self, data_file: DataFile) -> DataFile:
         if len(data_file._data) >= len(DATA_FILE_TYPE[3].fields):
             return data_file
         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)}
+        v3_field_names = {field.name for field in DATA_FILE_TYPE[3].fields}
+        # V1-only fields such as block_size_in_bytes have no V3 counterpart and are dropped
+        args = {
+            field.name: value
+            for field, value in zip(DATA_FILE_TYPE[source_version].fields, data_file._data, strict=True)
+            if field.name in v3_field_names
+        }
         return DataFile.from_args(_table_format_version=3, **args)
 
     def prepare_entry(self, entry: ManifestEntry) -> ManifestEntry: