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

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)
 

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

@@ -1188,3 +1188,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]