From d161e78677016609138d55d0d00ba2f31afa33f5 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Wed, 8 Jul 2026 20:10:02 +0900 Subject: [PATCH 1/3] 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 Date: Wed, 8 Jul 2026 21:13:38 +0900 Subject: [PATCH 2/3] 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 Date: Thu, 9 Jul 2026 09:07:22 +0900 Subject: [PATCH 3/3] 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])