From 4b4ac26e54098e932d3c17d60f3a9e4b602f882f Mon Sep 17 00:00:00 2001 From: hedger9487 Date: Tue, 25 Aug 2026 04:44:21 +0800 Subject: [PATCH 1/4] Manifest: Use int for equality_ids in manifest schema per Iceberg spec (#3840) --- pyiceberg/avro/resolver.py | 5 +- pyiceberg/manifest.py | 4 +- tests/utils/test_manifest.py | 97 ++++++++++++++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 3 deletions(-) diff --git a/pyiceberg/avro/resolver.py b/pyiceberg/avro/resolver.py index 81b573aa79..1e9e18fb48 100644 --- a/pyiceberg/avro/resolver.py +++ b/pyiceberg/avro/resolver.py @@ -461,7 +461,10 @@ 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): + pass + else: + promote(primitive, expected_primitive) return super().primitive(primitive, expected_primitive) diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 4a61cc5de2..612064285d 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.", ), diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py index 6811ab2942..19342f7abe 100644 --- a/tests/utils/test_manifest.py +++ b/tests/utils/test_manifest.py @@ -1296,3 +1296,100 @@ 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) + + +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" + + +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", + ) 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.avro", "rb") as f: + reader = fastavro.reader(f) + records = list(reader) + writer_schema = reader.writer_schema + for field in writer_schema["fields"]: + if field["name"] == "data_file": + for df_field in field["type"]["fields"]: + if df_field["name"] == "equality_ids": + df_field["type"][1]["items"] = "long" + + legacy_file_path = tmp_path / "test_legacy_modified.avro" + with open(legacy_file_path, "wb") as f: + fastavro.writer(f, writer_schema, records) + + mf = ManifestFile.from_args( + manifest_path=f"file://{legacy_file_path}", + manifest_length=1000, + partition_spec_id=0, + added_snapshot_id=12345, + sequence_number=1, + partitions=[], + ) + entries = mf.fetch_manifest_entry(io) + assert len(entries) == 1 + assert entries[0].data_file.equality_ids == [1, 2, 3] From 2266aaa66fad35c2c7d425582bfd166bf87333a0 Mon Sep 17 00:00:00 2001 From: hedger9487 Date: Wed, 26 Aug 2026 02:39:47 +0800 Subject: [PATCH 2/4] Support backwards-compatible long equality_ids with deprecation warning and config flag --- pyiceberg/avro/resolver.py | 8 +- pyiceberg/manifest.py | 64 +++++++++---- pyiceberg/table/__init__.py | 3 + tests/table/test_validate.py | 11 ++- tests/utils/test_manifest.py | 177 +++++++++++++++++------------------ 5 files changed, 150 insertions(+), 113 deletions(-) diff --git a/pyiceberg/avro/resolver.py b/pyiceberg/avro/resolver.py index 1e9e18fb48..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 @@ -462,7 +463,12 @@ def primitive(self, primitive: PrimitiveType, expected_primitive: IcebergType | # ensure that the type can be projected to the expected if primitive != expected_primitive: if isinstance(primitive, LongType) and isinstance(expected_primitive, IntegerType): - pass + 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) diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 612064285d..3742ce2f39 100644 --- a/pyiceberg/manifest.py +++ b/pyiceberg/manifest.py @@ -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): @@ -1078,6 +1092,7 @@ class ManifestWriter(ABC): _min_sequence_number: int | None _partitions: list[Record] _compression: AvroCompressionCodec + _legacy_equality_ids: bool def __init__( self, @@ -1086,6 +1101,7 @@ def __init__( output_file: OutputFile, snapshot_id: int, avro_compression: AvroCompressionCodec, + legacy_equality_ids: bool = False, ) -> None: self.closed = False self._spec = spec @@ -1102,6 +1118,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.""" @@ -1145,7 +1162,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) @@ -1258,8 +1277,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 @@ -1280,8 +1300,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 @@ -1313,11 +1334,16 @@ 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 9624eac981..fdc27fd90b 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 19342f7abe..dd88769975 100644 --- a/tests/utils/test_manifest.py +++ b/tests/utils/test_manifest.py @@ -1195,107 +1195,72 @@ 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) - - try: - assert manifest_module._manifest_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) + monkeypatch.setattr(manifest_module.Config, "get_int", lambda self, key: 2 if key == "manifest-cache-size" else None) + cache = manifest_module._ManifestCache() + assert cache.maxsize == 2 - 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") - - 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) + 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: @@ -1357,6 +1322,7 @@ def test_read_manifest_legacy_equality_ids_long_schema(tmp_path: Path) -> None: 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( @@ -1368,28 +1334,55 @@ def test_read_manifest_legacy_equality_ids_long_schema(tmp_path: Path) -> None: ) ) - with open(tmp_path / "test_legacy.avro", "rb") as f: - reader = fastavro.reader(f) - records = list(reader) - writer_schema = reader.writer_schema - for field in writer_schema["fields"]: - if field["name"] == "data_file": - for df_field in field["type"]["fields"]: - if df_field["name"] == "equality_ids": - df_field["type"][1]["items"] = "long" - - legacy_file_path = tmp_path / "test_legacy_modified.avro" - with open(legacy_file_path, "wb") as f: - fastavro.writer(f, writer_schema, records) - mf = ManifestFile.from_args( - manifest_path=f"file://{legacy_file_path}", + manifest_path=manifest_path, manifest_length=1000, partition_spec_id=0, added_snapshot_id=12345, sequence_number=1, partitions=[], ) - entries = mf.fetch_manifest_entry(io) + 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" From f8a3d37f72e750dd5bca5136839d11a56940b6f2 Mon Sep 17 00:00:00 2001 From: hedger9487 Date: Wed, 26 Aug 2026 03:04:35 +0800 Subject: [PATCH 3/4] Fix mypy and formatting issues in manifest tests --- pyiceberg/manifest.py | 8 ++------ tests/utils/test_manifest.py | 5 +++-- 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 3742ce2f39..4b2c7d75f4 100644 --- a/pyiceberg/manifest.py +++ b/pyiceberg/manifest.py @@ -1337,13 +1337,9 @@ def write_manifest( legacy_equality_ids: bool = False, ) -> ManifestWriter: if format_version == 1: - return ManifestWriterV1( - spec, schema, output_file, snapshot_id, avro_compression, legacy_equality_ids=legacy_equality_ids - ) + 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, legacy_equality_ids=legacy_equality_ids - ) + 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/tests/utils/test_manifest.py b/tests/utils/test_manifest.py index dd88769975..cc7de36f86 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 @@ -1233,7 +1232,9 @@ def test_manifest_cache_respects_positive_env_size(monkeypatch: pytest.MonkeyPat 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.""" - monkeypatch.setattr(manifest_module.Config, "get_int", lambda self, key: 2 if key == "manifest-cache-size" else None) + from pyiceberg.utils.config import Config + + monkeypatch.setattr(Config, "get_int", lambda self, key: 2 if key == "manifest-cache-size" else None) cache = manifest_module._ManifestCache() assert cache.maxsize == 2 From 05b4e0f3641712b1479cae1563989e26abd3a0f4 Mon Sep 17 00:00:00 2001 From: hedger9487 Date: Tue, 1 Sep 2026 03:27:00 +0800 Subject: [PATCH 4/4] Tolerate legacy long equality_ids in resolver and write int in manifest (#3840) --- pyiceberg/avro/resolver.py | 20 +++--- pyiceberg/manifest.py | 57 ++++++----------- pyiceberg/table/__init__.py | 3 - tests/avro/test_resolver.py | 85 +++++++++++++++++++++++++ tests/utils/test_manifest.py | 118 ++++++++++++++++------------------- 5 files changed, 167 insertions(+), 116 deletions(-) diff --git a/pyiceberg/avro/resolver.py b/pyiceberg/avro/resolver.py index ecd80052cf..0920b00c8c 100644 --- a/pyiceberg/avro/resolver.py +++ b/pyiceberg/avro/resolver.py @@ -14,8 +14,6 @@ # KIND, either express or implied. See the License for the # 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 @@ -379,6 +377,9 @@ def visit_geography(self, geography_type: "GeographyType", partner: IcebergType return BinaryWriter() +_MANIFEST_DATA_FILE_EQUALITY_IDS_ELEMENT_ID = 136 + + class ReadSchemaResolver(PrimitiveWithPartnerVisitor[IcebergType, Reader]): __slots__ = ("read_types", "read_enums", "context") read_types: dict[int, Callable[..., StructProtocol]] @@ -462,13 +463,14 @@ def primitive(self, primitive: PrimitiveType, expected_primitive: IcebergType | # ensure that the type can be projected to the expected if 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, - ) + is_manifest_data_file = getattr(self.read_types.get(2), "__name__", "") == "DataFile" + if ( + is_manifest_data_file + and _MANIFEST_DATA_FILE_EQUALITY_IDS_ELEMENT_ID in self.context + and isinstance(primitive, LongType) + and isinstance(expected_primitive, IntegerType) + ): + pass else: promote(primitive, expected_primitive) diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 4b2c7d75f4..e66bbeb9da 100644 --- a/pyiceberg/manifest.py +++ b/pyiceberg/manifest.py @@ -433,9 +433,7 @@ def __repr__(self) -> str: } -def data_file_with_partition( - partition_type: StructType, format_version: TableVersion, legacy_equality_ids: bool = False -) -> StructType: +def data_file_with_partition(partition_type: StructType, format_version: TableVersion) -> StructType: data_file_partition_type = StructType( *[ NestedField( @@ -448,32 +446,20 @@ def data_file_with_partition( ] ) - 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", - ) - ) - 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.", - ) + 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", ) - else: - fields.append(field) - - return StructType(*fields) + if field.field_id == 102 + else field + for field in DATA_FILE_TYPE[format_version].fields + ] + ) class DataFile(Record): @@ -1092,7 +1078,6 @@ class ManifestWriter(ABC): _min_sequence_number: int | None _partitions: list[Record] _compression: AvroCompressionCodec - _legacy_equality_ids: bool def __init__( self, @@ -1101,7 +1086,6 @@ def __init__( output_file: OutputFile, snapshot_id: int, avro_compression: AvroCompressionCodec, - legacy_equality_ids: bool = False, ) -> None: self.closed = False self._spec = spec @@ -1118,7 +1102,6 @@ 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.""" @@ -1164,7 +1147,6 @@ 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), - legacy_equality_ids=self._legacy_equality_ids, ) return manifest_entry_schema_with_data_file(format_version=format_version, data_file=data_file_type) @@ -1277,9 +1259,8 @@ 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, legacy_equality_ids=legacy_equality_ids) + super().__init__(spec, schema, output_file, snapshot_id, avro_compression) def content(self) -> ManifestContent: return ManifestContent.DATA @@ -1300,9 +1281,8 @@ 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, legacy_equality_ids=legacy_equality_ids) + super().__init__(spec, schema, output_file, snapshot_id, avro_compression) def content(self) -> ManifestContent: return ManifestContent.DATA @@ -1334,12 +1314,11 @@ 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, legacy_equality_ids=legacy_equality_ids) + return ManifestWriterV1(spec, schema, output_file, snapshot_id, avro_compression) elif format_version == 2: - return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression, legacy_equality_ids=legacy_equality_ids) + return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression) 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 fdc27fd90b..9624eac981 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -167,9 +167,6 @@ 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/avro/test_resolver.py b/tests/avro/test_resolver.py index fc47ffec0c..a88618ac81 100644 --- a/tests/avro/test_resolver.py +++ b/tests/avro/test_resolver.py @@ -418,3 +418,88 @@ def test_writer_missing_optional_in_read_schema() -> None: expected = StructWriter(field_writers=((None, OptionWriter(option=StringWriter())),)) assert actual == expected + + +def test_resolver_long_to_int_isolated_to_manifest_equality_ids() -> None: + from pyiceberg.manifest import DataFile + + # 1. User table schema: Long -> Int must fail even if field ID is 136 + user_file_schema = Schema(NestedField(field_id=136, name="col", field_type=LongType(), required=True)) + user_read_schema = Schema(NestedField(field_id=136, name="col", field_type=IntegerType(), required=True)) + + with pytest.raises(ResolveError, match="Cannot promote long to int"): + resolve_reader(file_schema=user_file_schema, read_schema=user_read_schema) + + # 2. Manifest DataFile schema with equality_ids (field 135, element 136): Long -> Int must succeed + manifest_file_schema = Schema( + NestedField( + field_id=2, + name="data_file", + field_type=StructType( + NestedField( + field_id=135, + name="equality_ids", + field_type=ListType(element_id=136, element_type=LongType(), element_required=True), + required=False, + ) + ), + required=True, + ) + ) + manifest_read_schema = Schema( + NestedField( + field_id=2, + name="data_file", + field_type=StructType( + NestedField( + field_id=135, + name="equality_ids", + field_type=ListType(element_id=136, element_type=IntegerType(), element_required=True), + required=False, + ) + ), + required=True, + ) + ) + + # Without read_types={2: DataFile}, it fails because it's not recognized as a manifest data_file + with pytest.raises(ResolveError, match="Cannot promote long to int"): + resolve_reader(file_schema=manifest_file_schema, read_schema=manifest_read_schema) + + # With read_types={2: DataFile}, equality_ids Long -> Int succeeds! + reader = resolve_reader( + file_schema=manifest_file_schema, + read_schema=manifest_read_schema, + read_types={2: DataFile}, + ) + assert reader is not None + + # 3. Manifest DataFile schema with other field (e.g. record_count field 103): + # Long -> Int must FAIL even with read_types={2: DataFile} + manifest_record_count_file_schema = Schema( + NestedField( + field_id=2, + name="data_file", + field_type=StructType( + NestedField(field_id=103, name="record_count", field_type=LongType(), required=True), + ), + required=True, + ) + ) + manifest_record_count_read_schema = Schema( + NestedField( + field_id=2, + name="data_file", + field_type=StructType( + NestedField(field_id=103, name="record_count", field_type=IntegerType(), required=True), + ), + required=True, + ) + ) + + with pytest.raises(ResolveError, match="Cannot promote long to int"): + resolve_reader( + file_schema=manifest_record_count_file_schema, + read_schema=manifest_record_count_read_schema, + read_types={2: DataFile}, + ) diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py index cc7de36f86..12d01349d2 100644 --- a/tests/utils/test_manifest.py +++ b/tests/utils/test_manifest.py @@ -21,13 +21,15 @@ import fastavro import pytest +from pydantic_core import to_json import pyiceberg.manifest as manifest_module -from pyiceberg.avro.codecs import AvroCompressionCodec +from pyiceberg.avro.codecs import AVRO_CODEC_KEY, AvroCompressionCodec from pyiceberg.avro.file import AvroOutputFile from pyiceberg.io import load_file_io from pyiceberg.io.pyarrow import PyArrowFileIO from pyiceberg.manifest import ( + DATA_FILE_TYPE, MANIFEST_ENTRY_SCHEMAS, MANIFEST_LIST_FILE_SCHEMAS, DataFile, @@ -41,6 +43,7 @@ _inherit_from_manifest, _manifests, clear_manifest_cache, + manifest_entry_schema_with_data_file, read_manifest_list, write_manifest, write_manifest_list, @@ -49,7 +52,7 @@ from pyiceberg.schema import Schema from pyiceberg.table.snapshots import Operation, Snapshot, Summary from pyiceberg.typedef import Record, TableVersion -from pyiceberg.types import IntegerType, NestedField +from pyiceberg.types import IntegerType, ListType, LongType, NestedField, StructType @pytest.fixture(autouse=True) @@ -1264,7 +1267,8 @@ def test_negative_manifest_cache_size_raises_value_error(monkeypatch: pytest.Mon manifest_module._ManifestCache() -def test_write_manifest_equality_ids_int_schema(tmp_path: Path) -> None: +@pytest.mark.parametrize("equality_ids", [[1, 2, 3], None, []]) +def test_write_manifest_equality_ids_int_schema(tmp_path: Path, equality_ids: list[int] | None) -> None: io = PyArrowFileIO() schema = Schema(NestedField(1, "x", IntegerType())) df = DataFile.from_args( @@ -1274,7 +1278,7 @@ def test_write_manifest_equality_ids_int_schema(tmp_path: Path) -> None: partition=(), record_count=10, file_size_in_bytes=100, - equality_ids=[1, 2, 3], + equality_ids=equality_ids, ) manifest_path = f"file://{tmp_path / 'test_equality_ids.avro'}" with write_manifest( @@ -1303,7 +1307,8 @@ def test_write_manifest_equality_ids_int_schema(tmp_path: Path) -> None: assert df_fields["equality_ids"]["type"][1]["items"] == "int" -def test_read_manifest_legacy_equality_ids_long_schema(tmp_path: Path) -> None: +@pytest.mark.parametrize("equality_ids", [[1, 2, 3], None, []]) +def test_read_manifest_legacy_equality_ids_long_schema(tmp_path: Path, equality_ids: list[int] | None) -> None: io = PyArrowFileIO() schema = Schema(NestedField(1, "x", IntegerType())) df = DataFile.from_args( @@ -1313,26 +1318,50 @@ def test_read_manifest_legacy_equality_ids_long_schema(tmp_path: Path) -> None: partition=(), record_count=10, file_size_in_bytes=100, - equality_ids=[1, 2, 3], + equality_ids=equality_ids, ) 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, + + legacy_data_file_type = StructType( + *[ + 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.", ) + if field.field_id == 135 + else field + for field in DATA_FILE_TYPE[2].fields + ] + ) + legacy_manifest_schema = manifest_entry_schema_with_data_file(2, legacy_data_file_type) + + output_file = io.new_output(manifest_path) + with AvroOutputFile[ManifestEntry]( + output_file=output_file, + file_schema=legacy_manifest_schema, + record_schema=MANIFEST_ENTRY_SCHEMAS[2], + schema_name="manifest_entry", + metadata={ + "schema": schema.model_dump_json(), + "partition-spec": to_json(UNPARTITIONED_PARTITION_SPEC.fields).decode("utf-8"), + "partition-spec-id": "0", + "format-version": "2", + AVRO_CODEC_KEY: "null", + }, + ) as writer: + writer.write_block( + [ + ManifestEntry.from_args( + status=ManifestEntryStatus.ADDED, + snapshot_id=12345, + sequence_number=1, + file_sequence_number=1, + data_file=df, + ) + ] ) mf = ManifestFile.from_args( @@ -1343,47 +1372,6 @@ def test_read_manifest_legacy_equality_ids_long_schema(tmp_path: Path) -> None: sequence_number=1, partitions=[], ) - with pytest.deprecated_call(match="Encountered non-compliant manifest with long equality_ids"): - entries = mf.fetch_manifest_entry(io) + 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" + assert entries[0].data_file.equality_ids == equality_ids