fix(avro): keep the table readable after dropping an old partition column

After a partition field is removed and its source column is dropped,
the old spec's partition field has the unknown type. Rewriting its
manifest wrapped the Avro "null" type in another null union, which is
invalid Avro and made the table unreadable. Write unknown fields as a
plain "null", like Java's AvroSchemaUtil.toOption.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: hzmsh

pyiceberg/avro/resolver.py

@@ -147,7 +147,7 @@ def struct(self, struct: StructType, field_results: list[Writer]) -> Writer:
         return StructWriter(tuple((pos, result) for pos, result in enumerate(field_results)))
 
     def field(self, field: NestedField, field_result: Writer) -> Writer:
-        return field_result if field.required else OptionWriter(field_result)
+        return field_result if field.required or isinstance(field.field_type, UnknownType) else OptionWriter(field_result)
 
     def list(self, list_type: ListType, element_result: Writer) -> Writer:
         return ListWriter(element_result)
@@ -302,7 +302,9 @@ def struct(self, file_schema: StructType, record_struct: IcebergType | None, fil
         return StructWriter(field_writers=tuple(results))
 
     def field(self, file_field: NestedField, record_type: IcebergType | None, field_writer: Writer) -> Writer:
-        return field_writer if file_field.required else OptionWriter(field_writer)
+        return (
+            field_writer if file_field.required or isinstance(file_field.field_type, UnknownType) else OptionWriter(field_writer)
+        )
 
     def list(self, file_list_type: ListType, file_list: IcebergType | None, element_writer: Writer) -> Writer:
         return ListWriter(element_writer if file_list_type.element_required else OptionWriter(element_writer))

pyiceberg/utils/schema_conversion.py

@@ -534,7 +534,8 @@ def field(self, field: NestedField, field_result: AvroType) -> AvroType:
         result = {
             "name": sanitized_name,
             FIELD_ID_PROP: field.field_id,
-            "type": field_result if field.required else ["null", field_result],
+            # Unknown maps to Avro "null", and a union cannot contain "null" twice
+            "type": field_result if field.required or isinstance(field.field_type, UnknownType) else ["null", field_result],
         }
 
         if original_name != sanitized_name:

tests/avro/test_file.py

@@ -54,6 +54,7 @@
     TimestampType,
     TimestamptzType,
     TimeType,
+    UnknownType,
     UUIDType,
 )
 from pyiceberg.utils.schema_conversion import AvroSchemaConversion
@@ -544,3 +545,23 @@ def test_avro_output_file_tell_includes_buffered_records() -> None:
             out.write_block([manifest_entry(0)])
 
             assert out.tell() > after_header
+
+
+def test_write_optional_unknown_type() -> None:
+    # A partition field whose source column was dropped has the unknown type
+    schema = Schema(
+        NestedField(field_id=1, name="before", field_type=StringType(), required=False),
+        NestedField(field_id=2, name="dropped", field_type=UnknownType(), required=False),
+        NestedField(field_id=3, name="after", field_type=LongType(), required=True),
+    )
+
+    with TemporaryDirectory() as tmpdir:
+        tmp_avro_file = tmpdir + "/unknown.avro"
+        with avro.AvroOutputFile[Record](PyArrowFileIO().new_output(tmp_avro_file), schema, "unknown") as out:
+            out.write_block([Record("hello", None, 42)])
+
+        with avro.AvroFile[Record](PyArrowFileIO().new_input(tmp_avro_file), schema, {-1: Record}) as avro_reader:
+            assert list(avro_reader) == [Record("hello", None, 42)]
+
+        with open(tmp_avro_file, "rb") as fo:
+            assert list(reader(fo=fo)) == [{"before": "hello", "dropped": None, "after": 42}]

tests/catalog/test_catalog_behaviors.py

@@ -848,6 +848,28 @@ def test_write_and_evolve(catalog: Catalog, format_version: int) -> None:
         txn.delete("foo = 'a'")
 
 
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_delete_after_dropping_old_partition_source_column(catalog: Catalog, format_version: int) -> None:
+    identifier = f"default.delete_after_dropping_old_partition_source_v{format_version}"
+    catalog.create_namespace_if_not_exists("default")
+
+    pa_table = pa.Table.from_pydict({"id": [1, 2], "category": ["a", "b"]})
+    tbl = catalog.create_table(identifier=identifier, schema=pa_table.schema, properties={"format-version": str(format_version)})
+    with tbl.update_spec() as spec_update:
+        spec_update.add_identity("category")
+    tbl.append(pa_table)
+
+    with tbl.update_spec() as spec_update:
+        spec_update.remove_field("category")
+    with tbl.update_schema() as schema_update:
+        schema_update.delete_column("category")
+
+    # Rewrites the manifest of the old spec, whose partition source column is gone
+    tbl.delete("id == 1")
+
+    assert catalog.load_table(identifier).scan().to_arrow()["id"].to_pylist() == [2]
+
+
 # Merge manifests
 @pytest.mark.parametrize("format_version", [1, 2])
 def test_merge_manifests_local_file_system(catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int) -> None: