Fix union schema resolution in writer
MODIFY
fastavro/_writer.pyx
fastavro/_writer.pyx
@@ -437,6 +437,10 @@ cpdef write_union(bytearray fo, datum, schema):
zero-based position within the union of the schema of its value. The value
is then encoded per the indicated schema within the union."""
+ cdef int best_match_index
+ cdef int most_fields
+ cdef int index
+ cdef int fields
if isinstance(datum, tuple):
(name, datum) = datum
for index, candidate in enumerate(schema):
@@ -449,12 +453,22 @@ cpdef write_union(bytearray fo, datum, schema):
raise ValueError(msg)
else:
pytype = type(datum)
+ best_match_index = -1
+ most_fields = -1
for index, candidate in enumerate(schema):
if validate(datum, candidate):
- break
+ if extract_record_type(candidate) == 'record':
+ fields = len(candidate['fields'])
+ if fields > most_fields:
+ best_match_index = index
+ most_fields = fields
else:
+ best_match_index = index
+ break
+ if best_match_index < 0:
msg = '%r (type %s) do not match %s' % (datum, pytype, schema)
raise ValueError(msg)
+ index = best_match_index
# write data
write_long(fo, index)

MODIFY
fastavro/_writer_py.py
fastavro/_writer_py.py
@@ -395,12 +395,22 @@ def write_union(fo, datum, schema):
raise ValueError(msg)
else:
pytype = type(datum)
+ best_match_index = -1
+ most_fields = -1
for index, candidate in enumerate(schema):
if validate(datum, candidate):
- break
+ if extract_record_type(candidate) == 'record':
+ fields = len(candidate['fields'])
+ if fields > most_fields:
+ best_match_index = index
+ most_fields = fields
else:
+ best_match_index = index
+ break
+ if best_match_index < 0:
msg = '%r (type %s) do not match %s' % (datum, pytype, schema)
raise ValueError(msg)
+ index = best_match_index
# write data
write_long(fo, index)

MODIFY
tests/test_fastavro.py
tests/test_fastavro.py
@@ -990,3 +990,53 @@ def test_writer_class_split_files(tmpdir):
new_interim_record_counts.append(len(new_records))
assert new_records == records
assert interim_record_counts == new_interim_record_counts
+
+
+def test_union_records():
+ #
+ schema = {
+ 'name': 'test_name',
+ 'namespace': 'test',
+ 'type': 'record',
+ 'fields': [
+ {
+ 'name': 'val',
+ 'type': [
+ {
+ 'name': 'a',
+ 'namespace': 'common',
+ 'type': 'record',
+ 'fields': [
+ {'name': 'x', 'type': 'int'},
+ {'name': 'y', 'type': 'int'},
+ ],
+ },
+ {
+ 'name': 'b',
+ 'namespace': 'common',
+ 'type': 'record',
+ 'fields': [
+ {'name': 'x', 'type': 'int'},
+ {'name': 'y', 'type': 'int'},
+ {'name': 'z', 'type': ['null', 'int']},
+ ],
+ }
+ ]
+ }
+ ]
+ }
+
+ data = [{
+ 'val': {
+ 'x': 3,
+ 'y': 4,
+ 'z': 5,
+ }
+ }]
+ new_file = MemoryIO()
+ fastavro.writer(new_file, schema, data)
+ new_file.seek(0)
+
+ new_reader = fastavro.reader(new_file)
+ new_records = list(new_reader)
+ assert new_records == data

Fix union schema resolution in writer