Tolerate legacy long equality_ids in resolver and write int in manifest (#3840)

Author: hedger9487

pyiceberg/avro/resolver.py

@@ -14,8 +14,6 @@
 # KIND, either express or implied.  See the License for the
 # 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
 
@@ -379,6 +377,9 @@ def visit_geography(self, geography_type: "GeographyType", partner: IcebergType
         return BinaryWriter()
 
 
+_MANIFEST_DATA_FILE_EQUALITY_IDS_ELEMENT_ID = 136
+
+
 class ReadSchemaResolver(PrimitiveWithPartnerVisitor[IcebergType, Reader]):
     __slots__ = ("read_types", "read_enums", "context")
     read_types: dict[int, Callable[..., StructProtocol]]
@@ -462,13 +463,14 @@ def primitive(self, primitive: PrimitiveType, expected_primitive: IcebergType |
 
             # ensure that the type can be projected to the expected
             if 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,
-                    )
+                is_manifest_data_file = getattr(self.read_types.get(2), "__name__", "") == "DataFile"
+                if (
+                    is_manifest_data_file
+                    and _MANIFEST_DATA_FILE_EQUALITY_IDS_ELEMENT_ID in self.context
+                    and isinstance(primitive, LongType)
+                    and isinstance(expected_primitive, IntegerType)
+                ):
+                    pass
                 else:
                     promote(primitive, expected_primitive)
 

pyiceberg/manifest.py

@@ -433,9 +433,7 @@ def __repr__(self) -> str:
 }
 
 
-def data_file_with_partition(
-    partition_type: StructType, format_version: TableVersion, legacy_equality_ids: bool = False
-) -> StructType:
+def data_file_with_partition(partition_type: StructType, format_version: TableVersion) -> StructType:
     data_file_partition_type = StructType(
         *[
             NestedField(
@@ -448,32 +446,20 @@ def data_file_with_partition(
         ]
     )
 
-    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",
-                )
-            )
-        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.",
-                )
+    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",
             )
-        else:
-            fields.append(field)
-
-    return StructType(*fields)
+            if field.field_id == 102
+            else field
+            for field in DATA_FILE_TYPE[format_version].fields
+        ]
+    )
 
 
 class DataFile(Record):
@@ -1072,7 +1058,6 @@ class ManifestWriter(ABC):
     _min_sequence_number: int | None
     _partitions: list[Record]
     _compression: AvroCompressionCodec
-    _legacy_equality_ids: bool
 
     def __init__(
         self,
@@ -1081,7 +1066,6 @@ def __init__(
         output_file: OutputFile,
         snapshot_id: int,
         avro_compression: AvroCompressionCodec,
-        legacy_equality_ids: bool = False,
     ) -> None:
         self.closed = False
         self._spec = spec
@@ -1098,7 +1082,6 @@ 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."""
@@ -1144,7 +1127,6 @@ 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),
-            legacy_equality_ids=self._legacy_equality_ids,
         )
         return manifest_entry_schema_with_data_file(format_version=format_version, data_file=data_file_type)
 
@@ -1257,9 +1239,8 @@ 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, legacy_equality_ids=legacy_equality_ids)
+        super().__init__(spec, schema, output_file, snapshot_id, avro_compression)
 
     def content(self) -> ManifestContent:
         return ManifestContent.DATA
@@ -1280,9 +1261,8 @@ 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, legacy_equality_ids=legacy_equality_ids)
+        super().__init__(spec, schema, output_file, snapshot_id, avro_compression)
 
     def content(self) -> ManifestContent:
         return ManifestContent.DATA
@@ -1314,12 +1294,11 @@ 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, legacy_equality_ids=legacy_equality_ids)
+        return ManifestWriterV1(spec, schema, output_file, snapshot_id, avro_compression)
     elif format_version == 2:
-        return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression, legacy_equality_ids=legacy_equality_ids)
+        return ManifestWriterV2(spec, schema, output_file, snapshot_id, avro_compression)
     else:
         raise ValueError(f"Cannot write manifest for table version: {format_version}")
 

pyiceberg/table/__init__.py

@@ -167,9 +167,6 @@ 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/avro/test_resolver.py

@@ -409,3 +409,88 @@ def test_writer_missing_optional_in_read_schema() -> None:
     expected = StructWriter(field_writers=((None, OptionWriter(option=StringWriter())),))
 
     assert actual == expected
+
+
+def test_resolver_long_to_int_isolated_to_manifest_equality_ids() -> None:
+    from pyiceberg.manifest import DataFile
+
+    # 1. User table schema: Long -> Int must fail even if field ID is 136
+    user_file_schema = Schema(NestedField(field_id=136, name="col", field_type=LongType(), required=True))
+    user_read_schema = Schema(NestedField(field_id=136, name="col", field_type=IntegerType(), required=True))
+
+    with pytest.raises(ResolveError, match="Cannot promote long to int"):
+        resolve_reader(file_schema=user_file_schema, read_schema=user_read_schema)
+
+    # 2. Manifest DataFile schema with equality_ids (field 135, element 136): Long -> Int must succeed
+    manifest_file_schema = Schema(
+        NestedField(
+            field_id=2,
+            name="data_file",
+            field_type=StructType(
+                NestedField(
+                    field_id=135,
+                    name="equality_ids",
+                    field_type=ListType(element_id=136, element_type=LongType(), element_required=True),
+                    required=False,
+                )
+            ),
+            required=True,
+        )
+    )
+    manifest_read_schema = Schema(
+        NestedField(
+            field_id=2,
+            name="data_file",
+            field_type=StructType(
+                NestedField(
+                    field_id=135,
+                    name="equality_ids",
+                    field_type=ListType(element_id=136, element_type=IntegerType(), element_required=True),
+                    required=False,
+                )
+            ),
+            required=True,
+        )
+    )
+
+    # Without read_types={2: DataFile}, it fails because it's not recognized as a manifest data_file
+    with pytest.raises(ResolveError, match="Cannot promote long to int"):
+        resolve_reader(file_schema=manifest_file_schema, read_schema=manifest_read_schema)
+
+    # With read_types={2: DataFile}, equality_ids Long -> Int succeeds!
+    reader = resolve_reader(
+        file_schema=manifest_file_schema,
+        read_schema=manifest_read_schema,
+        read_types={2: DataFile},
+    )
+    assert reader is not None
+
+    # 3. Manifest DataFile schema with other field (e.g. record_count field 103):
+    # Long -> Int must FAIL even with read_types={2: DataFile}
+    manifest_record_count_file_schema = Schema(
+        NestedField(
+            field_id=2,
+            name="data_file",
+            field_type=StructType(
+                NestedField(field_id=103, name="record_count", field_type=LongType(), required=True),
+            ),
+            required=True,
+        )
+    )
+    manifest_record_count_read_schema = Schema(
+        NestedField(
+            field_id=2,
+            name="data_file",
+            field_type=StructType(
+                NestedField(field_id=103, name="record_count", field_type=IntegerType(), required=True),
+            ),
+            required=True,
+        )
+    )
+
+    with pytest.raises(ResolveError, match="Cannot promote long to int"):
+        resolve_reader(
+            file_schema=manifest_record_count_file_schema,
+            read_schema=manifest_record_count_read_schema,
+            read_types={2: DataFile},
+        )