feat(parquet): opt-in streaming reads for large data pages

[joechenrh]

Rationale for this change

Addresses #865. Today the reader materializes each full uncompressed data page before decoding, so peak memory scales with the page's uncompressed size — a problem for files with very large data pages (e.g. a 1 GB compressed / 1.4 GB uncompressed page). This adds an opt-in path that keeps peak memory near the requested batch size for the common large-page shape.

Approach was discussed and agreed on the issue.

What changes are included in this PR

Opt-in via ReaderProperties.EnablePageStreaming (default false, no behavior change). When set, eligible data pages skip full-page decompression and decode their values incrementally; every ineligible page falls back to the existing materialized path.

Scope (first cut):

• Data Page V1 + V2, PLAIN encoding
• Physical types byte_array and fixed_len_byte_array
• Codecs UNCOMPRESSED / GZIP / BROTLI / ZSTD (explicit allowlist — Snappy is a raw block whereas its NewReader is the framed format, and LZ4_RAW has no streaming reader)

Design:

• parquet/internal/encoding/streaming: a small ValueBuffer — an incremental byte source that reads from an io.Reader, slides as values are consumed, grows to fit a single oversized value, and owns the page's stream (drained + closed on release).
• The materialized PlainByteArrayDecoder / PlainFixedLenByteArrayDecoder keep their []byte logic byte-for-byte, gaining only a src field + a one-line guard in Decode/Discard. Streaming decode lives beside them; SetData resets src, so a cached decoder reused for a materialized page reverts. Other decoders are untouched.
• Page reader: for eligible pages, V2 levels are read raw (uncompressed, header byte-lengths) and V1 levels are peeled off the decompressed stream using the same length rules as LevelDecoder.SetData; values become a ValueBuffer over the compressed region. On page release the underlying stream is drained + closed so the reader lands on the next page header even if values were skipped.
• Column reader: the same cached PLAIN decoder is fed via SetSource (streaming) or SetData (materialized). The public Page/DataPage interfaces are unchanged (the streaming accessor is reached via an unexported interface, so external implementations don't break).

Are these changes tested?

Yes.

• All existing encoding tests pass unchanged (materialized parity); the streaming decoders are additionally exercised with a one-byte-at-a-time reader and oversized values.
• End-to-end round-trip (parquet/file/streaming_page_test.go): streaming vs materialized reads are asserted identical across V1/V2 and each codec, for required and nullable columns (the nullable case covers the V1 def-level peel + streaming DecodeSpaced), including values larger than a page and the stream buffer.

Are there any user-facing changes?

One new opt-in field, ReaderProperties.EnablePageStreaming (default false). No change to existing behavior or public interfaces.

Follow-ups (out of scope here): fixed-width numeric PLAIN types, raw-Snappy and LZ4_RAW streaming, and a memory benchmark.

[zeroshade]

Thanks for tackling #865 — the incremental-decode design is clean and the commit series is easy to follow.

I reviewed the read path in depth and ran the new streaming tests plus the existing encoding and parquet/file suites with -race; all pass and materialized parity holds. The drain-before-advance ordering in Next() (release the previous page, then read the next header), the codec allowlist (correctly excluding raw-block Snappy even though it implements StreamingCodec), the V1 level byte-length parity with LevelDecoder.SetData, and the DecodeSpaced -> Decode streaming dispatch all check out.

A few things I'd like addressed before merge:

1. Page.Data() contract on streaming pages (blocking). A streaming page's buf holds only the rep/def level region, so the public Page.Data() returns truncated bytes when a page is obtained via the public GetColumnPageReader with EnablePageStreaming set. Since internal callers now go through levelBuffer()/valueBuffer(), I'd like us to either gate streaming to the internal Column() path or document Data() explicitly (see inline).
2. Unbounded allocation from an untrusted RLE length prefix in readLevelData.
3. Test coverage for BIT_PACKED levels, the skip/Discard path, repeated columns, and ineligible-codec fallback.

Details inline. Nice work overall — with the Data() semantics resolved this is close.

[zeroshade]

Enabling streaming here also affects the public GetColumnPageReader: a streaming DataPageV1/V2's buf holds only the rep/def level region, so Page.Data() returns just the levels instead of the whole page. That's a silent semantic change to a public method for anyone reading Data() off a data page with EnablePageStreaming set.

Now that the internal callers read levels/values through levelBuffer()/valueBuffer(), the original blocker to keeping GetColumnPageReader materialized-only (the revert in 2f5aab9) looks gone. Could we gate streaming to the internal Column() path, or — failing that — document Data() as level-only/undefined for streaming pages? This is the main thing I'd like resolved before merge.

[joechenrh]

Add document for both EnablePageStreaming and Data() to clarify the change. My judgment is as follows. I wonder if you have any opinions:

I find that pqarrow also uses GetColumnPageReader, so I prefer explicitly document this semantic change. Besides, since this new option is off by default, existing downstream user codes shouldn't be affected when bumping the package. When a user (or an AI agent) wants to enable the option, they can find this changes from the comments from EnablePageStreaming.

[zeroshade]

n comes straight from the 4-byte RLE length prefix in the page and is used to make([]byte, n) before any bounds check, so a corrupt/malicious V1 page with a bogus prefix (e.g. ~4 GiB) triggers a huge allocation before the following io.ReadFull fails.

The materialized LevelDecoder.SetData guards this with nbytes > len(data)-4 against the already-bounded page buffer; the streaming path has no equivalent ceiling (V1 carries no per-level byte length in the header). Please bound n (e.g. against lenCompressed/lenUncompressed) before allocating.

[joechenrh]

Yeah, updated to bound the allocation with uncompressed size in 3cd0076

[zeroshade]

Minor/consistency: LevelDecoder.SetData computes this length with the overflow-checked shared_utils.Mul(nbuffered, bitWidth), whereas here it's a raw int64(numValues) * int64(bitWidth). Not exploitable for real level bit-widths, but matching the sibling code's checked multiply would keep the two paths in sync.

[joechenrh]

Nice catch, I didn't notice that my AI has modified this part during reviewing. Updated in aeeb992

[zeroshade]

Two notes on the backing buffer:

• It's a fresh 1 MiB (defaultStreamBufferSize) allocation per streaming page via the Go allocator rather than the configured memory.Allocator, so it's untracked by CheckedAllocator and adds GC churn on files with many eligible pages. Consider pooling it (and/or sizing relative to the read batch).
• Worth calling out in the EnablePageStreaming doc that peak memory is really batch + ~1 MiB + largest-value growth + levels, not just the batch size.

[joechenrh]

In 1b82270, I cap the size of the buffer with the uncompressed size and manage all the buffers with Allocator. And the comment for EnablePageStreaming is also aligned with the new behavior.

[joechenrh]

Oh, maybe we can't use allocator here.

[joechenrh]

Updated, we should store the buffer inside the valueBuffer, and resue them for each Decode call.

[zeroshade]

Since this ships new decode/peel logic, a few gaps I'd like covered:

• BIT_PACKED levels — only RLE levels are exercised, but readLevelData has a BIT_PACKED branch.
• Skip path — discardStreaming (via SeekToRow/skipRows) isn't covered; the tests only do full ReadBatch.
• Repeated columns (maxRepLevel > 0) — the V1 rep-level peel in readLevelData is untested (current cases are flat required/optional).
• Ineligible-codec fallback — a case with Snappy or LZ4_RAW + EnablePageStreaming=true asserting correct (materialized) reads would lock in the allowlist behavior.

[zeroshade]

Thanks for tackling #865 — the incremental-decode design is clean and the commit series is easy to follow.

I reviewed the read path in depth and ran the new streaming tests plus the existing encoding and parquet/file suites with -race; all pass and materialized parity holds. The drain-before-advance ordering in Next() (release the previous page, then read the next header), the codec allowlist (correctly excluding raw-block Snappy even though it implements StreamingCodec), the V1 level byte-length parity with LevelDecoder.SetData, and the DecodeSpaced -> Decode streaming dispatch all check out.

A few things I'd like addressed before merge:

1. Page.Data() contract on streaming pages (blocking). A streaming page's buf holds only the rep/def level region, so the public Page.Data() returns truncated bytes when a page is obtained via the public GetColumnPageReader with EnablePageStreaming set. Since internal callers now go through levelBuffer()/valueBuffer(), I'd like us to either gate streaming to the internal Column() path or document Data() explicitly (see inline).
2. Unbounded allocation from an untrusted RLE length prefix in readLevelData.
3. Test coverage for BIT_PACKED levels, the skip/Discard path, repeated columns, and ineligible-codec fallback.

Details inline. Nice work overall — with the Data() semantics resolved this is close.

Accidental duplicate submission — dismissing this copy; the identical active review remains.

[joechenrh]

I have tried my best to control the size (as well as the scope) of this PR, but it still exceeds 1,000 1,500 rows 😢. Since it's not a small fix, I think I need more testing before merging it.

[zeroshade]

Re-review of the latest push — thanks for the very thorough follow-up. All three of my blocking items are addressed, and I verified locally (built ./parquet/..., ran the TestPageStreaming* suite with PARQUET_TEST_DATA set — green):

1. Page.Data() contract — documented on both EnablePageStreaming and Data(), with streaming kept to the eligible internal path via columnCanStream. Your rationale (off by default, pqarrow uses the same accessor, documented semantic) is sound.
2. Unbounded RLE level allocation — now bounded by the uncompressed page size in the streaming level read (setStreamingValues), and the BIT_PACKED length uses the checked multiply. Both inline points resolved.
3. Coverage — TestPageStreaming{ByteArraySkip,FLBASkip,Repeated,ReadBatchInPageRepeated,IneligibleCodecFallback,AllocatorBalance,PeakMemoryBounded,PageReaderCloseFreesStream} cover the skip / repeated / ineligible-codec / allocator-tracking paths I asked for.

Scope note: this pass verified the above + ran the streaming suite; I haven't deep-audited the newer perf commits (zstd decoder pooling, chunked FLBA skip, V2 streaming hardening). Since it's large and you mentioned wanting more testing, I'll hold a formal approve until you say it's ready — ping me for a final pass. Great iteration.