max_compressed_length
def max_compressed_length(n: Int) -> IntUpper bound on the compressed size of an n-byte input — matches the C++ reference's MaxCompressedLength(n) = 32 + n + n/6.
snappy.raw
Snappy raw block format: a leading varint giving the uncompressed length, followed by a stream of literal / copy elements. See https://github.com/google/snappy/blob/main/format_description.txt.
Compression is a single-pass greedy LZ77 matcher: a flat hash table maps
the multiplicative hash of each 4-byte window (same constant the C++
reference uses, 0x1e35a7bd) to the most recent position with that hash,
and any 4-byte match found there is greedily extended and emitted as a
copy. This is the reference's "fast" strategy (no lazy matching, no hash
chaining) — it trades some ratio for a simple, branch-light inner loop.
Functions
uncompressed_length
def uncompressed_length(data: Span[UInt8]) -> IntDecode the leading varint (the decompressed size) without decompressing the rest of data.
compress
def compress(data: Span[UInt8]) -> List[UInt8]Compress data into a Snappy raw block.
decompress_into
def decompress_into[origin: MutOrigin, //](data: Span[UInt8], dst: Span[UInt8, origin]) -> IntDecompress the raw Snappy block data into dst, which must be at least uncompressed_length(data) bytes. Returns the number of bytes written. Raises on any structural corruption.
Literals and non-overlapping copies move 16 bytes at a time. Both are
allowed to write past the end of the element they are copying, because
the write always lands inside the part of dst the block has already
been checked to fit in; the bytes written past the element are
overwritten by whatever comes next. Only the last few bytes of the output,
where that slack runs out, take the byte-at-a-time path.
decompress
def decompress(data: Span[UInt8]) -> List[UInt8]Decompress a Snappy raw block into a freshly allocated List[UInt8].
validate
def validate(data: Span[UInt8]) -> BoolBest-effort check: does data parse as a well-formed raw Snappy block?