Avro: Cache Avro schema conversion in AvroFileHeader.get_schema (#3662)

pyiceberg/avro/file.py

@@ -25,6 +25,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 from enum import Enum
+from functools import lru_cache
 from types import TracebackType
 from typing import (
     Generic,
@@ -68,6 +69,12 @@
 _SCHEMA_KEY = "avro.schema"
 
 
+@lru_cache(maxsize=128)
+def _parse_avro_schema(avro_schema_string: str) -> Schema:
+    avro_schema = json.loads(avro_schema_string)
+    return AvroSchemaConversion().avro_to_iceberg(avro_schema)
+
+
 class AvroFileHeader(Record):
     @property
     def magic(self) -> bytes:
@@ -97,9 +104,7 @@ def compression_codec(self) -> type[Codec] | None:
 
     def get_schema(self) -> Schema:
         if _SCHEMA_KEY in self.meta:
-            avro_schema_string = self.meta[_SCHEMA_KEY]
-            avro_schema = json.loads(avro_schema_string)
-            return AvroSchemaConversion().avro_to_iceberg(avro_schema)
+            return _parse_avro_schema(self.meta[_SCHEMA_KEY])
         else:
             raise ValueError("No schema found in Avro file headers")
 

tests/avro/test_file.py

@@ -87,6 +87,18 @@ def test_missing_schema() -> None:
     assert "No schema found in Avro file headers" in str(exc_info.value)
 
 
+def test_get_schema_is_cached() -> None:
+    schema_json = '{"type": "record", "name": "r", "fields": [{"name": "id", "type": "int", "field-id": 1}]}'
+    header1 = AvroFileHeader(bytes(0), {"avro.schema": schema_json}, bytes(16))
+    header2 = AvroFileHeader(bytes(0), {"avro.schema": schema_json}, bytes(16))
+
+    schema1 = header1.get_schema()
+    schema2 = header2.get_schema()
+
+    assert schema1 == schema2
+    assert schema1 is schema2
+
+
 # helper function to serialize our objects to dicts to enable
 # direct comparison with the dicts returned by fastavro
 def todict(obj: Any) -> Any: