This is an automated email from the ASF dual-hosted git repository.
fokko pushed a commit to branch branch-1.9
in repository https://gitbox.apache.org/repos/asf/avro.git
The following commit(s) were added to refs/heads/branch-1.9 by this push:
new 349d35c AVRO-1837: Logical Types: date, time & timestamp (#545)
349d35c is described below
commit 349d35c632e87b453dda3392673a325fbd4a3d81
Author: Michael A. Smith <[email protected]>
AuthorDate: Wed Jun 19 08:27:14 2019 -0400
AVRO-1837: Logical Types: date, time & timestamp (#545)
* AVRO-1837: Logicaltype support for date, time and timestamp
* AVRO-1837: add timezones.py file
* AVRO-1837: fix init file in py
closes #207
closes #82
(cherry picked from commit 81605f7d6467e687c518be5c3097521edcbb9494)
---
lang/py/src/avro/__init__.py | 2 +-
lang/py/src/avro/{__init__.py => constants.py} | 19 ++-
lang/py/src/avro/io.py | 172 ++++++++++++++++++++++++-
lang/py/src/avro/schema.py | 106 ++++++++++++++-
lang/py/src/avro/{__init__.py => timezones.py} | 30 ++++-
lang/py/test/test_io.py | 36 +++++-
lang/py/test/test_schema.py | 66 ++++++++++
7 files changed, 416 insertions(+), 15 deletions(-)
diff --git a/lang/py/src/avro/__init__.py b/lang/py/src/avro/__init__.py
index da51d9b..d7e2896 100644
--- a/lang/py/src/avro/__init__.py
+++ b/lang/py/src/avro/__init__.py
@@ -14,5 +14,5 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-__all__ = ['schema', 'io', 'datafile', 'protocol', 'ipc']
+__all__ = ['schema', 'io', 'datafile', 'protocol', 'ipc', 'constants',
'timezones']
diff --git a/lang/py/src/avro/__init__.py b/lang/py/src/avro/constants.py
similarity index 69%
copy from lang/py/src/avro/__init__.py
copy to lang/py/src/avro/constants.py
index da51d9b..fa0b7d3 100644
--- a/lang/py/src/avro/__init__.py
+++ b/lang/py/src/avro/constants.py
@@ -14,5 +14,22 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-__all__ = ['schema', 'io', 'datafile', 'protocol', 'ipc']
+"""
+Contains Constants for Python Avro
+"""
+DATE = "date"
+DECIMAL = "decimal"
+TIMESTAMP_MICROS = "timestamp-micros"
+TIMESTAMP_MILLIS = "timestamp-millis"
+TIME_MICROS = "time-micros"
+TIME_MILLIS = "time-millis"
+
+SUPPORTED_LOGICAL_TYPE = [
+ DATE,
+ DECIMAL,
+ TIMESTAMP_MICROS,
+ TIMESTAMP_MILLIS,
+ TIME_MICROS,
+ TIME_MILLIS
+]
diff --git a/lang/py/src/avro/io.py b/lang/py/src/avro/io.py
index 2901660..9484f98 100644
--- a/lang/py/src/avro/io.py
+++ b/lang/py/src/avro/io.py
@@ -38,8 +38,11 @@ uses the following mapping:
"""
import struct
from avro import schema
+from avro import constants
+from avro import timezones
import sys
from binascii import crc32
+import datetime
from decimal import Decimal
from decimal import getcontext
@@ -104,6 +107,9 @@ class SchemaResolutionException(schema.AvroException):
# Validate
#
+def _is_timezone_aware_datetime(dt):
+ return dt.tzinfo is not None and dt.tzinfo.utcoffset(dt) is not None
+
def validate(expected_schema, datum):
"""Determine if a python datum is an instance of a schema."""
schema_type = expected_schema.type
@@ -119,10 +125,20 @@ def validate(expected_schema, datum):
return isinstance(datum, Decimal)
return isinstance(datum, str)
elif schema_type == 'int':
- return ((isinstance(datum, int) or isinstance(datum, long))
+ if hasattr(expected_schema, 'logical_type'):
+ if expected_schema.logical_type == constants.DATE:
+ return isinstance(datum, datetime.date)
+ elif expected_schema.logical_type == constants.TIME_MILLIS:
+ return isinstance(datum, datetime.time)
+ return (isinstance(datum, (int, long))
and INT_MIN_VALUE <= datum <= INT_MAX_VALUE)
elif schema_type == 'long':
- return ((isinstance(datum, int) or isinstance(datum, long))
+ if hasattr(expected_schema, 'logical_type'):
+ if expected_schema.logical_type == constants.TIME_MICROS:
+ return isinstance(datum, datetime.time)
+ elif expected_schema.logical_type in [constants.TIMESTAMP_MILLIS,
constants.TIMESTAMP_MICROS]:
+ return isinstance(datum, datetime.datetime) and
_is_timezone_aware_datetime(datum)
+ return (isinstance(datum, (int, long))
and LONG_MIN_VALUE <= datum <= LONG_MAX_VALUE)
elif schema_type in ['float', 'double']:
return (isinstance(datum, int) or isinstance(datum, long)
@@ -280,6 +296,66 @@ class BinaryDecoder(object):
"""
return unicode(self.read_bytes(), "utf-8")
+ def read_date_from_int(self):
+ """
+ int is decoded as python date object.
+ int stores the number of days from
+ the unix epoch, 1 January 1970 (ISO calendar).
+ """
+ days_since_epoch = self.read_int()
+ return datetime.date(1970, 1, 1) + datetime.timedelta(days_since_epoch)
+
+ def _build_time_object(self, value, scale_to_micro):
+ value = value * scale_to_micro
+ value, microseconds = value // 1000000, value % 1000000
+ value, seconds = value // 60, value % 60
+ value, minutes = value // 60, value % 60
+ hours = value
+
+ return datetime.time(
+ hour=hours,
+ minute=minutes,
+ second=seconds,
+ microsecond=microseconds
+ )
+
+ def read_time_millis_from_int(self):
+ """
+ int is decoded as python time object which represents
+ the number of milliseconds after midnight, 00:00:00.000.
+ """
+ milliseconds = self.read_int()
+ return self._build_time_object(milliseconds, 1000)
+
+ def read_time_micros_from_long(self):
+ """
+ long is decoded as python time object which represents
+ the number of microseconds after midnight, 00:00:00.000000.
+ """
+ microseconds = self.read_long()
+ return self._build_time_object(microseconds, 1)
+
+ def read_timestamp_millis_from_long(self):
+ """
+ long is decoded as python datetime object which represents
+ the number of milliseconds from the unix epoch, 1 January 1970.
+ """
+ timestamp_millis = self.read_long()
+ timedelta = datetime.timedelta(microseconds=timestamp_millis * 1000)
+ unix_epoch_datetime = datetime.datetime(1970, 1, 1, 0, 0, 0, 0,
tzinfo=timezones.utc)
+ return unix_epoch_datetime + timedelta
+
+ def read_timestamp_micros_from_long(self):
+ """
+ long is decoded as python datetime object which represents
+ the number of microseconds from the unix epoch, 1 January 1970.
+ """
+ timestamp_micros = self.read_long()
+ timedelta = datetime.timedelta(microseconds=timestamp_micros)
+ unix_epoch_datetime = datetime.datetime(1970, 1, 1, 0, 0, 0, 0,
tzinfo=timezones.utc)
+ return unix_epoch_datetime + timedelta
+
+
def check_crc32(self, bytes):
checksum = STRUCT_CRC32.unpack(self.read(4))[0];
if crc32(bytes) & 0xffffffff != checksum:
@@ -478,6 +554,56 @@ class BinaryEncoder(object):
"""
self.write(STRUCT_CRC32.pack(crc32(bytes) & 0xffffffff));
+ def write_date_int(self, datum):
+ """
+ Encode python date object as int.
+ It stores the number of days from
+ the unix epoch, 1 January 1970 (ISO calendar).
+ """
+ delta_date = datum - datetime.date(1970, 1, 1)
+ self.write_int(delta_date.days)
+
+ def write_time_millis_int(self, datum):
+ """
+ Encode python time object as int.
+ It stores the number of milliseconds from midnight, 00:00:00.000
+ """
+ milliseconds = datum.hour*3600000 + datum.minute * 60000 + datum.second *
1000 + datum.microsecond // 1000
+ self.write_int(milliseconds)
+
+ def write_time_micros_long(self, datum):
+ """
+ Encode python time object as long.
+ It stores the number of microseconds from midnight, 00:00:00.000000
+ """
+ microseconds = datum.hour*3600000000 + datum.minute * 60000000 +
datum.second * 1000000 + datum.microsecond
+ self.write_long(microseconds)
+
+ def _timedelta_total_microseconds(self, timedelta):
+ return (
+ timedelta.microseconds + (timedelta.seconds + timedelta.days * 24 *
3600) * 10 ** 6)
+
+ def write_timestamp_millis_long(self, datum):
+ """
+ Encode python datetime object as long.
+ It stores the number of milliseconds from midnight of unix epoch, 1
January 1970.
+ """
+ datum = datum.astimezone(tz=timezones.utc)
+ timedelta = datum - datetime.datetime(1970, 1, 1, 0, 0, 0, 0,
tzinfo=timezones.utc)
+ milliseconds = self._timedelta_total_microseconds(timedelta) / 1000
+ self.write_long(long(milliseconds))
+
+ def write_timestamp_micros_long(self, datum):
+ """
+ Encode python datetime object as long.
+ It stores the number of microseconds from midnight of unix epoch, 1
January 1970.
+ """
+ datum = datum.astimezone(tz=timezones.utc)
+ timedelta = datum - datetime.datetime(1970, 1, 1, 0, 0, 0, 0,
tzinfo=timezones.utc)
+ microseconds = self._timedelta_total_microseconds(timedelta)
+ self.write_long(long(microseconds))
+
+
#
# DatumReader/Writer
#
@@ -583,9 +709,26 @@ class DatumReader(object):
elif writers_schema.type == 'string':
return decoder.read_utf8()
elif writers_schema.type == 'int':
- return decoder.read_int()
+ if (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.DATE):
+ return decoder.read_date_from_int()
+ elif (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.TIME_MILLIS):
+ return decoder.read_time_millis_from_int()
+ else:
+ return decoder.read_int()
elif writers_schema.type == 'long':
- return decoder.read_long()
+ if (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.TIME_MICROS):
+ return decoder.read_time_micros_from_long()
+ elif (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.TIMESTAMP_MILLIS):
+ return decoder.read_timestamp_millis_from_long()
+ elif (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.TIMESTAMP_MICROS):
+ return decoder.read_timestamp_micros_from_long()
+ else:
+ return decoder.read_long()
elif writers_schema.type == 'float':
return decoder.read_float()
elif writers_schema.type == 'double':
@@ -909,9 +1052,26 @@ class DatumWriter(object):
elif writers_schema.type == 'string':
encoder.write_utf8(datum)
elif writers_schema.type == 'int':
- encoder.write_int(datum)
+ if (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.DATE):
+ encoder.write_date_int(datum)
+ elif (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.TIME_MILLIS):
+ encoder.write_time_millis_int(datum)
+ else:
+ encoder.write_int(datum)
elif writers_schema.type == 'long':
- encoder.write_long(datum)
+ if (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.TIME_MICROS):
+ encoder.write_time_micros_long(datum)
+ elif (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.TIMESTAMP_MILLIS):
+ encoder.write_timestamp_millis_long(datum)
+ elif (hasattr(writers_schema, 'logical_type') and
+ writers_schema.logical_type == constants.TIMESTAMP_MICROS):
+ encoder.write_timestamp_micros_long(datum)
+ else:
+ encoder.write_long(datum)
elif writers_schema.type == 'float':
encoder.write_float(datum)
elif writers_schema.type == 'double':
diff --git a/lang/py/src/avro/schema.py b/lang/py/src/avro/schema.py
index 012109f..406ba56 100644
--- a/lang/py/src/avro/schema.py
+++ b/lang/py/src/avro/schema.py
@@ -41,6 +41,8 @@ try:
except ImportError:
import simplejson as json
+from avro import constants
+
#
# Constants
#
@@ -347,7 +349,7 @@ class DecimalLogicalSchema(LogicalSchema):
raise SchemaParseException("Invalid DECIMAL scale %s. Cannot be greater
than precision %s"
%(scale, precision))
- LogicalSchema.__init__(self, 'decimal')
+ super(DecimalLogicalSchema, self).__init__('decimal')
def _max_precision(self):
raise NotImplementedError()
@@ -788,6 +790,91 @@ class RecordSchema(NamedSchema):
to_cmp = json.loads(str(self))
return to_cmp == json.loads(str(that))
+
+#
+# Logical Type
+#
+
+class LogicalSchema(object):
+ def __init__(self, logical_type):
+ self.logical_type = logical_type
+
+
+#
+# Date Type
+#
+
+class DateSchema(LogicalSchema, PrimitiveSchema):
+ def __init__(self, other_props=None):
+ LogicalSchema.__init__(self, constants.DATE)
+ PrimitiveSchema.__init__(self, 'int', other_props)
+
+ def to_json(self, names=None):
+ return self.props
+
+ def __eq__(self, that):
+ return self.props == that.props
+
+#
+# time-millis Type
+#
+
+class TimeMillisSchema(LogicalSchema, PrimitiveSchema):
+ def __init__(self, other_props=None):
+ LogicalSchema.__init__(self, constants.TIME_MILLIS)
+ PrimitiveSchema.__init__(self, 'int', other_props)
+
+ def to_json(self, names=None):
+ return self.props
+
+ def __eq__(self, that):
+ return self.props == that.props
+
+#
+# time-micros Type
+#
+
+class TimeMicrosSchema(LogicalSchema, PrimitiveSchema):
+ def __init__(self, other_props=None):
+ LogicalSchema.__init__(self, constants.TIME_MICROS)
+ PrimitiveSchema.__init__(self, 'long', other_props)
+
+ def to_json(self, names=None):
+ return self.props
+
+ def __eq__(self, that):
+ return self.props == that.props
+
+#
+# timestamp-millis Type
+#
+
+class TimestampMillisSchema(LogicalSchema, PrimitiveSchema):
+ def __init__(self, other_props=None):
+ LogicalSchema.__init__(self, constants.TIMESTAMP_MILLIS)
+ PrimitiveSchema.__init__(self, 'long', other_props)
+
+ def to_json(self, names=None):
+ return self.props
+
+ def __eq__(self, that):
+ return self.props == that.props
+
+#
+# timestamp-micros Type
+#
+
+class TimestampMicrosSchema(LogicalSchema, PrimitiveSchema):
+ def __init__(self, other_props=None):
+ LogicalSchema.__init__(self, constants.TIMESTAMP_MICROS)
+ PrimitiveSchema.__init__(self, 'long', other_props)
+
+ def to_json(self, names=None):
+ return self.props
+
+ def __eq__(self, that):
+ return self.props == that.props
+
#
# Module Methods
#
@@ -817,11 +904,20 @@ def make_avsc_object(json_data, names=None):
logical_type = None
if 'logicalType' in json_data:
logical_type = json_data.get('logicalType')
- if logical_type != 'decimal':
- raise SchemaParseException("Currently does not support %s logical type"
% logical_type)
+ if logical_type not in constants.SUPPORTED_LOGICAL_TYPE:
+ raise SchemaParseException("Currently does not support %s logical
type" % logical_type)
if type in PRIMITIVE_TYPES:
- if type == 'bytes':
- if logical_type == 'decimal':
+ if type == 'int' and logical_type == constants.DATE:
+ return DateSchema(other_props)
+ if type == 'int' and logical_type == constants.TIME_MILLIS:
+ return TimeMillisSchema(other_props=other_props)
+ if type == 'long' and logical_type == constants.TIME_MICROS:
+ return TimeMicrosSchema(other_props=other_props)
+ if type == 'long' and logical_type == constants.TIMESTAMP_MILLIS:
+ return TimestampMillisSchema(other_props=other_props)
+ if type == 'long' and logical_type == constants.TIMESTAMP_MICROS:
+ return TimestampMicrosSchema(other_props=other_props)
+ if type == 'bytes' and logical_type == constants.DECIMAL:
precision = json_data.get('precision')
scale = 0 if json_data.get('scale') is None else
json_data.get('scale')
return BytesDecimalSchema(precision, scale, other_props)
diff --git a/lang/py/src/avro/__init__.py b/lang/py/src/avro/timezones.py
similarity index 60%
copy from lang/py/src/avro/__init__.py
copy to lang/py/src/avro/timezones.py
index da51d9b..f26c4c5 100644
--- a/lang/py/src/avro/__init__.py
+++ b/lang/py/src/avro/timezones.py
@@ -14,5 +14,33 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-__all__ = ['schema', 'io', 'datafile', 'protocol', 'ipc']
+from datetime import datetime
+from datetime import timedelta
+from datetime import tzinfo
+
+class UTCTzinfo(tzinfo):
+ def utcoffset(self, dt):
+ return timedelta(0)
+
+ def tzname(self, dt):
+ return "UTC"
+
+ def dst(self, dt):
+ return timedelta(0)
+
+utc = UTCTzinfo()
+
+
+# Test Time Zone with fixed offset and no DST
+class TSTTzinfo(tzinfo):
+ def utcoffset(self, dt):
+ return timedelta(hours=10)
+
+ def tzname(self, dt):
+ return "TST"
+
+ def dst(self, dt):
+ return timedelta(0)
+
+tst = TSTTzinfo()
diff --git a/lang/py/test/test_io.py b/lang/py/test/test_io.py
index df8b180..e4c0ce2 100644
--- a/lang/py/test/test_io.py
+++ b/lang/py/test/test_io.py
@@ -22,11 +22,13 @@ try:
except ImportError:
from StringIO import StringIO
from binascii import hexlify
+import datetime
import set_avro_test_path
from avro import schema
from avro import io
+from avro import timezones
SCHEMAS_TO_VALIDATE = (
('"null"', None),
@@ -48,6 +50,35 @@ SCHEMAS_TO_VALIDATE = (
('{"type": "array", "items": "long"}', [1, 3, 2]),
('{"type": "map", "values": "long"}', {'a': 1, 'b': 3, 'c': 2}),
('["string", "null", "long"]', None),
+ ('{"type": "int", "logicalType": "date"}', datetime.date(2000, 1, 1)),
+ ('{"type": "int", "logicalType": "time-millis"}', datetime.time(23, 59, 59,
999000)),
+ ('{"type": "int", "logicalType": "time-millis"}', datetime.time(0, 0, 0,
000000)),
+ ('{"type": "long", "logicalType": "time-micros"}', datetime.time(23, 59, 59,
999999)),
+ ('{"type": "long", "logicalType": "time-micros"}', datetime.time(0, 0, 0,
000000)),
+ (
+ '{"type": "long", "logicalType": "timestamp-millis"}',
+ datetime.datetime(1000, 1, 1, 0, 0, 0, 000000, tzinfo=timezones.utc)
+ ),
+ (
+ '{"type": "long", "logicalType": "timestamp-millis"}',
+ datetime.datetime(9999, 12, 31, 23, 59, 59, 999000, tzinfo=timezones.utc)
+ ),
+ (
+ '{"type": "long", "logicalType": "timestamp-millis"}',
+ datetime.datetime(2000, 1, 18, 2, 2, 1, 100000, tzinfo=timezones.tst)
+ ),
+ (
+ '{"type": "long", "logicalType": "timestamp-micros"}',
+ datetime.datetime(1000, 1, 1, 0, 0, 0, 000000, tzinfo=timezones.utc)
+ ),
+ (
+ '{"type": "long", "logicalType": "timestamp-micros"}',
+ datetime.datetime(9999, 12, 31, 23, 59, 59, 999999, tzinfo=timezones.utc)
+ ),
+ (
+ '{"type": "long", "logicalType": "timestamp-micros"}',
+ datetime.datetime(2000, 1, 18, 2, 2, 1, 123499, tzinfo=timezones.tst)
+ ),
("""\
{"type": "record",
"name": "Test",
@@ -211,7 +242,10 @@ class TestIO(unittest.TestCase):
if isinstance(round_trip_datum, Decimal):
round_trip_datum = round_trip_datum.to_eng_string()
datum = str(datum)
- if datum == round_trip_datum: correct += 1
+ elif isinstance(round_trip_datum, datetime.datetime):
+ datum = datum.astimezone(tz=timezones.utc)
+ if datum == round_trip_datum:
+ correct += 1
self.assertEquals(correct, len(SCHEMAS_TO_VALIDATE))
#
diff --git a/lang/py/test/test_schema.py b/lang/py/test/test_schema.py
index e556ae2..a32c3f8 100644
--- a/lang/py/test/test_schema.py
+++ b/lang/py/test/test_schema.py
@@ -313,6 +313,67 @@ DECIMAL_LOGICAL_TYPE = [
"scale": 2}""", True)
]
+DATE_LOGICAL_TYPE = [
+ ExampleSchema("""{
+ "type": "int",
+ "logicalType": "date"} """, True),
+ ExampleSchema("""{
+ "type": "int",
+ "logicalType": "date1"} """, False),
+ ExampleSchema("""{
+ "type": "long",
+ "logicalType": "date"} """, False),
+]
+
+TIMEMILLIS_LOGICAL_TYPE = [
+ ExampleSchema("""{
+ "type": "int",
+ "logicalType": "time-millis"} """, True),
+ ExampleSchema("""{
+ "type": "int",
+ "logicalType": "time-milis"} """, False),
+ ExampleSchema("""{
+ "type": "long",
+ "logicalType": "time-millis"} """, False),
+]
+
+TIMEMICROS_LOGICAL_TYPE = [
+ ExampleSchema("""{
+ "type": "long",
+ "logicalType": "time-micros"} """, True),
+ ExampleSchema("""{
+ "type": "long",
+ "logicalType": "time-micro"} """, False),
+ ExampleSchema("""{
+ "type": "int",
+ "logicalType": "time-micros"} """, False),
+]
+
+TIMESTAMPMILLIS_LOGICAL_TYPE = [
+ ExampleSchema("""{
+ "type": "long",
+ "logicalType": "timestamp-millis"} """, True),
+ ExampleSchema("""{
+ "type": "long",
+ "logicalType": "timestamp-milis"} """, False),
+ ExampleSchema("""{
+ "type": "int",
+ "logicalType": "timestamp-millis"} """, False),
+]
+
+TIMESTAMPMICROS_LOGICAL_TYPE = [
+ ExampleSchema("""{
+ "type": "long",
+ "logicalType": "timestamp-micros"} """, True),
+ ExampleSchema("""{
+ "type": "long",
+ "logicalType": "timestamp-micro"} """, False),
+ ExampleSchema("""{
+ "type": "int",
+ "logicalType": "timestamp-micros"} """, False),
+]
+
+
EXAMPLES = PRIMITIVE_EXAMPLES
EXAMPLES += FIXED_EXAMPLES
EXAMPLES += ENUM_EXAMPLES
@@ -322,6 +383,11 @@ EXAMPLES += UNION_EXAMPLES
EXAMPLES += RECORD_EXAMPLES
EXAMPLES += DOC_EXAMPLES
EXAMPLES += DECIMAL_LOGICAL_TYPE
+EXAMPLES += DATE_LOGICAL_TYPE
+EXAMPLES += TIMEMILLIS_LOGICAL_TYPE
+EXAMPLES += TIMEMICROS_LOGICAL_TYPE
+EXAMPLES += TIMESTAMPMILLIS_LOGICAL_TYPE
+EXAMPLES += TIMESTAMPMICROS_LOGICAL_TYPE
VALID_EXAMPLES = [e for e in EXAMPLES if e.valid]
