diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 88ca051015..03a4416bee 100644 --- a/pyiceberg/manifest.py +++ b/pyiceberg/manifest.py @@ -1297,6 +1297,8 @@ def prepare_entry(self, entry: ManifestEntry) -> ManifestEntry: class ManifestWriterV2(ManifestWriter): + _content: ManifestContent + def __init__( self, spec: PartitionSpec, @@ -1304,11 +1306,13 @@ def __init__( output_file: OutputFile, snapshot_id: int, avro_compression: AvroCompressionCodec, + content: ManifestContent = ManifestContent.DATA, ): super().__init__(spec, schema, output_file, snapshot_id, avro_compression) + self._content = content def content(self) -> ManifestContent: - return ManifestContent.DATA + return self._content @property def version(self) -> TableVersion: @@ -1318,7 +1322,7 @@ def version(self) -> TableVersion: def _meta(self) -> dict[str, str]: return { **super()._meta, - "content": "data", + "content": "data" if self._content == ManifestContent.DATA else "deletes", } def prepare_entry(self, entry: ManifestEntry) -> ManifestEntry: @@ -1337,11 +1341,14 @@ def write_manifest( output_file: OutputFile, snapshot_id: int, avro_compression: AvroCompressionCodec, + content: ManifestContent = ManifestContent.DATA, ) -> ManifestWriter: if format_version == 1: + if content != ManifestContent.DATA: + raise ValidationError("Cannot write delete manifests in a v1 table") return ManifestWriterV1(spec, schema, output_file, snapshot_id, avro_compression) elif format_version == 2: - return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression) + return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression, content) 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 331146346e..3c7d569abc 100644 --- a/tests/utils/test_manifest.py +++ b/tests/utils/test_manifest.py @@ -26,6 +26,7 @@ import pyiceberg.manifest as manifest_module from pyiceberg.avro.codecs import AvroCompressionCodec from pyiceberg.avro.file import AvroFile, AvroOutputFile +from pyiceberg.exceptions import ValidationError from pyiceberg.io import load_file_io from pyiceberg.io.pyarrow import PyArrowFileIO from pyiceberg.manifest import ( @@ -1376,3 +1377,104 @@ 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("content", [ManifestContent.DATA, ManifestContent.DELETES]) +def test_write_manifest_content( + generated_manifest_file_file_v2: str, + test_schema: Schema, + test_partition_spec: PartitionSpec, + content: ManifestContent, +) -> None: + """A v2 manifest must record the content it was asked to write. + + Without this the writer always claims `data`, so a delete manifest written by + PyIceberg would be read back as a data manifest and its entries applied as + live data files. + """ + io = load_file_io() + snapshot = Snapshot( + snapshot_id=25, + parent_snapshot_id=19, + timestamp_ms=1602638573590, + manifest_list=generated_manifest_file_file_v2, + summary=Summary(Operation.APPEND), + schema_id=3, + ) + manifest_entries = snapshot.manifests(io)[0].fetch_manifest_entry(io) + + with TemporaryDirectory() as tmpdir: + tmp_avro_file = tmpdir + "/test_write_manifest_content.avro" + with write_manifest( + format_version=2, + spec=test_partition_spec, + schema=test_schema, + output_file=io.new_output(tmp_avro_file), + snapshot_id=8744736658442914487, + avro_compression="deflate", + content=content, + ) as writer: + for entry in manifest_entries: + writer.add_entry(entry) + new_manifest = writer.to_manifest_file() + + assert new_manifest.content == content + _verify_metadata_with_fastavro( + tmp_avro_file, + {"content": "data" if content == ManifestContent.DATA else "deletes"}, + ) + + +def test_write_manifest_defaults_to_data_content( + generated_manifest_file_file_v2: str, + test_schema: Schema, + test_partition_spec: PartitionSpec, +) -> None: + """Callers that do not ask for a content type still get a data manifest.""" + io = load_file_io() + snapshot = Snapshot( + snapshot_id=25, + parent_snapshot_id=19, + timestamp_ms=1602638573590, + manifest_list=generated_manifest_file_file_v2, + summary=Summary(Operation.APPEND), + schema_id=3, + ) + manifest_entries = snapshot.manifests(io)[0].fetch_manifest_entry(io) + + with TemporaryDirectory() as tmpdir: + tmp_avro_file = tmpdir + "/test_write_manifest_default_content.avro" + with write_manifest( + format_version=2, + spec=test_partition_spec, + schema=test_schema, + output_file=io.new_output(tmp_avro_file), + snapshot_id=8744736658442914487, + avro_compression="deflate", + ) as writer: + for entry in manifest_entries: + writer.add_entry(entry) + assert writer.to_manifest_file().content == ManifestContent.DATA + + +def test_write_manifest_v1_rejects_delete_content( + test_schema: Schema, + test_partition_spec: PartitionSpec, +) -> None: + """v1 has no delete files, so asking for a delete manifest must fail loudly. + + `ManifestListWriterV1` already refuses to store one; rejecting it at the + writer keeps a v1 table from producing a manifest it could never reference. + """ + io = load_file_io() + with TemporaryDirectory() as tmpdir: + with pytest.raises(ValidationError, match="Cannot write delete manifests in a v1 table"): + write_manifest( + format_version=1, + spec=test_partition_spec, + schema=test_schema, + output_file=io.new_output(tmpdir + "/test_write_manifest_v1_deletes.avro"), + snapshot_id=8744736658442914487, + avro_compression="deflate", + content=ManifestContent.DELETES, + )