diff --git a/pyiceberg/avro/resolver.py b/pyiceberg/avro/resolver.py index 81b573aa79..ec76b331aa 100644 --- a/pyiceberg/avro/resolver.py +++ b/pyiceberg/avro/resolver.py @@ -461,10 +461,25 @@ def primitive(self, primitive: PrimitiveType, expected_primitive: IcebergType | # ensure that the type can be projected to the expected if primitive != expected_primitive: + if self._is_legacy_long_equality_ids(primitive, expected_primitive): + return IntegerReader() promote(primitive, expected_primitive) return super().primitive(primitive, expected_primitive) + def _is_legacy_long_equality_ids(self, primitive: PrimitiveType, expected_primitive: PrimitiveType) -> bool: + # PyIceberg previously wrote the wrong schema, list, for equality_ids; the Iceberg spec requires list. + # The default schema now uses list, but the schema promotion rules do not allow reading long as int. + # Ints and longs share the same Avro encoding, so allow this exact mismatch for the equality_ids element (136) + # when reading existing files. + # See: https://github.com/apache/iceberg-python/issues/3840 + # See: https://iceberg.apache.org/spec/#manifests + return ( + self.context == [2, 135, 136] # field id path from manifest_entry: data_file (2), equality_ids (135), element (136) + and isinstance(primitive, LongType) + and isinstance(expected_primitive, IntegerType) + ) + def visit_boolean(self, boolean_type: BooleanType, partner: IcebergType | None) -> Reader: return BooleanReader() diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 7037c69437..2b54fff509 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 ff91224e4d..331146346e 100644 --- a/tests/utils/test_manifest.py +++ b/tests/utils/test_manifest.py @@ -25,10 +25,11 @@ import pyiceberg.manifest as manifest_module from pyiceberg.avro.codecs import AvroCompressionCodec -from pyiceberg.avro.file import AvroOutputFile +from pyiceberg.avro.file import AvroFile, 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, @@ -50,7 +51,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) @@ -294,6 +295,57 @@ def write_and_read(file_name: str, data_file: DataFile) -> DataFile: assert delete_file.content_size_in_bytes == 46 +def test_read_legacy_long_equality_ids(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Manifests written by older PyIceberg versions with equality_ids as list can still be read. + + PyIceberg previously wrote the wrong schema, list, for equality_ids; the Iceberg spec requires list. + The default schema now uses list, so reading existing manifests relies on the fallback in the Avro resolver. + See: https://github.com/apache/iceberg-python/issues/3840 + See: https://iceberg.apache.org/spec/#manifests + """ + io = PyArrowFileIO() + manifest_path = str(tmp_path / "manifest.avro") + + entry = ManifestEntry.from_args( + status=ManifestEntryStatus.ADDED, + snapshot_id=1, + sequence_number=1, + file_sequence_number=1, + data_file=DataFile.from_args( + content=DataFileContent.EQUALITY_DELETES, + file_path="s3://bucket/deletes.parquet", + file_format=FileFormat.PARQUET, + partition=Record(), + record_count=10, + file_size_in_bytes=1024, + equality_ids=[1, 2], + ), + ) + + # Write the manifest as older PyIceberg versions did, with equality_ids as list + legacy_data_file_type = StructType( + *[ + NestedField(135, "equality_ids", ListType(136, LongType()), required=False) if field.field_id == 135 else field + for field in DATA_FILE_TYPE[2].fields + ] + ) + with monkeypatch.context() as legacy: + legacy.setitem(DATA_FILE_TYPE, 2, legacy_data_file_type) + with write_manifest( + format_version=2, + spec=UNPARTITIONED_PARTITION_SPEC, + schema=Schema(NestedField(1, "foo", IntegerType(), False)), + output_file=io.new_output(manifest_path), + snapshot_id=1, + avro_compression="null", + ) as writer: + writer.add_entry(entry) + with AvroFile[ManifestEntry](io.new_input(manifest_path)) as avro_file: + assert avro_file.schema.find_field("data_file.equality_ids").field_type == ListType(136, LongType()) + + assert writer.to_manifest_file().fetch_manifest_entry(io)[0].data_file.equality_ids == [1, 2] + + def test_read_manifest_list(generated_manifest_file_file_v1: str) -> None: input_file = PyArrowFileIO().new_input(generated_manifest_file_file_v1) manifest_list = list(read_manifest_list(input_file))[0]