Python: Add Avro read path

python/.pre-commit-config.yaml

@@ -54,4 +54,4 @@ repos:
     rev: '4.0.1'
     hooks:
       - id: flake8
-        args: [ "--ignore=E501,W503" ]
+        args: [ "--ignore=E501,W503,E203" ]

python/LICENSE

@@ -202,3 +202,14 @@
    limitations under the License.
 
 --------------------------------------------------------------------------------
+
+This product includes code from Apache Avro.
+
+* Code for initializing the Avro (de)compression codecs
+* The Binary decoder for reading in an Avro byte stream
+
+Copyright: 2014-2022 The Apache Software Foundation.
+Home page: https://avro.apache.org/
+License: https://www.apache.org/licenses/LICENSE-2.0
+
+--------------------------------------------------------------------------------

python/pyproject.toml

@@ -43,10 +43,15 @@ mmh3 = "^3.0.0"
 
 pyarrow = { version = "^8.0.0", optional = true }
 
+zstandard = { version = "^0.17.0", optional = true }
+
+python-snappy = { version = "^0.6.1", optional = true }
+
 [tool.poetry.dev-dependencies]
 pytest = "^7.0.0"
 pytest-checkdocs = "^2.0.0"
 pre-commit = "^2.0.0"
+fastavro = "^1.5.1"
 coverage = { version = "^6.0.0", extras = ["toml"] }
 
 [build-system]
@@ -55,6 +60,8 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry.extras]
 pyarrow = ["pyarrow"]
+snappy = ["python-snappy"]
+python-snappy = ["zstandard"]
 
 [tool.black]
 line-length = 130
@@ -79,5 +86,13 @@ warn_unreachable = true
 module = "mypy-pyarrow.*"
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "mypy-snappy.*"
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "mypy-zstandard.*"
+ignore_missing_imports = true
+
 [tool.coverage.run]
 source = ['src/']

python/spellcheck-dictionary.txt

@@ -57,3 +57,9 @@ UnboundPredicate
 BoundPredicate
 BooleanExpression
 BooleanExpressionVisitor
+zigzag
+unix
+zlib
+Codecs
+codecs
+uri

python/src/iceberg/avro/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

python/src/iceberg/avro/codecs/__init__.py

