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

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
 
@@ -461,7 +462,15 @@ 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):
+                    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)
 
         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.",
         ),
@@ -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):
@@ -1058,6 +1072,7 @@ class ManifestWriter(ABC):
     _min_sequence_number: int | None
     _partitions: list[Record]
     _compression: AvroCompressionCodec
+    _legacy_equality_ids: bool
 
     def __init__(
         self,
@@ -1066,6 +1081,7 @@ def __init__(
         output_file: OutputFile,
         snapshot_id: int,
         avro_compression: AvroCompressionCodec,
+        legacy_equality_ids: bool = False,
     ) -> None:
         self.closed = False
         self._spec = spec
@@ -1082,6 +1098,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."""
@@ -1125,7 +1142,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)
 
@@ -1238,8 +1257,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
@@ -1260,8 +1280,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
@@ -1293,11 +1314,12 @@ 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}")
 

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

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()):