AvroWriter Issue: Incorrect iceberg_to_avro Schema Conversion for Decimal, Fixed, and UUID

[HonahX]

Apache Iceberg version

main (development)

Please describe the bug 🐞

DecimalType

Currently, if we use AvroOutputFile to write decimal values to an avro file, sometimes the result file cannot be successfully read by other avro reader such as fastavro.

The cause is that when encoding and writing decimal value, we obey the iceberg specification and write it as a fixed. However, in the avro file's schema, the current implementation specify it as a variable-length binary:

iceberg-python/pyiceberg/utils/schema_conversion.py

Lines 569 to 572 in 553695e

	def visit_decimal(self, decimal_type: DecimalType) -> AvroType:
	return {"type": "bytes", "logicalType": "decimal", "precision": decimal_type.precision, "scale": decimal_type.scale}

I think this should be changed to

    def visit_decimal(self, decimal_type: DecimalType) -> AvroType:
        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}"
}

So that other avro reader can correctly interpret the encoded value as fixed-length bytes instead of trying to read the length.

I think this is also the root cause of the failure I observed when using ManifestWriter to write manifest entry for table partitioned by decimalType col. I ran some local test and verified that the above change could fix this issue.

FixedType and UUIDType

For Fixed and UUID, I think the current conversion miss the required name field: https://avro.apache.org/docs/1.11.1/specification/#fixed

iceberg-python/pyiceberg/utils/schema_conversion.py

Lines 567 to 568 in 553695e

	def visit_fixed(self, fixed_type: FixedType) -> AvroType:
	return {"type": "fixed", "size": len(fixed_type)}

iceberg-python/pyiceberg/utils/schema_conversion.py

Lines 605 to 606 in 553695e

	def visit_uuid(self, uuid_type: UUIDType) -> AvroType:
	return {"type": "fixed", "size": "16", "logicalType": "uuid"}

The fastavro will complain

fastavro._schema_common.SchemaParseException: "name" is a required field missing from the schema: {'type': 'fixed', 'size': 16}

Once we fix these, I think the ManifestWriter should work with all types of partition values

[HonahX]

For refernce, I used the following code (added in test/avro/test_file.py) to reproduce the decimal error

from fastavro import reader, writer
    inner_struct_type = StructType(fields=[NestedField(field_id=1000, name='decimalCol', field_type=DecimalType(precision=10, scale=2), required=True)])
    schema = Schema(
        NestedField(
            field_id=102,
            name="partition",
            field_type=inner_struct_type,
            required=True,
            doc="Partition data tuple, schema based on the partition spec",
        ),
        NestedField(field_id=103, name="record_count", field_type=LongType(), required=True,
                    doc="Number of records in the file"),
    )

    class InnerDecimalRecord(Record):
        decimalCol: Decimal

        def __init__(self, *data: Any, **named_data: Any) -> None:
            super().__init__(*data, **{"struct": inner_struct_type, **named_data})

    class NestedRecord(Record):
        partition: InnerDecimalRecord
        record_count: int

        def __init__(self, *data: Any, **named_data: Any) -> None:
            super().__init__(*data, **{"struct": schema.as_struct(), **named_data})

    record = NestedRecord(
        partition=InnerDecimalRecord(Decimal("0.00")),
        record_count=131327,
    )

    with TemporaryDirectory() as tmpdir:
        tmp_avro_file = tmpdir + "/nested_decimal.avro"
        # write to disk
        with avro.AvroOutputFile[NestedRecord](
            PyArrowFileIO().new_output(tmp_avro_file), schema, "nested_decimal_schema"
        ) as out:
            out.write_block([record])
            assert record == avro_entry

        # read from disk using fastavro
        with open(tmp_avro_file, "rb") as fo:
            r = reader(fo=fo)

            it = iter(r)

            fa_entry = next(it)

        assert todict(record) == fa_entry