Skip unknown nested fields when resolving Avro read schemas

pyiceberg/avro/resolver.py

@@ -548,6 +548,10 @@ def schema_partner(self, partner: IcebergType | None) -> IcebergType | None:
         raise ResolveError(f"File/read schema are not aligned for schema, got {partner}")
 
     def field_partner(self, partner: IcebergType | None, field_id: int, field_name: str) -> IcebergType | None:
+        # A file-only field has no partner; its nested fields are read and dropped
+        if partner is None:
+            return None
+
         if isinstance(partner, StructType):
             field = partner.field(field_id)
         else:
@@ -556,18 +560,27 @@ def field_partner(self, partner: IcebergType | None, field_id: int, field_name:
         return field.field_type if field else None
 
     def list_element_partner(self, partner_list: IcebergType | None) -> IcebergType | None:
+        if partner_list is None:
+            return None
+
         if isinstance(partner_list, ListType):
             return partner_list.element_type
 
         raise ResolveError(f"File/read schema are not aligned for list, got {partner_list}")
 
     def map_key_partner(self, partner_map: IcebergType | None) -> IcebergType | None:
+        if partner_map is None:
+            return None
+
         if isinstance(partner_map, MapType):
             return partner_map.key_type
 
         raise ResolveError(f"File/read schema are not aligned for map, got {partner_map}")
 
     def map_value_partner(self, partner_map: IcebergType | None) -> IcebergType | None:
+        if partner_map is None:
+            return None
+
         if isinstance(partner_map, MapType):
             return partner_map.value_type
 

tests/avro/test_file.py

@@ -41,16 +41,20 @@
 from pyiceberg.schema import Schema
 from pyiceberg.typedef import Record, TableVersion
 from pyiceberg.types import (
+    BinaryType,
     BooleanType,
     DateType,
     DecimalType,
     DoubleType,
     FixedType,
     FloatType,
     IntegerType,
+    ListType,
     LongType,
+    MapType,
     NestedField,
     StringType,
+    StructType,
     TimestampNanoType,
     TimestampType,
     TimestamptzNanoType,
@@ -341,6 +345,63 @@ def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: Ta
                 raise ValueError(f"Unsupported version: {format_version}")
 
 
+def test_read_manifest_entry_with_unknown_nested_fields() -> None:
+    # Writers may add fields to data_file that the read schema does not know; they must be skipped by field ID
+    read_schema = MANIFEST_ENTRY_SCHEMAS[2]
+    data_file_fields = list(read_schema.find_field("data_file").field_type.fields)
+    unknown_fields = [
+        NestedField(1250, "unknown_map", MapType(1260, IntegerType(), 1270, BinaryType()), required=False),
+        NestedField(1280, "unknown_list", ListType(1290, LongType()), required=False),
+        NestedField(
+            1300,
+            "unknown_struct",
+            StructType(NestedField(1310, "inner_map", MapType(1320, IntegerType(), 1330, StringType()), required=False)),
+            required=False,
+        ),
+    ]
+    # Place the unknown fields before known ones to check that the decoder skips their bytes
+    pos = next(i for i, field in enumerate(data_file_fields) if field.name == "lower_bounds")
+    file_data_file = StructType(*data_file_fields[:pos], *unknown_fields, *data_file_fields[pos:])
+    file_schema = Schema(
+        *[
+            NestedField(field.field_id, field.name, file_data_file, required=field.required)
+            if field.name == "data_file"
+            else field
+            for field in read_schema.fields
+        ]
+    )
+
+    data_file = DataFile.from_args(
+        content=DataFileContent.DATA,
+        file_path="s3://some-path/some-file.parquet",
+        file_format=FileFormat.PARQUET,
+        partition=Record(),
+        record_count=131327,
+        file_size_in_bytes=220669226,
+        lower_bounds={1: b"aaaaaaaaaaaaaaaa"},
+        upper_bounds={1: b"zzzzzzzzzzzzzzzz"},
+        sort_order_id=4,
+    )
+    entry = ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, snapshot_id=8638475580105682862, data_file=data_file)
+
+    record = todict(entry)
+    record["data_file"]["unknown_map"] = todict({1: b"\x01\x02"})
+    record["data_file"]["unknown_list"] = [1, 2, 3]
+    record["data_file"]["unknown_struct"] = {"inner_map": todict({2: "value"})}
+
+    with TemporaryDirectory() as tmpdir:
+        tmp_avro_file = tmpdir + "/manifest_entry.avro"
+        with open(tmp_avro_file, "wb") as out:
+            writer(out, AvroSchemaConversion().iceberg_to_avro(file_schema, schema_name="manifest_entry"), [record])
+
+        with avro.AvroFile[ManifestEntry](
+            input_file=PyArrowFileIO().new_input(tmp_avro_file),
+            read_schema=read_schema,
+            read_types={-1: ManifestEntry, 2: DataFile},
+        ) as avro_reader:
+            assert list(avro_reader) == [entry]
+
+
 @pytest.mark.parametrize("is_required", [True, False])
 def test_all_primitive_types(is_required: bool) -> None:
     all_primitives_schema = Schema(