Avro: name fixed types by field-id instead of tracking defined names

Review feedback: a fixed, decimal or uuid type now gets the field-id of
the field, list element or map key/value that uses it in its name, the
same way records are named r<field-id>. Field-ids are unique, so every
definition has its own name and the writer no longer needs to track which
names it has defined. AVRO_NAMED_TYPES is a frozenset.

Author: ozenyunus35

pyiceberg/utils/schema_conversion.py

@@ -82,7 +82,7 @@
 
 
 # Avro types that have a name and can be referred to by it later in the schema
-AVRO_NAMED_TYPES = {"record", "enum", "fixed"}
+AVRO_NAMED_TYPES = frozenset({"record", "enum", "fixed"})
 
 
 class AvroSchemaConversion:
@@ -513,7 +513,6 @@ class ConvertSchemaToAvro(SchemaVisitorPerPrimitiveType[AvroType]):
     last_list_field_id: int
     last_map_key_field_id: int
     last_map_value_field_id: int
-    defined_names: set[str]
 
     def __init__(self, schema_name: str | None) -> None:
         """Convert an Iceberg schema to an Avro schema.
@@ -522,18 +521,16 @@ def __init__(self, schema_name: str | None) -> None:
             schema_name: The name of the root record.
         """
         self.schema_name = schema_name
-        self.defined_names: set[str] = set()
 
-    def _define_once(self, named_type: dict[str, Any]) -> AvroType:
-        """Return the definition of a named type the first time, and a reference to its name after that.
+    @staticmethod
+    def _name_fixed(result: AvroType, field_id: int) -> None:
+        """Add the field-id to the name of a fixed type.
 
