fix issue where writer and reader schemas would collide
MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
@@ -556,7 +556,7 @@ cpdef read_union(
elif return_record_name and extract_record_type(idx_schema) not in AVRO_TYPES:
# idx_schema is a named type
return (
- named_schemas[idx_schema]["name"],
+ named_schemas["writer"][idx_schema]["name"],
_read_data(
fo, idx_schema, named_schemas, None, return_record_name, return_record_name_override
)
@@ -688,11 +688,6 @@ cpdef _read_data(
record_type = extract_record_type(writer_schema)
if reader_schema:
- # If the schemas are the same, set the reader schema to None so that no
- # schema resolution is done for this call or future recursive calls
- if writer_schema == reader_schema:
- reader_schema = None
- else:
reader_schema = match_schemas(writer_schema, reader_schema)
try:
@@ -753,9 +748,9 @@ cpdef _read_data(
else:
return _read_data(
fo,
- named_schemas[record_type],
+ named_schemas["writer"][record_type],
named_schemas,
- named_schemas.get(reader_schema),
+ named_schemas["reader"].get(reader_schema),
return_record_name,
return_record_name_override,
)
@@ -812,7 +807,7 @@ cpdef _skip_data(
elif record_type == "record" or record_type == "error":
skip_record(fo, writer_schema, named_schemas)
else:
- _skip_data(fo, named_schemas[record_type], named_schemas)
+ _skip_data(fo, named_schemas["writer"][record_type], named_schemas)
cpdef skip_sync(fo, sync_marker):
@@ -1029,16 +1024,16 @@ class file_reader:
self._schema = json.loads(self.metadata["avro.schema"])
self.codec = self.metadata.get("avro.codec", "null")
- self._named_schemas = {}
+ self._named_schemas = {"writer": {}, "reader": {}}
if reader_schema:
self.reader_schema = parse_schema(
- reader_schema, self._named_schemas, _write_hint=False
+ reader_schema, self._named_schemas["reader"], _write_hint=False
)
else:
self.reader_schema = None
self.writer_schema = parse_schema(
- self._schema, self._named_schemas, _write_hint=False, _force=True
+ self._schema, self._named_schemas["writer"], _write_hint=False, _force=True
)
self._elems = None
@@ -1096,11 +1091,11 @@ cpdef schemaless_reader(fo, writer_schema, reader_schema=None,
# No need for the reader schema if they are the same
reader_schema = None
- named_schemas = {}
- writer_schema = parse_schema(writer_schema, named_schemas)
+ named_schemas = {"writer": {}, "reader": {}}
+ writer_schema = parse_schema(writer_schema, named_schemas["writer"])
if reader_schema:
- reader_schema = parse_schema(reader_schema)
+ reader_schema = parse_schema(reader_schema, named_schemas["reader"])
return _read_data(
fo,

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
@@ -14,7 +14,7 @@ from datetime import datetime, timezone
from decimal import Context
from io import BytesIO
from struct import error as StructError
-from typing import IO, Union, Optional, Generic, TypeVar, Iterator
+from typing import IO, Union, Optional, Generic, TypeVar, Iterator, Dict
from .io.binary_decoder import BinaryDecoder
from .io.json_decoder import AvroJSONDecoder
@@ -457,7 +457,7 @@ def read_union(
elif return_record_name and extract_record_type(idx_schema) not in AVRO_TYPES:
# idx_schema is a named type
return (
- named_schemas[idx_schema]["name"],
+ named_schemas["writer"][idx_schema]["name"],
read_data(
decoder,
idx_schema,
@@ -627,11 +627,6 @@ def read_data(
record_type = extract_record_type(writer_schema)
if reader_schema:
- # If the schemas are the same, set the reader schema to None so that no
- # schema resolution is done for this call or future recursive calls
- if writer_schema == reader_schema:
- reader_schema = None
- else:
reader_schema = match_schemas(writer_schema, reader_schema)
reader_fn = READERS.get(record_type)
@@ -661,9 +656,9 @@ def read_data(
else:
return read_data(
decoder,
- named_schemas[record_type],
+ named_schemas["writer"][record_type],
named_schemas,
- named_schemas.get(reader_schema),
+ named_schemas["reader"].get(reader_schema),
return_record_name,
return_record_name_override,
)
@@ -676,7 +671,7 @@ def skip_data(decoder, writer_schema, named_schemas):
if reader_fn:
reader_fn(decoder, writer_schema, named_schemas)
else:
- skip_data(decoder, named_schemas[record_type], named_schemas)
+ skip_data(decoder, named_schemas["writer"][record_type], named_schemas)
def skip_sync(fo, sync_marker):
@@ -925,11 +920,12 @@ class file_reader(Generic[T]):
# If a decoder was not provided, assume binary
self.decoder = BinaryDecoder(fo_or_decoder)
- self._named_schemas = {}
+ self._named_schemas = {"writer": {}, "reader": {}}
if reader_schema:
self.reader_schema = parse_schema(
- reader_schema, self._named_schemas, _write_hint=False
+ reader_schema, self._named_schemas["reader"], _write_hint=False
)
+
else:
self.reader_schema = None
self.return_record_name = return_record_name
@@ -958,7 +954,7 @@ class file_reader(Generic[T]):
# Always parse the writer schema since it might have named types that
# need to be stored in self._named_types
self.writer_schema = parse_schema(
- self._schema, self._named_schemas, _write_hint=False, _force=True
+ self._schema, self._named_schemas["writer"], _write_hint=False, _force=True
)
@property
@@ -1049,10 +1045,12 @@ class reader(file_reader[AvroMessage]):
)
if isinstance(self.decoder, AvroJSONDecoder):
- self.decoder.configure(self.reader_schema, self._named_schemas)
+ self.decoder.configure(self.reader_schema, self._named_schemas["reader"])
self.writer_schema = self.reader_schema
self.reader_schema = None
+ self._named_schemas["writer"] = self._named_schemas["reader"]
+ self._named_schemas["reader"] = {}
def _elems():
while not self.decoder.done:
@@ -1195,11 +1193,11 @@ def schemaless_reader(
# No need for the reader schema if they are the same
reader_schema = None
- named_schemas: NamedSchemas = {}
- writer_schema = parse_schema(writer_schema, named_schemas)
+ named_schemas: Dict[str, NamedSchemas] = {"writer": {}, "reader": {}}
+ writer_schema = parse_schema(writer_schema, named_schemas["writer"])
if reader_schema:
- reader_schema = parse_schema(reader_schema)
+ reader_schema = parse_schema(reader_schema, named_schemas["reader"])
decoder = BinaryDecoder(fo)

MODIFY
tests/test_schema_evolution.py
tests/test_schema_evolution.py
@@ -705,3 +705,49 @@ def test_writer_enum_more_symbols_than_reader_enum():
else:
with pytest.raises(SchemaResolutionError):
list(fastavro.reader(bio, reader_schema))
+
+
+def test_union_of_schemas_evolution():
+ """https://github.com/fastavro/fastavro/issues/553"""
+ original_schema = [
+ {
+ "name": "other_record",
+ "type": "record",
+ "fields": [
+ {"name": "other_one", "type": "boolean"},
+ {"name": "other_two", "type": "float"},
+ ],
+ },
+ {
+ "name": "root_record",
+ "type": "record",
+ "fields": [
+ {"name": "one", "type": "boolean"},
+ {"name": "two", "type": "float"},
+ {"name": "three", "type": "other_record"},
+ ],
+ },
+ ]
+
+ new_schema = [
+ {
+ "name": "other_record",
+ "type": "record",
+ "fields": [
+ {"name": "other_two", "type": "float"},
+ {"name": "new_three", "type": "float", "default": 0.0},
+ ],
+ },
+ {
+ "name": "root_record",
+ "type": "record",
+ "fields": [{"name": "three", "type": "other_record"}],
+ },
+ ]
+
+ record = {"one": True, "two": 0.0, "three": {"other_one": False, "other_two": 1.0}}
+
+ binary = avro_to_bytes_with_schema(original_schema, record)
+
+ output_using_new_schema = bytes_with_schema_to_avro(new_schema, binary)
+ assert output_using_new_schema == {"three": {"other_two": 1.0, "new_three": 0.0}}

fix issue where writer and reader schemas would collide