diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 9842f79d8e..04aa134fe8 100644 --- a/pyiceberg/manifest.py +++ b/pyiceberg/manifest.py @@ -19,13 +19,14 @@ 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 from typing import ( Any, Literal, + cast, ) from cachetools import LRUCache @@ -532,6 +533,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 +795,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 = 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: + """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 cast(TableVersion, matches[0]) + POSITIONAL_DELETE_SCHEMA = Schema( NestedField(2147483546, "file_path", StringType()), NestedField(2147483545, "pos", IntegerType()) @@ -853,6 +905,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 @@ -873,7 +936,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: @@ -996,7 +1059,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: @@ -1284,6 +1347,78 @@ 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. + + 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]( + 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 + 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: + 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, @@ -1291,11 +1426,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, first_row_id) else: raise ValueError(f"Cannot write manifest for table version: {format_version}") @@ -1419,6 +1559,81 @@ 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. + + 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. + """ + 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[source_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: + 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 + wrapped_manifest_file.added_rows_count + return wrapped_manifest_file + + def write_manifest_list( format_version: TableVersion, output_file: OutputFile, @@ -1426,6 +1641,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 +1649,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("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/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/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 diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py index f2ae1e05ad..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, @@ -1146,3 +1147,423 @@ 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, + 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( + 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"] + 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" + # 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"]) +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] + + # 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 + # 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: + 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, + ) + + +@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" + + +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])