-        Avro allows only one definition per name; later uses must refer to it by name.
+        Avro allows only one definition per name, and a decimal, uuid or fixed type can be used by more
+        than one field. Field-ids are unique, so the names stay unique, the same way records are named.
         """
-        name = named_type["name"]
-        if name in self.defined_names:
-            return name
-        self.defined_names.add(name)
-        return named_type
+        if isinstance(result, dict) and result.get("type") == "fixed":
+            result["name"] = f"{result['name']}_{field_id}"
 
     def schema(self, schema: Schema, struct_result: AvroType) -> AvroType:
         if isinstance(struct_result, dict) and self.schema_name is not None:
@@ -556,6 +553,7 @@ def field(self, field: NestedField, field_result: AvroType) -> AvroType:
         # Sets the schema name
         if isinstance(field_result, dict) and field_result.get("type") == "record":
             field_result["name"] = f"r{field.field_id}"
+        self._name_fixed(field_result, field.field_id)
 
         original_name = field.name
         sanitized_name = make_compatible_name(original_name)
@@ -583,9 +581,12 @@ def list(self, list_type: ListType, element_result: AvroType) -> AvroType:
         # Sets the schema name in case of a record
         if isinstance(element_result, dict) and element_result.get("type") == "record":
             element_result["name"] = f"r{self.last_list_field_id}"
+        self._name_fixed(element_result, self.last_list_field_id)
         return {"type": "array", "element-id": self.last_list_field_id, "items": element_result}
 
     def map(self, map_type: MapType, key_result: AvroType, value_result: AvroType) -> AvroType:
+        self._name_fixed(key_result, self.last_map_key_field_id)
+        self._name_fixed(value_result, self.last_map_value_field_id)
         if isinstance(key_result, StringType):
             # Avro Maps does not support other keys than a String,
             return {
@@ -611,19 +612,17 @@ def map(self, map_type: MapType, key_result: AvroType, value_result: AvroType) -
             }
 
     def visit_fixed(self, fixed_type: FixedType) -> AvroType:
-        return self._define_once({"type": "fixed", "size": len(fixed_type), "name": f"fixed_{len(fixed_type)}"})
+        return {"type": "fixed", "size": len(fixed_type), "name": f"fixed_{len(fixed_type)}"}
 
     def visit_decimal(self, decimal_type: DecimalType) -> AvroType:
-        return self._define_once(
-            {
-                "type": "fixed",
-                "size": decimal_required_bytes(decimal_type.precision),
-                "logicalType": "decimal",
-                "precision": decimal_type.precision,
-                "scale": decimal_type.scale,
-                "name": f"decimal_{decimal_type.precision}_{decimal_type.scale}",
-            }
-        )
+        return {
+            "type": "fixed",
+            "size": decimal_required_bytes(decimal_type.precision),
+            "logicalType": "decimal",
+            "precision": decimal_type.precision,
+            "scale": decimal_type.scale,
+            "name": f"decimal_{decimal_type.precision}_{decimal_type.scale}",
+        }
 
     def visit_boolean(self, boolean_type: BooleanType) -> AvroType:
         return "boolean"
@@ -662,7 +661,7 @@ def visit_string(self, string_type: StringType) -> AvroType:
         return "string"
 
     def visit_uuid(self, uuid_type: UUIDType) -> AvroType:
-        return self._define_once({"type": "fixed", "size": 16, "logicalType": "uuid", "name": "uuid_fixed"})
+        return {"type": "fixed", "size": 16, "logicalType": "uuid", "name": "uuid_fixed"}
 
     def visit_binary(self, binary_type: BinaryType) -> AvroType:
         return "bytes"

tests/avro/test_file.py

@@ -456,7 +456,7 @@ def field_uuid(self) -> UUID:
 
 
 def test_repeated_named_types_round_trip() -> None:
-    """Named types used more than once are defined once in the file schema and read back by both readers"""
+    """Named types used by more than one field get unique names in the file schema and read back with both readers"""
     schema = Schema(
         NestedField(field_id=1, name="price", field_type=DecimalType(10, 2), required=True),
         NestedField(field_id=2, name="cost", field_type=DecimalType(10, 2), required=False),
@@ -485,8 +485,8 @@ def test_repeated_named_types_round_trip() -> None:
             file_fields = r.writer_schema["fields"]
             avro_entry_read_with_fastavro = list(next(iter(r)).values())
 
-    assert file_fields[1]["type"] == ["null", "decimal_10_2"]
-    assert file_fields[3]["type"] == ["null", "uuid_fixed"]
+    assert [f["name"] for f in (file_fields[0]["type"], file_fields[1]["type"][1])] == ["decimal_10_2_1", "decimal_10_2_2"]
+    assert [f["name"] for f in (file_fields[2]["type"], file_fields[3]["type"][1])] == ["uuid_fixed_3", "uuid_fixed_4"]
     for idx in range(len(record)):
         assert record[idx] == avro_entry[idx]
     # fastavro returns a uuid stored as fixed as its raw bytes

tests/utils/test_schema_conversion.py

@@ -445,7 +445,8 @@ def test_avro_to_iceberg_unknown_named_type_reference() -> None:
         AvroSchemaConversion().avro_to_iceberg(avro_schema)
 
 
-def test_iceberg_to_avro_defines_named_types_once() -> None:
+def test_iceberg_to_avro_names_fixed_types_by_field_id() -> None:
+    """A named type can be defined once per schema, so fixed types used by several fields get the field-id in their name"""
     schema = Schema(
         NestedField(field_id=1, name="a", field_type=DecimalType(10, 2), required=False),
         NestedField(field_id=2, name="b", field_type=DecimalType(10, 2), required=True),
@@ -460,22 +461,35 @@ def test_iceberg_to_avro_defines_named_types_once() -> None:
             required=True,
         ),
         NestedField(field_id=7, name="f", field_type=UUIDType(), required=False),
-        NestedField(field_id=8, name="g", field_type=DecimalType(9, 2), required=True),
+        NestedField(
+            field_id=8,
+            name="g",
+            field_type=ListType(element_id=9, element_type=UUIDType(), element_required=True),
+            required=True,
+        ),
+        NestedField(
+            field_id=10,
+            name="h",
+            field_type=MapType(key_id=11, key_type=UUIDType(), value_id=12, value_type=DecimalType(10, 2)),
+            required=True,
+        ),
         schema_id=1,
     )
 
     avro_schema = AvroSchemaConversion().iceberg_to_avro(schema, schema_name="r102")
     assert isinstance(avro_schema, dict)
     fields = avro_schema["fields"]
     nested_fields = fields[3]["type"]["fields"]
-
-    assert fields[0]["type"] == ["null", DECIMAL_10_2]
-    assert fields[1]["type"] == "decimal_10_2"
-    assert fields[2]["type"] == {"type": "fixed", "size": 3, "name": "fixed_3"}
-    assert nested_fields[0]["type"] == ["null", "fixed_3"]
-    assert nested_fields[1]["type"] == {"type": "fixed", "size": 16, "logicalType": "uuid", "name": "uuid_fixed"}
-    assert fields[4]["type"] == ["null", "uuid_fixed"]
-    # a different decimal is a different name, so it gets its own definition
-    assert fields[5]["type"]["name"] == "decimal_9_2"
-    # and the references read back as the same types
+    map_fields = fields[6]["type"]["items"]["fields"]
+
+    assert fields[0]["type"] == ["null", {**DECIMAL_10_2, "name": "decimal_10_2_1"}]
+    assert fields[1]["type"] == {**DECIMAL_10_2, "name": "decimal_10_2_2"}
+    assert fields[2]["type"] == {"type": "fixed", "size": 3, "name": "fixed_3_3"}
+    assert nested_fields[0]["type"] == ["null", {"type": "fixed", "size": 3, "name": "fixed_3_5"}]
+    assert nested_fields[1]["type"] == {"type": "fixed", "size": 16, "logicalType": "uuid", "name": "uuid_fixed_6"}
+    assert fields[4]["type"] == ["null", {"type": "fixed", "size": 16, "logicalType": "uuid", "name": "uuid_fixed_7"}]
+    assert fields[5]["type"]["items"]["name"] == "uuid_fixed_9"
+    assert map_fields[0]["type"]["name"] == "uuid_fixed_11"
+    assert map_fields[1]["type"]["name"] == "decimal_10_2_12"
+    # and the schema reads back as the same types
     assert AvroSchemaConversion().avro_to_iceberg(avro_schema) == schema