@@ -0,0 +1,40 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Contains Codecs for Python Avro.
+
+Note that the word "codecs" means "compression/decompression algorithms" in the
+Avro world (https://avro.apache.org/docs/current/spec.html#Object+Container+Files),
+so don't confuse it with the Python's "codecs", which is a package mainly for
+converting character sets (https://docs.python.org/3/library/codecs.html).
+"""
+from __future__ import annotations
+
+from iceberg.avro.codecs.bzip2 import BZip2Codec
+from iceberg.avro.codecs.codec import Codec
+from iceberg.avro.codecs.deflate import DeflateCodec
+from iceberg.avro.codecs.snappy_codec import SnappyCodec
+from iceberg.avro.codecs.zstandard_codec import ZStandardCodec
+
+KNOWN_CODECS: dict[str, type[Codec] | None] = {
+    "null": None,
+    "bzip2": BZip2Codec,
+    "snappy": SnappyCodec,
+    "zstandard": ZStandardCodec,
+    "deflate": DeflateCodec,
+}

python/src/iceberg/avro/codecs/bzip2.py

@@ -0,0 +1,43 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from iceberg.avro.codecs.codec import Codec
+
+try:
+    import bz2
+
+    class BZip2Codec(Codec):
+        @staticmethod
+        def compress(data: bytes) -> tuple[bytes, int]:
+            compressed_data = bz2.compress(data)
+            return compressed_data, len(compressed_data)
+
+        @staticmethod
+        def decompress(data: bytes) -> bytes:
+            return bz2.decompress(data)
+
+except ImportError:
+
+    class BZip2Codec(Codec):  # type: ignore
+        @staticmethod
+        def compress(data: bytes) -> tuple[bytes, int]:
+            raise ImportError("Python bzip2 support not installed, please install the extension")
+
+        @staticmethod
+        def decompress(data: bytes) -> bytes:
+            raise ImportError("Python bzip2 support not installed, please install the extension")

python/src/iceberg/avro/codecs/codec.py

@@ -0,0 +1,33 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+
+class Codec(ABC):
+    """Abstract base class for all Avro codec classes."""
+
+    @staticmethod
+    @abstractmethod
+    def compress(data: bytes) -> tuple[bytes, int]:
+        ...
+
+    @staticmethod
+    @abstractmethod
+    def decompress(data: bytes) -> bytes:
+        ...

python/src/iceberg/avro/codecs/deflate.py

@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import zlib
+
+from iceberg.avro.codecs.codec import Codec
+
+
+class DeflateCodec(Codec):
+    @staticmethod
+    def compress(data: bytes) -> tuple[bytes, int]:
+        # The first two characters and last character are zlib
+        # wrappers around deflate data.
+        compressed_data = zlib.compress(data)[2:-1]
+        return compressed_data, len(compressed_data)
+
+    @staticmethod
+    def decompress(data: bytes) -> bytes:
+        # -15 is the log of the window size; negative indicates
+        # "raw" (no zlib headers) decompression.  See zlib.h.
+        return zlib.decompress(data, -15)

python/src/iceberg/avro/codecs/snappy_codec.py

@@ -0,0 +1,69 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import binascii
+import struct
+
+from iceberg.avro.codecs.codec import Codec
+
+STRUCT_CRC32 = struct.Struct(">I")  # big-endian unsigned int
+
+try:
+    import snappy
+
+    class SnappyCodec(Codec):
+        @staticmethod
+        def _check_crc32(bytes_: bytes, checksum: bytes) -> None:
+            """Incrementally compute CRC-32 from bytes and compare to a checksum
+
+            Args:
+              bytes_ (bytes): The bytes to check against `checksum`
+              checksum (bytes): Byte representation of a checksum
+
+            Raises:
+              ValueError: If the computed CRC-32 does not match the checksum
+            """
+            if binascii.crc32(bytes_) & 0xFFFFFFFF != STRUCT_CRC32.unpack(checksum)[0]:
+                raise ValueError("Checksum failure")
+
+        @staticmethod
+        def compress(data: bytes) -> tuple[bytes, int]:
+            compressed_data = snappy.compress(data)
+            # A 4-byte, big-endian CRC32 checksum
+            compressed_data += STRUCT_CRC32.pack(binascii.crc32(data) & 0xFFFFFFFF)
+            return compressed_data, len(compressed_data)
+
+        @staticmethod
+        def decompress(data: bytes) -> bytes:
+            # Compressed data includes a 4-byte CRC32 checksum
+            data = data[0:-4]
+            uncompressed = snappy.decompress(data)
+            checksum = data[-4:]
+            SnappyCodec._check_crc32(uncompressed, checksum)
+            return uncompressed
+
+except ImportError:
+
+    class SnappyCodec(Codec):  # type: ignore
+        @staticmethod
+        def compress(data: bytes) -> tuple[bytes, int]:
+            raise ImportError("Snappy support not installed, please install using `pip install pyiceberg[snappy]`")
+
+        @staticmethod
+        def decompress(data: bytes) -> bytes:
+            raise ImportError("Snappy support not installed, please install using `pip install pyiceberg[snappy]`")

python/src/iceberg/avro/codecs/zstandard_codec.py

@@ -0,0 +1,53 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from io import BytesIO
+
+from iceberg.avro.codecs.codec import Codec
+
+try:
+    from zstandard import ZstdCompressor, ZstdDecompressor
+
+    class ZStandardCodec(Codec):
+        @staticmethod
+        def compress(data: bytes) -> tuple[bytes, int]:
+            compressed_data = ZstdCompressor().compress(data)
+            return compressed_data, len(compressed_data)
+
+        @staticmethod
+        def decompress(data: bytes) -> bytes:
+            uncompressed = bytearray()
+            dctx = ZstdDecompressor()
+            with dctx.stream_reader(BytesIO(data)) as reader:
+                while True:
+                    chunk = reader.read(16384)
+                    if not chunk:
+                        break
+                    uncompressed.extend(chunk)
+            return uncompressed
+
+except ImportError:
+
+    class ZStandardCodec(Codec):  # type: ignore
+        @staticmethod
+        def compress(data: bytes) -> tuple[bytes, int]:
+            raise ImportError("Zstandard support not installed, please install using `pip install pyiceberg[zstandard]`")
+
+        @staticmethod
+        def decompress(data: bytes) -> bytes:
+            raise ImportError("Zstandard support not installed, please install using `pip install pyiceberg[zstandard]`")