Added fixed decimal type
MODIFY
fastavro/reader.py
fastavro/reader.py
@@ -182,8 +182,7 @@ def read_time_micros(data, writer_schema=None, reader_schema=None):
def read_bytes_decimal(data, writer_schema=None, reader_schema=None):
"""
- Decimal is encoded as fixed. Fixed instances are encoded using the
- number of bytes declared in the schema.
+ Decimal is encoded as bytes.
based on https://github.com/apache/avro/pull/82/
"""
scale = writer_schema['scale']
@@ -214,6 +213,39 @@ def read_bytes_decimal(data, writer_schema=None, reader_schema=None):
return scaled_datum
+def read_fixed_decimal(data, writer_schema=None, reader_schema=None):
+ """
+ Decimal is encoded as fixed. Fixed instances are encoded using the
+ number of bytes declared in the schema.
+ based on https://github.com/apache/avro/pull/82/
+ """
+ scale = writer_schema['scale']
+ precision = writer_schema['precision']
+ size = writer_schema['size']
+
+ datum_byte = str2ints(data)
+
+ unscaled_datum = 0
+ msb = fstint(data)
+ leftmost_bit = (msb >> 7) & 1
+ if leftmost_bit == 1:
+ modified_first_byte = datum_byte[0] ^ (1 << 7)
+ datum_byte = [modified_first_byte] + datum_byte[1:]
+ for offset in xrange(size):
+ unscaled_datum <<= 8
+ unscaled_datum += datum_byte[offset]
+ unscaled_datum += pow(-2, (size * 8) - 1)
+ else:
+ for offset in xrange(size):
+ unscaled_datum <<= 8
+ unscaled_datum += (datum_byte[offset])
+
+ with localcontext() as ctx:
+ ctx.prec = precision
+ scaled_datum = Decimal(unscaled_datum).scaleb(-scale)
+ return scaled_datum
+
+
def read_long(fo, writer_schema=None, reader_schema=None):
"""int and long values are written using variable-length, zig-zag
coding."""
@@ -435,6 +467,7 @@ LOGICAL_READERS = {
'long-timestamp-micros': read_timestamp_micros,
'int-date': read_date,
'bytes-decimal': read_bytes_decimal,
+ 'fixed-decimal': read_fixed_decimal,
'string-uuid': read_uuid,
'int-time-millis': read_time_millis,
'long-time-micros': read_time_micros,

MODIFY
fastavro/writer.py
fastavro/writer.py
@@ -139,6 +139,67 @@ def prepare_bytes_decimal(data, schema):
return tmp.getvalue()
+def prepare_fixed_decimal(data, schema):
+ if not isinstance(data, decimal.Decimal):
+ return data
+ scale = schema['scale']
+ size = schema['size']
+
+ # based on https://github.com/apache/avro/pull/82/
+
+ sign, digits, exp = data.as_tuple()
+
+ if -exp > scale:
+ raise AssertionError(
+ 'Scale provided in schema does not match the decimal')
+ delta = exp + scale
+ if delta > 0:
+ digits = digits + (0,) * delta
+
+ unscaled_datum = 0
+ for digit in digits:
+ unscaled_datum = (unscaled_datum * 10) + digit
+
+ # 2.6 support
+ if not hasattr(unscaled_datum, 'bit_length'):
+ bits_req = len(bin(abs(unscaled_datum))) - 2
+ else:
+ bits_req = unscaled_datum.bit_length() + 1
+
+ size_in_bits = size * 8
+ offset_bits = size_in_bits - bits_req
+
+ mask = 2 ** size_in_bits - 1
+ bit = 1
+ for i in range(bits_req):
+ mask ^= bit
+ bit <<= 1
+
+ if bits_req < 8:
+ bytes_req = 1
+ else:
+ bytes_req = bits_req // 8
+ if bits_req % 8 != 0:
+ bytes_req += 1
+
+ tmp = MemoryIO()
+
+ if sign:
+ unscaled_datum = (1 << bits_req) - unscaled_datum
+ unscaled_datum = mask | unscaled_datum
+ for index in range(size - 1, -1, -1):
+ bits_to_write = unscaled_datum >> (8 * index)
+ tmp.write(mk_bits(bits_to_write & 0xff))
+ else:
+ for i in range(offset_bits//8):
+ tmp.write(mk_bits(0))
+ for index in range(bytes_req - 1, -1, -1):
+ bits_to_write = unscaled_datum >> (8 * index)
+ tmp.write(mk_bits(bits_to_write & 0xff))
+
+ return tmp.getvalue()
+
+
def write_int(fo, datum, schema=None):
"""int and long values are written using variable-length, zig-zag coding.
"""
@@ -275,7 +336,10 @@ def validate(datum, schema):
return isinstance(datum, (int, long, float))
if record_type == 'fixed':
- return isinstance(datum, bytes) and len(datum) == schema['size']
+ return (
+ (isinstance(datum, bytes) and len(datum) == schema['size'])
+ or (isinstance(datum, decimal.Decimal))
+ )
if record_type == 'union':
if isinstance(datum, tuple):
@@ -368,6 +432,7 @@ LOGICAL_WRITERS = {
'long-timestamp-micros': prepare_timestamp_micros,
'int-date': prepare_date,
'bytes-decimal': prepare_bytes_decimal,
+ 'fixed-decimal': prepare_fixed_decimal,
'string-uuid': prepare_uuid,
'int-time-millis': prepare_time_millis,
'long-time-micros': prepare_time_micros,

MODIFY
tests/test_complex.py
tests/test_complex.py
@@ -16,7 +16,7 @@ schema = {
"logicalType": "timestamp-micros"}]
},
{
- "name": "array_decimal",
+ "name": "array_bytes_decimal",
"type": ["null", {"type": "array",
"items": {"type": "bytes",
"logicalType": "decimal",
@@ -24,6 +24,16 @@ schema = {
"scale": 6, }
}]
},
+ {
+ "name": "array_fixed_decimal",
+ "type": ["null", {"type": "array",
+ "items": {"type": "fixed",
+ "size": 8,
+ "logicalType": "decimal",
+ "precision": 18,
+ "scale": 6, }
+ }]
+ },
{
"name": "array_record",
"type": {"type": "array", "items": {
@@ -70,10 +80,25 @@ def deserialize(schema, binary):
def test_complex_schema():
data1 = {
'array_string': ['a', "b", "c"],
- 'array_decimal': [Decimal("123.456")],
'multi_union_time': datetime.datetime.now(),
+ 'array_bytes_decimal': [Decimal("123.456")],
+ 'array_fixed_decimal': [Decimal("123.456")],
'array_record': [dict(f1="1", f2=Decimal("123.456"))]
}
binary = serialize(schema, data1)
data2 = deserialize(schema, binary)
assert (data1 == data2)
+
+
+def test_complex_schema_nulls():
+ data1 = {
+ 'array_string': ['a', "b", "c"],
+ 'array_record': [dict(f1="1", f2=Decimal("123.456"))]
+ }
+ binary = serialize(schema, data1)
+ data2 = deserialize(schema, binary)
+ data1_compare = data1
+ data1_compare.update(
+ {'multi_union_time': None, 'array_bytes_decimal': None,
+ 'array_fixed_decimal': None})
+ assert (data1_compare == data2)

MODIFY
tests/test_logical_types.py
tests/test_logical_types.py
@@ -114,6 +114,7 @@ def test_null():
assert (data2['date'] is None)
+# test bytes decimal
schema_top = {
"name": "n",
"namespace": "namespace",
@@ -132,28 +133,34 @@ def test_top():
assert (data1 == data2)
-def test_negative():
+def test_bytes_decimal_negative():
data1 = Decimal("-2.90")
binary = serialize(schema_top, data1)
data2 = deserialize(schema_top, binary)
assert (data1 == data2)
-def test_zero():
+def test_bytes_decimal_zero():
data1 = Decimal("0.0")
binary = serialize(schema_top, data1)
data2 = deserialize(schema_top, binary)
assert (data1 == data2)
-def test_bytes_decimal():
+def test_bytes_decimal_positive():
data1 = Decimal("123.456")
binary = serialize(schema_top, data1)
data2 = deserialize(schema_top, binary)
assert (data1 == data2)
-schema_leftmost = {
+def test_bytes_decimal_scale():
+ data1 = Decimal("123.456678") # does not fit scale
+ with pytest.raises(AssertionError):
+ serialize(schema_top, data1)
+
+
+schema_bytes_decimal_leftmost = {
"name": "n",
"namespace": "namespace",
"type": "bytes",
@@ -163,13 +170,67 @@ schema_leftmost = {
}
-def test_leftmost():
- binary = serialize(schema_leftmost, b'\xd5F\x80')
- data2 = deserialize(schema_leftmost, binary)
+def test_bytes_decimal_leftmost():
+ binary = serialize(schema_bytes_decimal_leftmost, b'\xd5F\x80')
+ data2 = deserialize(schema_bytes_decimal_leftmost, binary)
assert (Decimal("-2.80") == data2)
-def test_scale():
+# test fixed decimal
+schema_fixed_decimal = {
+ "name": "n",
+ "namespace": "namespace",
+ "type": "fixed",
+ "size": 8,
+ "logicalType": "decimal",
+ "precision": 15,
+ "scale": 3,
+}
+
+
+def test_fixed_decimal_negative():
+ data1 = Decimal("-2.90")
+ binary = serialize(schema_fixed_decimal, data1)
+ data2 = deserialize(schema_fixed_decimal, binary)
+ assert (data1 == data2)
+ assert (len(binary) == schema_fixed_decimal['size'])
+
+
+def test_fixed_decimal_zero():
+ data1 = Decimal("0.0")
+ binary = serialize(schema_fixed_decimal, data1)
+ data2 = deserialize(schema_fixed_decimal, binary)
+ assert (data1 == data2)
+ assert (len(binary) == schema_fixed_decimal['size'])
+
+
+def test_fixed_decimal_positive():
+ data1 = Decimal("123.456")
+ binary = serialize(schema_fixed_decimal, data1)
+ data2 = deserialize(schema_fixed_decimal, binary)
+ assert (data1 == data2)
+ assert (len(binary) == schema_fixed_decimal['size'])
+
+
+def test_fixed_decimal_scale():
data1 = Decimal("123.456678") # does not fit scale
with pytest.raises(AssertionError):
- serialize(schema_top, data1)
+ serialize(schema_fixed_decimal, data1)
+
+
+schema_fixed_decimal_leftmost = {
+ "name": "n",
+ "namespace": "namespace",
+ "type": "fixed",
+ "size": 8,
+ "logicalType": "decimal",
+ "precision": 18,
+ "scale": 6,
+}
+
+
+def test_fixed_decimal_binary():
+ binary = serialize(schema_fixed_decimal_leftmost,
+ b'\xFF\xFF\xFF\xFF\xFF\xd5F\x80')
+ data2 = deserialize(schema_fixed_decimal_leftmost, binary)
+ assert (Decimal("-2.80") == data2)

Added fixed decimal type (#105)