pyiceberg/avro/codecs/snappy_codec.py (32 lines of code) (raw):

# 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 pyiceberg.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]`")