RFC: Adaptive bounded streaming of large column chunks in the async Parquet reader

[adriangb]

Which issue does this PR close?

None yet — opened as a draft RFC + prototype for discussion (design + benchmarks below). Happy to file a tracking issue if there is interest in the direction.

Rationale for this change

ParquetPushDecoder (and therefore ParquetRecordBatchStream) materializes every projected byte of a row group before decode starts. For wide-value columns this is painful: a measured production case has 100 MB single-row-group files whose projected columns are ~97 MB, so peak reader residency ≈ whole row group × concurrent partitions, and decode cannot start until the last byte of the slowest range arrives.

Object store GETs return a body stream, so one ranged GET over a huge chunk can be consumed incrementally: streaming does not require more requests. This PR adds an opt-in adaptive bounded streaming mode:

• Column chunks below a threshold (default 16 MiB) keep today's coalesced-materialize path bit-for-bit.
• Larger chunks are fetched with a single ranged request each whose body feeds the decoder through a bounded buffer, decoded in windows whose boundaries fall on selected-row multiples of batch_size — so the emitted batch sequence is byte-identical to the materialized path (property-tested).
• Windows reuse the existing sparse-selection machinery end to end (per-window RowSelection → scan_ranges → ColumnChunkData::Sparse → ParquetRecordBatchReader); no changes to SerializedPageReader, ArrayReader, RecordReader, or batch assembly.
• Decode of earlier windows overlaps transfer of later ones; peak residency is bounded by the window size (default 8 MiB, with a per-projected-column floor) instead of the row group size.

One fundamental tradeoff, stated up front: bounded windowed decode needs per-large-column byte progress (batches zip columns), so adjacent large columns cannot share one GET. Strict request-count equality with today's maximally-coalesced plan is impossible in exactly the wide-projection case; instead the planner guarantees a provable bounded-inflation invariant: small columns never gain requests, groups without large chunks issue exactly today's GETs, and every extra GET transfers ≥ stream_threshold bytes.

What changes are included in this PR?

• BoundedStreamingOptions + ArrowReaderBuilder::with_bounded_streaming (default off; everything is unchanged unless enabled).
• New optional AsyncFileReader::get_bytes_stream(range) -> Option<BoxFuture<'static, Result<BoxStream<'static, Result<Bytes>>>>> (default None = unsupported → materialize fallback, so every existing implementor keeps working). ParquetObjectReader implements it via get_opts.
• Push decoder: DecodingWindows state + window planner (reader_builder/windows.rs), ParquetPushDecoder::upcoming_fetch_plan() (full remaining need + streamable spans, so callers can plan coalesced fetches up front while NeedsData stays the "now" backpressure signal). Dictionary pages and window-boundary pages are ref-counted and pinned across the windows that share them.
• AdaptiveFetcher (used by ParquetRecordBatchStream, exported with an async fetch_more API for external push-decoder drivers like DataFusion): replicates object_store range coalescing to reproduce today's request plan, materializes small groups through get_byte_ranges unchanged, opens streamed requests concurrently, drains bodies via bounded channels with backpressure, aborts promptly on drop.

Companion DataFusion draft (wiring + benchmarks): apache/datafusion PRs linked below.

Benchmarks (via DataFusion benchmarks/, details in the companion PR)

Production shape (8 × 94 MB single-row-group files, 92 MB blob column, full-scan checksum query):

	GET requests	wall (SSD, median)	peak RSS
off	8	29.9 ms	1261 MB
on	8	29.7 ms	702 MB (−44%)

TPC-H SF1 / TPC-DS SF1 / ClickBench without page indexes: no request inflation, wall neutral within noise (streaming never engages below the threshold / without an offset index).

Known issue (why this is a draft): on ClickBench rewritten with page indexes, queries over large RLE_DICTIONARY string chunks regress ~+14% total (e.g. Q22 +38%): every decode window currently re-decodes the chunk's dictionary page. Fix identified but not implemented: cache decoded dictionaries across the windows of a row group. SELECT *-style projections show the predicted bounded inflation (each extra GET ≥ 16 MiB).

Are these changes tested?

Yes: mixed tiny/huge projections with request accounting, streamed chunks + RowSelection fuzz (byte-identical output vs. the materialized path across random selections/batch sizes/projections), offset/limit, cancel mid-stream. The existing parquet suite is green with the option off and on.

Are there any user-facing changes?

New opt-in API only (BoundedStreamingOptions, with_bounded_streaming, get_bytes_stream, upcoming_fetch_plan, AdaptiveFetcher). Default behavior is unchanged. When enabled, next_row_group() yields one reader per window rather than per row group (documented).

🤖 Generated with Claude Code

https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG