add zstandard codec support
MODIFY
.travis.yml
.travis.yml
@@ -4,6 +4,7 @@ dist: xenial
before_install:
- sudo apt-get install -y libsnappy-dev
- pip install python-snappy
+ - pip install zstandard
install:
- pip install -r developer_requirements.txt
script:

MODIFY
README.md
README.md
@@ -32,7 +32,7 @@ encoding/decoding).
* Schemaless Reader
* JSON Writer
* JSON Reader
-* Snappy and Deflate codecs
+* Codecs (Snappy, Deflate, Zstandard)
* Schema resolution
* Aliases
* Logical Types

MODIFY
azure-pipelines.yml
azure-pipelines.yml
@@ -37,6 +37,7 @@ jobs:
- script: |
brew install snappy
CPPFLAGS="-I/usr/local/include -L/usr/local/lib" python -m pip install python-snappy
+ pip install zstandard
- script: python -m pip install --upgrade -r developer_requirements.txt
displayName: 'Install dependencies'

MODIFY
docs/index.rst
docs/index.rst
@@ -27,7 +27,7 @@ Supported Features
* Schemaless Reader
* JSON Writer
* JSON Reader
-* Snappy and Deflate codecs
+* Codecs (Snappy, Deflate, Zstandard)
* Schema resolution
* Aliases
* Logical Types

MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
@@ -21,7 +21,7 @@ from ._six import (
from ._schema import extract_record_type, extract_logical_type, parse_schema
from ._schema_common import SCHEMA_DEFS
from ._read_common import (
- SchemaResolutionError, MAGIC, SYNC_SIZE, HEADER_SCHEMA
+ SchemaResolutionError, MAGIC, SYNC_SIZE, HEADER_SCHEMA, missing_codec_lib
)
from ._timezone import utc
from .const import (
@@ -610,6 +610,7 @@ BLOCK_READERS = {
'deflate': deflate_read_block
}
+
cpdef snappy_read_block(fo):
length = read_long(fo)
data = fo.read(length - 4)
@@ -620,15 +621,25 @@ cpdef snappy_read_block(fo):
try:
import snappy
except ImportError:
- def no_snappy(decoder):
- raise ValueError(
- "snappy codec is supported but you need to install python-snappy"
- )
- BLOCK_READERS['snappy'] = no_snappy
+ BLOCK_READERS['snappy'] = missing_codec_lib("snappy", "python-snappy")
else:
BLOCK_READERS['snappy'] = snappy_read_block
+cpdef zstandard_read_block(fo):
+ length = read_long(fo)
+ data = fo.read(length)
+ return MemoryIO(zstd.ZstdDecompressor().decompress(data))
+
+
+try:
+ import zstandard as zstd
+except ImportError:
+ BLOCK_READERS["zstandard"] = missing_codec_lib("zstandard", "zstandard")
+else:
+ BLOCK_READERS["zstandard"] = zstandard_read_block
+
+
def _iter_avro_records(fo, header, codec, writer_schema, reader_schema):
cdef int32 i

MODIFY
fastavro/_read_common.py
fastavro/_read_common.py
@@ -28,3 +28,12 @@ HEADER_SCHEMA = {
class SchemaResolutionError(Exception):
pass
+
+
+def missing_codec_lib(codec, library):
+ def missing(fo):
+ raise ValueError(
+ "{} codec is supported but you ".format(codec)
+ + "need to install {}".format(library)
+ )
+ return missing

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
@@ -23,7 +23,7 @@ from .six import (
from .schema import extract_record_type, extract_logical_type, parse_schema
from ._schema_common import SCHEMA_DEFS
from ._read_common import (
- SchemaResolutionError, MAGIC, SYNC_SIZE, HEADER_SCHEMA
+ SchemaResolutionError, MAGIC, SYNC_SIZE, HEADER_SCHEMA, missing_codec_lib
)
from ._timezone import utc
from .const import (
@@ -481,15 +481,25 @@ def snappy_read_block(decoder):
try:
import snappy
except ImportError:
- def no_snappy(decoder):
- raise ValueError(
- "snappy codec is supported but you need to install python-snappy"
- )
- BLOCK_READERS['snappy'] = no_snappy
+ BLOCK_READERS['snappy'] = missing_codec_lib("snappy", "python-snappy")
else:
BLOCK_READERS['snappy'] = snappy_read_block
+def zstandard_read_block(decoder):
+ length = read_long(decoder)
+ data = decoder.read_fixed(length)
+ return MemoryIO(zstd.ZstdDecompressor().decompress(data))
+
+
+try:
+ import zstandard as zstd
+except ImportError:
+ BLOCK_READERS["zstandard"] = missing_codec_lib("zstandard", "zstandard")
+else:
+ BLOCK_READERS["zstandard"] = zstandard_read_block
+
+
def _iter_avro_records(decoder, header, codec, writer_schema, reader_schema):
"""Return iterator over avro records."""
sync_marker = header['sync']

MODIFY
fastavro/_write.pyx
fastavro/_write.pyx
@@ -371,14 +371,19 @@ BLOCK_WRITERS = {
}
+def _missing_dependency(codec, library):
+ def missing(fo, block_bytes):
+ raise ValueError(
+ "{} codec is supported but you ".format(codec)
+ + "need to install {}".format(library)
+ )
+ return missing
+
+
try:
import snappy
except ImportError:
- def no_snappy(encoder, block_bytes):
- raise ValueError(
- "snappy codec is supported but you need to install python-snappy"
- )
- BLOCK_WRITERS['snappy'] = no_snappy
+ BLOCK_WRITERS['snappy'] = _missing_dependency("snappy", "python-snappy")
else:
BLOCK_WRITERS['snappy'] = snappy_write_block
@@ -396,6 +401,23 @@ cpdef snappy_write_block(object fo, bytes block_bytes):
fo.write(tmp)
+try:
+ import zstandard as zstd
+except ImportError:
+ BLOCK_WRITERS["zstandard"] = _missing_dependency("zstandard", "zstandard")
+else:
+ BLOCK_WRITERS["zstandard"] = zstandard_write_block
+
+
+cpdef zstandard_write_block(object fo, bytes block_bytes):
+ """Write block in "zstandard" codec."""
+ cdef bytearray tmp = bytearray()
+ data = zstd.ZstdCompressor().compress(block_bytes)
+ write_long(tmp, len(data))
+ fo.write(tmp)
+ fo.write(data)
+
+
cdef class MemoryIO(object):
cdef bytearray value

MODIFY
fastavro/_write_py.py
fastavro/_write_py.py
@@ -265,10 +265,18 @@ BLOCK_WRITERS = {
}
+def _missing_codec_lib(codec, library):
+ def missing(encoder, block_bytes):
+ raise ValueError(
+ "{} codec is supported but you ".format(codec)
+ + "need to install {}".format(library)
+ )
+ return missing
+
+
def snappy_write_block(encoder, block_bytes):
"""Write block in "snappy" codec."""
data = snappy.compress(block_bytes)
-
encoder.write_long(len(data) + 4) # for CRC
encoder._fo.write(data)
encoder.write_crc32(block_bytes)
@@ -277,15 +285,26 @@ def snappy_write_block(encoder, block_bytes):
try:
import snappy
except ImportError:
- def no_snappy(encoder, block_bytes):
- raise ValueError(
- "snappy codec is supported but you need to install python-snappy"
- )
- BLOCK_WRITERS['snappy'] = no_snappy
+ BLOCK_WRITERS['snappy'] = _missing_codec_lib("snappy", "python-snappy")
else:
BLOCK_WRITERS['snappy'] = snappy_write_block
+def zstandard_write_block(encoder, block_bytes):
+ """Write block in "zstandard" codec."""
+ data = zstd.ZstdCompressor().compress(block_bytes)
+ encoder.write_long(len(data))
+ encoder._fo.write(data)
+
+
+try:
+ import zstandard as zstd
+except ImportError:
+ BLOCK_WRITERS["zstandard"] = _missing_codec_lib("zstandard", "zstandard")
+else:
+ BLOCK_WRITERS["zstandard"] = zstandard_write_block
+
+
class GenericWriter(object):
def __init__(self,

MODIFY
setup.py
setup.py
@@ -92,7 +92,9 @@ setup(
'Topic :: Scientific/Engineering :: Information Analysis',
],
extras_require={
+ 'codecs': ['python-snappy', 'zstandard'],
'snappy': ['python-snappy'],
+ 'zstandard': ['zstandard'],
},
tests_require=['pytest', 'flake8', 'check-manifest'],
setup_requires=setup_requires

MODIFY
tests/test_compression.py
tests/test_compression.py
@@ -6,8 +6,9 @@ from fastavro.six import MemoryIO
import fastavro
+@pytest.mark.parametrize("codec", ["snappy", "zstandard"])
@pytest.mark.skipif(os.name == "nt", reason="A pain to set up on windows")
-def test_snappy():
+def test_snappy(codec):
schema = {
"doc": "A weather reading.",
"name": "Weather",
@@ -28,15 +29,16 @@ def test_snappy():
]
file = MemoryIO()
- fastavro.writer(file, schema, records, codec="snappy")
+ fastavro.writer(file, schema, records, codec=codec)
file.seek(0)
out_records = list(fastavro.reader(file))
assert records == out_records
-@pytest.mark.skipif(os.name != "nt", reason="snappy is present")
-def test_snappy_not_installed():
+@pytest.mark.parametrize("codec", ["snappy", "zstandard"])
+@pytest.mark.skipif(os.name != "nt", reason="codec is present")
+def test_snappy_not_installed(codec):
schema = {
"doc": "A weather reading.",
"name": "Weather",
@@ -59,9 +61,9 @@ def test_snappy_not_installed():
file = MemoryIO()
with pytest.raises(
ValueError,
- match="snappy codec is supported but you need to install python-snappy"
+ match="{} codec is supported but you need to install".format(codec)
):
- fastavro.writer(file, schema, records, codec="snappy")
+ fastavro.writer(file, schema, records, codec=codec)
def test_unsupported_codec():

add zstandard codec support