adriangb opened a new pull request, #10410: URL: https://github.com/apache/arrow-rs/pull/10410
# 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.com/claude-code) https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
