Write equality_ids as list<int> and keep reading legacy list<long> manifests (#3840)

The Iceberg spec defines data_file.equality_ids as list<int>, but PyIceberg
wrote list<long>, which strict readers such as iceberg-cpp reject. Fix the
manifest schema so new manifests follow the spec, and let the Avro resolver
read the legacy long element as int for that field only, so existing
manifests stay readable without loosening type promotion anywhere else.

Closes #3840.

Co-authored-by: Kevin Liu <kevin.jq.liu@gmail.com>

Author: hedger9487

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]