fix: bound snappy uncompressed length to avoid OOM on untrusted block (#5)

## Problem

`DataFileReaderBase::readDataBlock()` decompressed a snappy-coded OCF
block with the **std::string overload** of `snappy::Uncompress()`, which
reads the declared uncompressed length from the varint prefix of the
block and `resize()`s the destination to that full length **before
decompressing a single byte**. The length is attacker-controlled for
untrusted Avro input (Iceberg manifest/data files, `... FORMAT Avro` in
PackDB). A tiny crafted block can declare a multi-GiB uncompressed
length → huge allocation → OOM before the reader notices the compressed
payload is far shorter.

Surfaced by PackDB's `avro_input` libFuzzer harness: two independent
crash inputs (220 B and ~1 KB) each drove a ~2.4 GiB `malloc`. Same
class as the `decodeString`/`decodeBytes` OOM fixed in #4, different
code path. Present verbatim in Apache Avro C++ `main` — not fixed
upstream.

## Fix

Read the declared length without allocating (`GetUncompressedLength`)
and pick a decode strategy — **this never rejects any block**:

- **modest length (≤ 64 MiB):** decompress in one shot into a pre-sized
buffer — the original fast path, no extra copy (the overwhelmingly
common case);
- **implausibly large length:** decompress through `SnappyStringSink`,
whose scattered path allocates the output in bounded ~64 KiB increments
as bytes are produced, so a bogus length aborts once the compressed
input is exhausted (allocating only what was really produced), while a
genuinely large block still decodes — just incrementally.

This keeps the bounded-growth safety of a pure-sink approach (same
spirit as #4 — allocate proportional to the bytes actually present)
**with no fixed expansion cap and no regression on the common fast
path**. The 64 MiB switch point bounds the worst-case up-front
allocation a crafted block can force onto the fast path and comfortably
exceeds real Avro block sizes.

### Why not a pure Sink (the obvious approach)?

Decompressing everything through the sink is safe but slow: snappy's
scattered path does an extra full copy of the output plus many 64 KiB
allocations instead of one contiguous decompress. That is a real
regression on multi-MiB blocks (see below), which is the hot path for
Iceberg reads. The hybrid confines the sink to the only case that needs
it.

### Why not a fixed expansion cap?

A "reject if uncompressed/compressed > N" cap can reject legitimately
highly-compressible data (snappy can exceed any fixed ratio on
repetitive input). The strategy-switch rejects nothing — large blocks
still decode, just via the bounded path.

## Performance

Microbenchmark of the changed operation in isolation (snappy decompress
only, `clang++-22 -O2 -DNDEBUG`, in-process; each row ~fixed total work
with warmup). `flat` = original std::string overload, `pure-sink` =
decompress everything through the sink, `hybrid` = this PR.

```
case                    flat ns    sink ns  hybrid ns  sink/flat   hyb/flat
16 KiB  ratio~2x            605        674        552      1.11x      0.91x
16 KiB  ratio~50x           777        805        760      1.04x      0.98x
256 KiB ratio~2x           9210      10516       9155      1.14x      0.99x
256 KiB ratio~10x         20204      21534      20167      1.07x      1.00x
1 MiB   ratio~4x          41719      65726      40163      1.58x      0.96x
4 MiB   ratio~4x         293636     352884     276994      1.20x      0.94x
16 MiB  ratio~4x        1720443    3527717    1537172      2.05x      0.89x
128 MiB ratio~4x       22060868   66452357   66990924      3.01x      3.04x
```

- **hybrid vs flat:** within measurement noise (0.89–1.00x) for every
block ≤ 64 MiB — i.e. no regression across the entire realistic Avro
block-size range.
- **pure-sink vs flat:** the approach this PR avoids — 1.1x on small
blocks but **1.6x at 1 MiB and ~2.0x at 16 MiB**.
- Only the 128 MiB case (> the 64 MiB switch point) takes the
incremental path and matches pure-sink (~3x). That regime is either a
bomb (which aborts early, never allocating the declared size) or a
genuinely huge block (correctness preserved; it would allocate that much
either way).

## Test

- Both fuzzer crash inputs run clean (exit 0, 0–1 ms); the bombs declare
> 64 MiB so they take the bounded sink path.
- Well-formed snappy Avro fixtures (`weather-snappy`, `nulls.snappy`,
`datapage_v2.snappy`, `alltypes_plain.snappy`) decode identically — the
sink appends exactly the bytes snappy produces, and the flat path is
unchanged.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: moshap-firebolt

lang/c++/impl/DataFile.cc

@@ -31,6 +31,7 @@
 
 #ifdef SNAPPY_CODEC_AVAILABLE
 #include <snappy.h>
+#include <snappy-sinksource.h>
 #endif
 
 namespace avro {
@@ -52,6 +53,31 @@ const string AVRO_ZSTD_CODEC("zstandard");
 
 #ifdef SNAPPY_CODEC_AVAILABLE
 const string AVRO_SNAPPY_CODEC = "snappy";
+
+// A snappy Sink that appends decompressed bytes into a std::string. It
+// overrides only Append() and leaves GetAppendBufferVariable() at the base
+// default (hands back only the tiny scratch buffer), so
+// snappy::Uncompress(Source*, Sink*) sees a scratch smaller than the declared
+// uncompressed length and takes its block-by-block scattered path instead of
+// pre-sizing one buffer to the full declared length. The destination then grows
+// as bytes are actually decompressed -- snappy hands over each block via
+// Append() and the std::string grows with its usual amortized (doubling)
+// strategy -- driven by the real output rather than the untrusted length prefix.
+// That is what prevents the OOM: the std::string overload of snappy::Uncompress
+// resize()s the destination to the full declared length up front, so a tiny
+// crafted block claiming multi-GiB blows up before a byte is read; growing as
+// bytes arrive instead means a bogus length simply fails once the compressed
+// input is exhausted, having allocated only on the order of what was really
+// produced -- the same bounded-growth strategy as the BinaryDecoder
+// length-prefix fix.
+class SnappyStringSink : public snappy::Sink {
+public:
+    explicit SnappyStringSink(std::string& dest) : dest_(dest) {}
+    void Append(const char* bytes, size_t n) override { dest_.append(bytes, n); }
+
+private:
+    std::string& dest_;
+};
 #endif
 
 const size_t minSyncInterval = 32;
@@ -471,8 +497,44 @@ void DataFileReaderBase::readDataBlock()
         int b4 = compressed_[len - 1] & 0xFF;
 
         checksum = (b1 << 24) + (b2 << 16) + (b3 << 8) + (b4);
-        if (!snappy::Uncompress(reinterpret_cast<const char*>(compressed_.data()),
-                len - 4, &uncompressed)) {
+
+        // The block's declared uncompressed length is a varint prefix and is
+        // attacker-controlled for untrusted input. The std::string overload of
+        // snappy::Uncompress resize()s the destination to that full length before
+        // decompressing a single byte, so a tiny crafted block claiming multi-GiB
+        // drives a huge allocation up front -- a decompression bomb -> OOM.
+        //
+        // Read the declared length without allocating (GetUncompressedLength) and
+        // pick a strategy from it -- this never rejects any block:
+        //  - modest length: decompress in one shot into a pre-sized buffer, the
+        //    fast path with no extra copy (the overwhelmingly common case);
+        //  - implausibly large length: decompress through SnappyStringSink,
+        //    whose scattered path grows the destination as bytes are produced
+        //    (never pre-sizing to the declared length), so a bogus length aborts
+        //    once the compressed input is exhausted -- having allocated only on
+        //    the order of what was really produced -- while a genuinely large
+        //    block still decodes, just incrementally.
+        // maxFlatSnappyDecode bounds the worst-case up-front allocation a crafted
+        // block can force onto the fast path; it comfortably exceeds real Avro
+        // block sizes, so legitimate data keeps the fast path.
+        uncompressed.clear();
+        const size_t maxFlatSnappyDecode = size_t(64) << 20; // 64 MiB
+        size_t uncompressedLen = 0;
+        bool decoded;
+        if (snappy::GetUncompressedLength(
+                    reinterpret_cast<const char*>(compressed_.data()), len - 4,
+                    &uncompressedLen) &&
+                uncompressedLen <= maxFlatSnappyDecode) {
+            decoded = snappy::Uncompress(
+                    reinterpret_cast<const char*>(compressed_.data()), len - 4,
+                    &uncompressed);
+        } else {
+            SnappyStringSink snappySink(uncompressed);
+            snappy::ByteArraySource snappySource(
+                    reinterpret_cast<const char*>(compressed_.data()), len - 4);
+            decoded = snappy::Uncompress(&snappySource, &snappySink);
+        }
+        if (!decoded) {
             throw Exception(
                     "Snappy Compression reported an error when decompressing");
         }