Fixed decimal serialization and deserialization
MODIFY
fastavro/_logical_writers.pyx
fastavro/_logical_writers.pyx
@@ -10,7 +10,7 @@ from cpython.int cimport PyInt_AS_LONG
from cpython.tuple cimport PyTuple_GET_ITEM
from fastavro import const
-from ._six import long, mk_bits
+from ._six import long, mk_bits, int_to_be_signed_bytes
from ._timezone import epoch
ctypedef long long long64
@@ -91,44 +91,31 @@ cpdef prepare_date(object data, schema):
cpdef prepare_bytes_decimal(object data, schema):
- cdef bytearray tmp
+ """Convert decimal.Decimal to bytes"""
if not isinstance(data, decimal.Decimal):
return data
scale = schema.get('scale', 0)
- # based on https://github.com/apache/avro/pull/82/
-
sign, digits, exp = data.as_tuple()
- if -exp > scale:
+ delta = exp + scale
+
+ if delta < 0:
raise ValueError(
'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
- bits_req = unscaled_datum.bit_length() + 1
-
- if sign:
- unscaled_datum = (1 << bits_req) - unscaled_datum
-
- bytes_req = bits_req // 8
- padding_bits = ~((1 << bits_req) - 1) if sign else 0
- packed_bits = padding_bits | unscaled_datum
-
- bytes_req += 1 if (bytes_req << 3) < bits_req else 0
+ unscaled_datum = 10 ** delta * unscaled_datum
- tmp = bytearray()
+ bytes_req = (unscaled_datum.bit_length() + 8) // 8
- for index in range(bytes_req - 1, -1, -1):
- bits_to_write = packed_bits >> (8 * index)
- tmp += mk_bits(bits_to_write & 0xff)
+ if sign:
+ unscaled_datum = -unscaled_datum
- return bytes(tmp)
+ return int_to_be_signed_bytes(unscaled_datum, bytes_req)
cpdef prepare_fixed_decimal(object data, schema):

MODIFY
fastavro/_logical_writers_py.py
fastavro/_logical_writers_py.py
@@ -8,7 +8,7 @@ from .const import (
MCS_PER_HOUR, MCS_PER_MINUTE, MCS_PER_SECOND, MLS_PER_HOUR, MLS_PER_MINUTE,
MLS_PER_SECOND, DAYS_SHIFT
)
-from .six import MemoryIO, long, mk_bits
+from .six import MemoryIO, long, mk_bits, int_to_be_signed_bytes
from ._timezone import epoch
@@ -56,39 +56,26 @@ def prepare_bytes_decimal(data, schema):
return data
scale = schema.get('scale', 0)
- # based on https://github.com/apache/avro/pull/82/
-
sign, digits, exp = data.as_tuple()
- if -exp > scale:
+ delta = exp + scale
+
+ if delta < 0:
raise ValueError(
'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
- bits_req = unscaled_datum.bit_length() + 1
-
- if sign:
- unscaled_datum = (1 << bits_req) - unscaled_datum
-
- bytes_req = bits_req // 8
- padding_bits = ~((1 << bits_req) - 1) if sign else 0
- packed_bits = padding_bits | unscaled_datum
-
- bytes_req += 1 if (bytes_req << 3) < bits_req else 0
+ unscaled_datum = 10 ** delta * unscaled_datum
- tmp = MemoryIO()
+ bytes_req = (unscaled_datum.bit_length() + 8) // 8
- for index in range(bytes_req - 1, -1, -1):
- bits_to_write = packed_bits >> (8 * index)
- tmp.write(mk_bits(bits_to_write & 0xff))
+ if sign:
+ unscaled_datum = -unscaled_datum
- return tmp.getvalue()
+ return int_to_be_signed_bytes(unscaled_datum, bytes_req)
def prepare_fixed_decimal(data, schema):

MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
@@ -17,7 +17,7 @@ from uuid import UUID
import json
from ._six import (
- btou, utob, iteritems, is_str, str2ints, fstint, long
+ btou, utob, iteritems, is_str, long, be_signed_bytes_to_int
)
from ._schema import extract_record_type, extract_logical_type, parse_schema
from ._schema_common import SCHEMA_DEFS
@@ -175,40 +175,11 @@ cpdef read_time_micros(data, writer_schema=None, reader_schema=None):
return datetime.time(h, m, s, mcs)
-cpdef read_bytes_decimal(data, writer_schema=None, reader_schema=None):
- size = len(data)
- return _read_decimal(data, size, writer_schema)
-
-
-cpdef read_fixed_decimal(data, writer_schema=None, reader_schema=None):
- size = writer_schema['size']
- return _read_decimal(data, size, writer_schema)
-
-
-cpdef _read_decimal(data, size, writer_schema):
- """
- based on https://github.com/apache/avro/pull/82/
- """
- cdef int32 offset
+cpdef read_decimal(data, writer_schema=None, reader_schema=None):
scale = writer_schema.get('scale', 0)
precision = writer_schema['precision']
- 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 range(size):
- unscaled_datum <<= 8
- unscaled_datum += datum_byte[offset]
- unscaled_datum += pow(-2, (size * 8) - 1)
- else:
- for offset in range(size):
- unscaled_datum <<= 8
- unscaled_datum += (datum_byte[offset])
+ unscaled_datum = be_signed_bytes_to_int(data)
with localcontext() as ctx:
ctx.prec = precision
@@ -498,8 +469,8 @@ LOGICAL_READERS = {
'long-timestamp-millis': read_timestamp_millis,
'long-timestamp-micros': read_timestamp_micros,
'int-date': read_date,
- 'bytes-decimal': read_bytes_decimal,
- 'fixed-decimal': read_fixed_decimal,
+ 'bytes-decimal': read_decimal,
+ 'fixed-decimal': read_decimal,
'string-uuid': read_uuid,
'int-time-millis': read_time_millis,
'long-time-micros': read_time_micros,

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
@@ -19,7 +19,7 @@ import json
from .io.binary_decoder import BinaryDecoder
from .io.json_decoder import AvroJSONDecoder
from .six import (
- xrange, btou, utob, iteritems, is_str, str2ints, fstint, long
+ xrange, btou, utob, iteritems, is_str, long, be_signed_bytes_to_int
)
from .schema import extract_record_type, extract_logical_type, parse_schema
from ._schema_common import SCHEMA_DEFS
@@ -153,39 +153,11 @@ def read_time_micros(data, writer_schema=None, reader_schema=None):
return datetime.time(h, m, s, mcs)
-def read_bytes_decimal(data, writer_schema=None, reader_schema=None):
- size = len(data)
- return _read_decimal(data, size, writer_schema)
-
-
-def read_fixed_decimal(data, writer_schema=None, reader_schema=None):
- size = writer_schema['size']
- return _read_decimal(data, size, writer_schema)
-
-
-def _read_decimal(data, size, writer_schema):
- """
- based on https://github.com/apache/avro/pull/82/
- """
+def read_decimal(data, writer_schema=None, reader_schema=None):
scale = writer_schema.get('scale', 0)
precision = writer_schema['precision']
- 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])
+ unscaled_datum = be_signed_bytes_to_int(data)
with localcontext() as ctx:
ctx.prec = precision
@@ -359,8 +331,8 @@ LOGICAL_READERS = {
'long-timestamp-millis': read_timestamp_millis,
'long-timestamp-micros': read_timestamp_micros,
'int-date': read_date,
- 'bytes-decimal': read_bytes_decimal,
- 'fixed-decimal': read_fixed_decimal,
+ 'bytes-decimal': read_decimal,
+ 'fixed-decimal': read_decimal,
'string-uuid': read_uuid,
'int-time-millis': read_time_millis,
'long-time-micros': read_time_micros,

MODIFY
fastavro/_six.pyx
fastavro/_six.pyx
@@ -65,6 +65,13 @@ if sys.version_info >= (3, 0):
else:
return False
+ def py3_int_to_be_signed_bytes(num, bytes_req):
+ return num.to_bytes(bytes_req, byteorder='big', signed=True)
+
+ def py3_be_signed_bytes_to_int(data):
+ return int.from_bytes(data, byteorder='big', signed=True)
+
+
else: # Python 2x
from cStringIO import StringIO as MemoryIO # noqa
from cStringIO import StringIO as StringIO # noqa
@@ -138,6 +145,20 @@ else: # Python 2x
else:
return False
+ def py2_int_to_be_signed_bytes(num, bytes_req):
+ if num < 0:
+ num = 2 ** (8 * bytes_req) + num
+ hex_str = '%x' % num
+ hex_str = ((bytes_req * 2) - len(hex_str)) * '0' + hex_str
+ return hex_str.decode('hex')
+
+ def py2_be_signed_bytes_to_int(data):
+ output = int(data.encode('hex'), 16)
+ bitsize = len(data) * 8
+ if output < (2 ** (bitsize - 1)):
+ return output
+ return output - (2 ** bitsize)
+
# We do it this way and not just redifine function since Cython do not like it
if sys.version_info >= (3, 0):
btou = py3_btou
@@ -152,6 +173,8 @@ if sys.version_info >= (3, 0):
str2ints = py3_bytes2ints
fstint = py3_fstint
appendable = py3_appendable
+ int_to_be_signed_bytes = py3_int_to_be_signed_bytes
+ be_signed_bytes_to_int = py3_be_signed_bytes_to_int
else:
btou = py2_btou
utob = py2_utob
@@ -165,3 +188,5 @@ else:
str2ints = py2_str2ints
fstint = py2_fstint
appendable = py2_appendable
+ int_to_be_signed_bytes = py2_int_to_be_signed_bytes
+ be_signed_bytes_to_int = py2_be_signed_bytes_to_int

MODIFY
fastavro/six.py
fastavro/six.py
@@ -64,6 +64,12 @@ if sys.version_info >= (3, 0):
else:
return False
+ def py3_int_to_be_signed_bytes(num, bytes_req):
+ return num.to_bytes(bytes_req, byteorder='big', signed=True)
+
+ def py3_be_signed_bytes_to_int(data):
+ return int.from_bytes(data, byteorder='big', signed=True)
+
else: # Python 2x
from cStringIO import StringIO as MemoryIO # noqa
from cStringIO import StringIO as StringIO # noqa
@@ -137,6 +143,20 @@ else: # Python 2x
else:
return False
+ def py2_int_to_be_signed_bytes(num, bytes_req):
+ if num < 0:
+ num = 2 ** (8 * bytes_req) + num
+ hex_str = '%x' % num
+ hex_str = ((bytes_req * 2) - len(hex_str)) * '0' + hex_str
+ return hex_str.decode('hex')
+
+ def py2_be_signed_bytes_to_int(data):
+ output = int(data.encode('hex'), 16)
+ bitsize = len(data) * 8
+ if output < (2 ** (bitsize - 1)):
+ return output
+ return output - (2 ** bitsize)
+
# We do it this way and not just redifine function since Cython do not like it
if sys.version_info >= (3, 0):
btou = py3_btou
@@ -151,6 +171,8 @@ if sys.version_info >= (3, 0):
str2ints = py3_bytes2ints
fstint = py3_fstint
appendable = py3_appendable
+ int_to_be_signed_bytes = py3_int_to_be_signed_bytes
+ be_signed_bytes_to_int = py3_be_signed_bytes_to_int
else:
btou = py2_btou
utob = py2_utob
@@ -164,3 +186,5 @@ else:
str2ints = py2_str2ints
fstint = py2_fstint
appendable = py2_appendable
+ int_to_be_signed_bytes = py2_int_to_be_signed_bytes
+ be_signed_bytes_to_int = py2_be_signed_bytes_to_int

MODIFY
tests/test_logical_types.py
tests/test_logical_types.py
@@ -143,30 +143,42 @@ schema_bytes_decimal = {
"namespace": "namespace",
"type": "bytes",
"logicalType": "decimal",
- "precision": 15,
+ "precision": 20,
"scale": 3,
}
-def test_bytes_decimal_negative():
- data1 = Decimal("-2.90")
- binary = serialize(schema_bytes_decimal, data1)
- data2 = deserialize(schema_bytes_decimal, binary)
- assert (data1 == data2)
-
-
-def test_bytes_decimal_zero():
- data1 = Decimal("0.0")
- binary = serialize(schema_bytes_decimal, data1)
- data2 = deserialize(schema_bytes_decimal, binary)
- assert (data1 == data2)
-
-
-def test_bytes_decimal_positive():
- data1 = Decimal("123.456")
- binary = serialize(schema_bytes_decimal, data1)
- data2 = deserialize(schema_bytes_decimal, binary)
- assert (data1 == data2)
+@pytest.mark.parametrize(
+ 'input_data, expected_binary',
+ [
+ (Decimal("0.0"), b'\x02\x00'),
+ (Decimal("-0.0"), b'\x02\x00'),
+ (Decimal("0.1"), b'\x02d'),
+ (Decimal("-0.1"), b'\x02\x9c'),
+ (Decimal("0.2"), b'\x04\x00\xc8'),
+ (Decimal("-0.2"), b'\x04\xff8'),
+ (Decimal("0.456"), b'\x04\x01\xc8'),
+ (Decimal("-0.456"), b'\x04\xfe8'),
+ (Decimal("2.55"), b'\x04\t\xf6'),
+ (Decimal("-2.55"), b'\x04\xf6\n'),
+ (Decimal("2.90"), b'\x04\x0bT'),
+ (Decimal("-2.90"), b'\x04\xf4\xac'),
+ (Decimal("123.456"), b'\x06\x01\xe2@'),
+ (Decimal("-123.456"), b'\x06\xfe\x1d\xc0'),
+ (Decimal("3245.234"), b'\x061\x84\xb2'),
+ (Decimal("-3245.234"), b'\x06\xce{N'),
+ (
+ Decimal("9999999999999999.456"),
+ b'\x12\x00\x8a\xc7#\x04\x89\xe7\xfd\xe0'
+ ),
+ (Decimal("-999999999999999.456"), b'\x10\xf2\x1fILX\x9c\x02 '),
+ ]
+)
+def test_bytes_decimal(input_data, expected_binary):
+ binary = serialize(schema_bytes_decimal, input_data)
+ assert (binary == expected_binary)
+ output_data = deserialize(schema_bytes_decimal, binary)
+ assert (input_data == output_data)
def test_bytes_decimal_scale():

Fixed decimal serialization and deserialization