Manifest: Use int for equality_ids in manifest schema per Iceberg spec (#3840)

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<long>, for equality_ids; the Iceberg spec requires list<int>.
+        # The default schema now uses list<int>, 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()
 

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.",
         ),

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<long> can still be read.
+
+    PyIceberg previously wrote the wrong schema, list<long>, for equality_ids; the Iceberg spec requires list<int>.
+    The default schema now uses list<int>, 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<long>
+    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]