diff --git a/pyiceberg/avro/resolver.py b/pyiceberg/avro/resolver.py index 81b573aa79..ecd80052cf 100644 --- a/pyiceberg/avro/resolver.py +++ b/pyiceberg/avro/resolver.py @@ -15,6 +15,7 @@ # specific language governing permissions and limitations # under the License. # pylint: disable=arguments-renamed,unused-argument +import warnings from collections.abc import Callable from enum import Enum @@ -461,7 +462,15 @@ def primitive(self, primitive: PrimitiveType, expected_primitive: IcebergType | # ensure that the type can be projected to the expected if primitive != expected_primitive: - promote(primitive, expected_primitive) + if isinstance(primitive, LongType) and isinstance(expected_primitive, IntegerType): + warnings.warn( + "Encountered non-compliant manifest with long equality_ids (spec requires int). " + "Support for legacy long equality_ids is deprecated and will be removed in a future release.", + DeprecationWarning, + stacklevel=2, + ) + else: + promote(primitive, expected_primitive) return super().primitive(primitive, expected_primitive) diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 37dbd04b13..96ea0e322d 100644 --- a/pyiceberg/manifest.py +++ b/pyiceberg/manifest.py @@ -295,7 +295,7 @@ def __repr__(self) -> str: NestedField( field_id=135, name="equality_ids", - field_type=ListType(element_id=136, element_type=LongType(), element_required=True), + field_type=ListType(element_id=136, element_type=IntegerType(), element_required=True), required=False, doc="Field ids used to determine row equality in equality delete files.", ), @@ -390,7 +390,7 @@ def __repr__(self) -> str: NestedField( field_id=135, name="equality_ids", - field_type=ListType(element_id=136, element_type=LongType(), element_required=True), + field_type=ListType(element_id=136, element_type=IntegerType(), element_required=True), required=False, doc="Field ids used to determine row equality in equality delete files.", ), @@ -433,7 +433,9 @@ def __repr__(self) -> str: } -def data_file_with_partition(partition_type: StructType, format_version: TableVersion) -> StructType: +def data_file_with_partition( + partition_type: StructType, format_version: TableVersion, legacy_equality_ids: bool = False +) -> StructType: data_file_partition_type = StructType( *[ NestedField( @@ -446,20 +448,32 @@ def data_file_with_partition(partition_type: StructType, format_version: TableVe ] ) - return StructType( - *[ - NestedField( - field_id=102, - name="partition", - field_type=data_file_partition_type, - required=True, - doc="Partition data tuple, schema based on the partition spec", + fields = [] + for field in DATA_FILE_TYPE[format_version].fields: + if field.field_id == 102: + fields.append( + NestedField( + field_id=102, + name="partition", + field_type=data_file_partition_type, + required=True, + doc="Partition data tuple, schema based on the partition spec", + ) ) - if field.field_id == 102 - else field - for field in DATA_FILE_TYPE[format_version].fields - ] - ) + elif field.field_id == 135 and legacy_equality_ids: + fields.append( + NestedField( + field_id=135, + name="equality_ids", + field_type=ListType(element_id=136, element_type=LongType(), element_required=True), + required=False, + doc="Field ids used to determine row equality in equality delete files.", + ) + ) + else: + fields.append(field) + + return StructType(*fields) class DataFile(Record): @@ -1058,6 +1072,7 @@ class ManifestWriter(ABC): _min_sequence_number: int | None _partitions: list[Record] _compression: AvroCompressionCodec + _legacy_equality_ids: bool def __init__( self, @@ -1066,6 +1081,7 @@ def __init__( output_file: OutputFile, snapshot_id: int, avro_compression: AvroCompressionCodec, + legacy_equality_ids: bool = False, ) -> None: self.closed = False self._spec = spec @@ -1082,6 +1098,7 @@ def __init__( self._min_sequence_number = None self._partitions = [] self._compression = avro_compression + self._legacy_equality_ids = legacy_equality_ids def __enter__(self) -> ManifestWriter: """Open the writer.""" @@ -1125,7 +1142,9 @@ def _meta(self) -> dict[str, str]: def _with_partition(self, format_version: TableVersion) -> Schema: data_file_type = data_file_with_partition( - format_version=format_version, partition_type=self._spec.partition_type(self._schema) + format_version=format_version, + partition_type=self._spec.partition_type(self._schema), + legacy_equality_ids=self._legacy_equality_ids, ) return manifest_entry_schema_with_data_file(format_version=format_version, data_file=data_file_type) @@ -1238,8 +1257,9 @@ def __init__( output_file: OutputFile, snapshot_id: int, avro_compression: AvroCompressionCodec, + legacy_equality_ids: bool = False, ): - super().__init__(spec, schema, output_file, snapshot_id, avro_compression) + super().__init__(spec, schema, output_file, snapshot_id, avro_compression, legacy_equality_ids=legacy_equality_ids) def content(self) -> ManifestContent: return ManifestContent.DATA @@ -1260,8 +1280,9 @@ def __init__( output_file: OutputFile, snapshot_id: int, avro_compression: AvroCompressionCodec, + legacy_equality_ids: bool = False, ): - super().__init__(spec, schema, output_file, snapshot_id, avro_compression) + super().__init__(spec, schema, output_file, snapshot_id, avro_compression, legacy_equality_ids=legacy_equality_ids) def content(self) -> ManifestContent: return ManifestContent.DATA @@ -1293,11 +1314,12 @@ def write_manifest( output_file: OutputFile, snapshot_id: int, avro_compression: AvroCompressionCodec, + legacy_equality_ids: bool = False, ) -> ManifestWriter: if format_version == 1: - return ManifestWriterV1(spec, schema, output_file, snapshot_id, avro_compression) + return ManifestWriterV1(spec, schema, output_file, snapshot_id, avro_compression, legacy_equality_ids=legacy_equality_ids) elif format_version == 2: - return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression) + return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression, legacy_equality_ids=legacy_equality_ids) else: raise ValueError(f"Cannot write manifest for table version: {format_version}") diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index bb879dfbce..67b84a6d28 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -167,6 +167,9 @@ class TableProperties: WRITE_AVRO_COMPRESSION = "write.avro.compression-codec" WRITE_AVRO_COMPRESSION_DEFAULT = "gzip" + WRITE_MANIFEST_LEGACY_LONG_EQUALITY_IDS = "write.manifest.legacy-long-equality-ids" + WRITE_MANIFEST_LEGACY_LONG_EQUALITY_IDS_DEFAULT = False + DEFAULT_WRITE_METRICS_MODE = "write.metadata.metrics.default" DEFAULT_WRITE_METRICS_MODE_DEFAULT = "truncate(16)" diff --git a/tests/table/test_validate.py b/tests/table/test_validate.py index a19983fd66..bca5bb2efe 100644 --- a/tests/table/test_validate.py +++ b/tests/table/test_validate.py @@ -22,7 +22,15 @@ from pyiceberg.exceptions import ValidationException from pyiceberg.io import FileIO -from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile +from pyiceberg.manifest import ( + DataFile, + DataFileContent, + ManifestContent, + ManifestEntry, + ManifestEntryStatus, + ManifestFile, + clear_manifest_cache, +) from pyiceberg.table import Table from pyiceberg.table.snapshots import Operation, Snapshot, Summary from pyiceberg.table.update.validate import ( @@ -42,6 +50,7 @@ def table_v2_with_extensive_snapshots_and_manifests( table_v2_with_extensive_snapshots: Table, ) -> tuple[Table, dict[int, list[ManifestFile]]]: """Fixture to create a table with extensive snapshots and manifests.""" + clear_manifest_cache() mock_manifests = {} for i, snapshot in enumerate(table_v2_with_extensive_snapshots.snapshots()): diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py index 0535ec01ed..13b5df8133 100644 --- a/tests/utils/test_manifest.py +++ b/tests/utils/test_manifest.py @@ -15,7 +15,6 @@ # specific language governing permissions and limitations # under the License. # pylint: disable=redefined-outer-name,arguments-renamed,fixme -import importlib from pathlib import Path from tempfile import TemporaryDirectory from typing import Any @@ -1087,104 +1086,196 @@ def test_clear_manifest_cache() -> None: def test_manifest_cache_can_be_disabled_with_size_zero(monkeypatch: pytest.MonkeyPatch) -> None: """Test that manifest-cache-size=0 disables caching.""" monkeypatch.setenv("PYICEBERG_MANIFEST_CACHE_SIZE", "0") - importlib.reload(manifest_module) + cache = manifest_module._ManifestCache() + assert cache.maxsize == 0 + assert len(cache) == 0 - try: - assert manifest_module._manifest_cache.maxsize == 0 - assert len(manifest_module._manifest_cache) == 0 - - io = PyArrowFileIO() - - with TemporaryDirectory() as tmp_dir: - list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="disabled", snapshot_id=1) - - manifests_first_call = manifest_module._manifests(io, list_path) - manifests_second_call = manifest_module._manifests(io, list_path) + io = PyArrowFileIO() + with TemporaryDirectory() as tmp_dir: + list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="disabled", snapshot_id=1) + monkeypatch.setattr(manifest_module, "_manifest_cache", cache) + manifests_first_call = manifest_module._manifests(io, list_path) + manifests_second_call = manifest_module._manifests(io, list_path) - assert len(manifest_module._manifest_cache) == 0 - assert manifests_first_call[0] is not manifests_second_call[0] - finally: - monkeypatch.delenv("PYICEBERG_MANIFEST_CACHE_SIZE", raising=False) - importlib.reload(manifest_module) + assert len(cache) == 0 + assert manifests_first_call[0] is not manifests_second_call[0] def test_manifest_cache_respects_positive_env_size(monkeypatch: pytest.MonkeyPatch) -> None: """Test that a positive manifest-cache-size enables a bounded cache.""" monkeypatch.setenv("PYICEBERG_MANIFEST_CACHE_SIZE", "1") - importlib.reload(manifest_module) - - try: - assert manifest_module._manifest_cache.maxsize == 1 - - io = PyArrowFileIO() - - with TemporaryDirectory() as tmp_dir: - first_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="first", snapshot_id=1) - second_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="second", snapshot_id=2) + cache = manifest_module._ManifestCache() + assert cache.maxsize == 1 - manifests_first_call = manifest_module._manifests(io, first_list_path) - manifests_second_call = manifest_module._manifests(io, first_list_path) - - assert manifests_first_call[0] is manifests_second_call[0] - assert len(manifest_module._manifest_cache) == 1 + io = PyArrowFileIO() + with TemporaryDirectory() as tmp_dir: + first_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="first", snapshot_id=1) + second_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="second", snapshot_id=2) + monkeypatch.setattr(manifest_module, "_manifest_cache", cache) + manifests_first_call = manifest_module._manifests(io, first_list_path) + manifests_second_call = manifest_module._manifests(io, first_list_path) - manifest_module._manifests(io, second_list_path) + assert manifests_first_call[0] is manifests_second_call[0] + assert len(cache) == 1 - assert len(manifest_module._manifest_cache) == 1 - finally: - monkeypatch.delenv("PYICEBERG_MANIFEST_CACHE_SIZE", raising=False) - importlib.reload(manifest_module) + manifest_module._manifests(io, second_list_path) + assert len(cache) == 1 def test_manifest_cache_reads_size_from_configuration_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: """Test that manifest-cache-size can be loaded from .pyiceberg.yaml.""" - config_dir = tmp_path / "config" - config_dir.mkdir() - (config_dir / ".pyiceberg.yaml").write_text("manifest-cache-size: 2\n", encoding="utf-8") - - monkeypatch.delenv("PYICEBERG_MANIFEST_CACHE_SIZE", raising=False) - monkeypatch.setenv("PYICEBERG_HOME", str(config_dir)) - importlib.reload(manifest_module) + from pyiceberg.utils.config import Config - try: - assert manifest_module._manifest_cache.maxsize == 2 + monkeypatch.setattr(Config, "get_int", lambda self, key: 2 if key == "manifest-cache-size" else None) + cache = manifest_module._ManifestCache() + assert cache.maxsize == 2 - io = PyArrowFileIO() - - with TemporaryDirectory() as tmp_dir: - first_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="first", snapshot_id=1) - second_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="second", snapshot_id=2) - third_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="third", snapshot_id=3) - - manifest_module._manifests(io, first_list_path) - manifest_module._manifests(io, second_list_path) - manifest_module._manifests(io, third_list_path) - - assert len(manifest_module._manifest_cache) == 2 - finally: - monkeypatch.delenv("PYICEBERG_HOME", raising=False) - importlib.reload(manifest_module) + io = PyArrowFileIO() + with TemporaryDirectory() as tmp_dir: + first_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="first", snapshot_id=1) + second_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="second", snapshot_id=2) + third_list_path = _create_test_manifest_list(manifest_module, io, tmp_dir, name="third", snapshot_id=3) + monkeypatch.setattr(manifest_module, "_manifest_cache", cache) + manifest_module._manifests(io, first_list_path) + manifest_module._manifests(io, second_list_path) + manifest_module._manifests(io, third_list_path) + assert len(cache) == 2 def test_invalid_manifest_cache_size_raises_value_error(monkeypatch: pytest.MonkeyPatch) -> None: """Test that invalid manifest-cache-size values raise a helpful error.""" monkeypatch.setenv("PYICEBERG_MANIFEST_CACHE_SIZE", "not-an-int") - - try: - with pytest.raises(ValueError, match="manifest-cache-size should be an integer or left unset"): - importlib.reload(manifest_module) - finally: - monkeypatch.delenv("PYICEBERG_MANIFEST_CACHE_SIZE", raising=False) - importlib.reload(manifest_module) + with pytest.raises(ValueError, match="manifest-cache-size should be an integer or left unset"): + manifest_module._ManifestCache() def test_negative_manifest_cache_size_raises_value_error(monkeypatch: pytest.MonkeyPatch) -> None: """Test that negative manifest-cache-size values raise a helpful error.""" monkeypatch.setenv("PYICEBERG_MANIFEST_CACHE_SIZE", "-1") + with pytest.raises(ValueError, match="manifest-cache-size should be a non-negative integer or left unset"): + manifest_module._ManifestCache() + + +def test_write_manifest_equality_ids_int_schema(tmp_path: Path) -> None: + io = PyArrowFileIO() + schema = Schema(NestedField(1, "x", IntegerType())) + df = DataFile.from_args( + content=DataFileContent.DATA, + file_path="file:///tmp/test.parquet", + file_format=FileFormat.PARQUET, + partition=(), + record_count=10, + file_size_in_bytes=100, + equality_ids=[1, 2, 3], + ) + manifest_path = f"file://{tmp_path / 'test_equality_ids.avro'}" + with write_manifest( + format_version=2, + spec=UNPARTITIONED_PARTITION_SPEC, + schema=schema, + output_file=io.new_output(manifest_path), + snapshot_id=12345, + avro_compression="null", + ) as writer: + writer.add_entry( + ManifestEntry.from_args( + status=ManifestEntryStatus.ADDED, + snapshot_id=12345, + sequence_number=1, + file_sequence_number=1, + data_file=df, + ) + ) + + with open(tmp_path / "test_equality_ids.avro", "rb") as f: + reader = fastavro.reader(f) + writer_schema = reader.writer_schema + fields = {f["name"]: f for f in writer_schema["fields"]} + df_fields = {f["name"]: f for f in fields["data_file"]["type"]["fields"]} + assert df_fields["equality_ids"]["type"][1]["items"] == "int" - try: - with pytest.raises(ValueError, match="manifest-cache-size should be a non-negative integer or left unset"): - importlib.reload(manifest_module) - finally: - monkeypatch.delenv("PYICEBERG_MANIFEST_CACHE_SIZE", raising=False) - importlib.reload(manifest_module) + +def test_read_manifest_legacy_equality_ids_long_schema(tmp_path: Path) -> None: + io = PyArrowFileIO() + schema = Schema(NestedField(1, "x", IntegerType())) + df = DataFile.from_args( + content=DataFileContent.DATA, + file_path="file:///tmp/test.parquet", + file_format=FileFormat.PARQUET, + partition=(), + record_count=10, + file_size_in_bytes=100, + equality_ids=[1, 2, 3], + ) + manifest_path = f"file://{tmp_path / 'test_legacy.avro'}" + with write_manifest( + format_version=2, + spec=UNPARTITIONED_PARTITION_SPEC, + schema=schema, + output_file=io.new_output(manifest_path), + snapshot_id=12345, + avro_compression="null", + legacy_equality_ids=True, + ) as writer: + writer.add_entry( + ManifestEntry.from_args( + status=ManifestEntryStatus.ADDED, + snapshot_id=12345, + sequence_number=1, + file_sequence_number=1, + data_file=df, + ) + ) + + mf = ManifestFile.from_args( + manifest_path=manifest_path, + manifest_length=1000, + partition_spec_id=0, + added_snapshot_id=12345, + sequence_number=1, + partitions=[], + ) + with pytest.deprecated_call(match="Encountered non-compliant manifest with long equality_ids"): + entries = mf.fetch_manifest_entry(io) + assert len(entries) == 1 + assert entries[0].data_file.equality_ids == [1, 2, 3] + + +def test_write_manifest_legacy_equality_ids_long_option(tmp_path: Path) -> None: + io = PyArrowFileIO() + schema = Schema(NestedField(1, "x", IntegerType())) + df = DataFile.from_args( + content=DataFileContent.DATA, + file_path="file:///tmp/test.parquet", + file_format=FileFormat.PARQUET, + partition=(), + record_count=10, + file_size_in_bytes=100, + equality_ids=[1, 2, 3], + ) + manifest_path = f"file://{tmp_path / 'test_legacy_opt.avro'}" + with write_manifest( + format_version=2, + spec=UNPARTITIONED_PARTITION_SPEC, + schema=schema, + output_file=io.new_output(manifest_path), + snapshot_id=12345, + avro_compression="null", + legacy_equality_ids=True, + ) as writer: + writer.add_entry( + ManifestEntry.from_args( + status=ManifestEntryStatus.ADDED, + snapshot_id=12345, + sequence_number=1, + file_sequence_number=1, + data_file=df, + ) + ) + + with open(tmp_path / "test_legacy_opt.avro", "rb") as f: + reader = fastavro.reader(f) + writer_schema = reader.writer_schema + fields = {f["name"]: f for f in writer_schema["fields"]} + df_fields = {f["name"]: f for f in fields["data_file"]["type"]["fields"]} + assert df_fields["equality_ids"]["type"][1]["items"] == "long"