alamb opened a new issue, #26682: URL: https://github.com/apache/datafusion/issues/26682
### Is your feature request related to a problem or challenge? I feel like there is no good writeup of the high level project / goals for the parquet reader / IO system that @adriangb, @Dandandan, @hhhizzz, @HippoBaro, @zhuqi-lucas @etseidl @XiangpengHao @Rich-T-kid @haohuaijin @neilconway and @mbutrovich and others has been working on recently, so I wanted to try and write it up Today the DataFusion Parquet scan reads an entire row group into memory in one request, decodes it, and only then fetches the next row group. This is a hard coded trade-off between I/O latency and memory that users can not change. The user visible effects are: 1. **Time to first batch** scales with row group size: on object storage the first rows of a large row group arrive only after the whole row group has been downloaded. 2. **Peak memory** per partition is at least one full row group (often 100MB+), regardless of `batch_size` or the memory pool limit, and is not reclaimable. 3. **I/O and decode do not overlap** within a partition: the fetch of row group N+1 does not start until row group N has been fully decoded. 4. **Enabling `pushdown_filters` changes the I/O pattern** to many smaller sequential requests (filter columns first, then projection columns). That can be slower on high latency stores and is one reason it is still off by default (see #20324, #24393, #24527). 5. **No way to tune for the storage system**: a local SSD or in-memory cache would benefit from more, smaller reads; a remote object store from fewer, larger reads. The only knob today is `metadata_size_hint`. * Items 1 and 2 are tracked at the reader level in apache/arrow-rs#11234. * Item 3 is already tracked by https://github.com/apache/datafusion/issues/18470 Visually, here is how the DataFusion reader read two row groups, RG1 and RG2 today: ```text time ──────────────────────────────────────────────────────────────────────────────────────────────────────────▶ ┌───────────────────────────┐ ┌───────────────────────────┐ I/O │ fetch ALL of RG 1 │ ~ idle ~ │ fetch ALL of RG 2 │ ~ idle ~ │ (one request, ~100 MB) │ │ (one request, ~100 MB) │ └───────────────────────────┘ └───────────────────────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌──────────────────────┐ ┌───────────────────────┐ CPU ~ idle ~ │ decode RG 1 │ ~ idle ~ │ decode RG 2 │ (waiting for I/O) │ (batch by batch) │ (waiting for I/O) │ (batch by batch) │ └──────────────────────┘ └───────────────────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌──────────────────────┐ ┌───────────────────────┐ Output no output │ batches of RG 1 │ no output │ batches of RG 2 │ (until ALL of RG 1 fetched) │ emitted │ (until ALL of RG 2 fetched) │ emitted │ └──────────────────────┘ └───────────────────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌───────────────────────────────────────┐ ┌────────────────────────────────────────┐ Memory │ RG 1 fully buffered (~100 MB) │ │ RG 2 fully buffered (~100 MB) │ │ from first byte fetched until last │ │from first byte fetched until last batch│ └───────────────────────────────────────┘ └────────────────────────────────────────┘ ``` However, depending on your systems IO speed and RAM budget/buffering and other factors you may want different patters. For example, on a object store which has high throughput but low latency you probably want to start next fetch as early as possible ``` time ──────────────────────────────────────────────────────────────────────────────────────────────────────────▶ ┌──────────────────────────────────────┐┌──────────────────────────────────────┐ I/O │fetch ALL of RG 1 (one large request) ││fetch ALL of RG 2, issued IMMEDIATELY │ ~ idle ~ │ long latency, high throughput ││ while RG 1 is still being decoded │(or start reading next RG) └──────────────────────────────────────┘└──────────────────────────────────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌────────────────────────┐ ┌────────────────────────┐ CPU ~ idle ~ │ decode RG 1 │ short wait │ decode RG 2 │ (unavoidable: first request in flight) │(overlaps fetch of RG 2)│ (I/O bound) │ (starts as soon as it │ └────────────────────────┘ └────────────────────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌────────────────────────┐ ┌────────────────────────┐ Output no output │ batches of RG 1 │ │ batches of RG 2 │ (waiting for the first request) │ emitted │ │ emitted │ └────────────────────────┘ └────────────────────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌────────────────────────────────────────────────────────────────┐ │RG 1 buffered (~100 MB) from first byte until last batch decoded│ │ │ Memory └────────────────────────────────────────────────────────────────┘ ┌────────────────────────────────────────────────────────────────┐ │ RG 2 buffered (~100 MB): overlaps RG 1, so peak = 2 row groups │ │ (~200 MB) │ └────────────────────────────────────────────────────────────────┘ ``` But on a LOCAL SSD, with a tiny latency per request, interleaving small fetches with decode is probably a better strategy: ``` time ──────────────────────────────────────────────────────────────────────────────────────────────────────────▶ ┌───────────┐┌───────────┐┌───────────┐┌───────────┐┌───────────┐┌───────────┐┌───────────┐ I/O │fetch RG 1 ││fetch RG 1 ││fetch RG 1 ││fetch RG 1 ││fetch RG 2 ││fetch RG 2 ││fetch RG 2 │ done │ pages 1-2 ││ pages 3-4 ││ pages 5-6 ││ pages 7-8 ││ pages 1-2 ││ pages 3-4 ││ pages 5-6 │ └───────────┘└───────────┘└───────────┘└───────────┘└───────────┘└───────────┘└───────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌───────────┐┌───────────┐┌───────────┐┌───────────┐┌───────────┐┌───────────┐┌───────────┐ CPU wait │decode RG 1││decode RG 1││decode RG 1││decode RG 1││decode RG 2││decode RG 2││decode RG 2│ (1st pages) │ pages 1-2 ││ pages 3-4 ││ pages 5-6 ││ pages 7-8 ││ pages 1-2 ││ pages 3-4 ││ pages 5-6 │ └───────────┘└───────────┘└───────────┘└───────────┘└───────────┘└───────────┘└───────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌─────────────────────────────────────────────────────────────────────────────────────────┐ Output no output │ batches emitted continuously: the first batch appears after a few pages, │ │ not after a whole row group │ └─────────────────────────────────────────────────────────────────────────────────────────┘ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌──────────────────────────────────────────────────────────────────────────────────────────────────────┐ Memory │ only the read-ahead buffer (e.g. 8 MB): the pages of the next batch or two, │ │ never a whole row group │ └──────────────────────────────────────────────────────────────────────────────────────────────────────┘ ``` There are many other combinations like if you have a shared cache, for example ### Describe the solution you'd like Make the Parquet scan I/O pattern configurable rather than single hard coded reader implementation. Some ideas: - Use the `ParquetPushDecoder` scan plan so DataFusion, not arrow-rs, decides which byte ranges to fetch and when (#26387, apache/arrow-rs#11234). - Fetch and decode at batch (or page) granularity with bounded read-ahead, so the first batch is returned after a few MB instead of a full row group and peak memory is bounded by the read-ahead budget rather than the row group size (#24086). - Prefetch the next row group (or next pages) while the current one is being decoded, so I/O and CPU overlap within a partition. - Keep the one-request-per-row-group pattern as an option when `pushdown_filters` is enabled, so filter pushdown never issues more requests than the default path (#24393, #24527). - Expose a small number of settings (for example a read-ahead budget in bytes and a request coalescing threshold) with defaults tuned for object storage, and document them in the tuning guide (#16178). - Add tests that assert on the object store request pattern (count and size of ranges) so regressions are caught in CI, as arrow-rs does (apache/arrow-rs#7971). ### Describe alternatives you've considered The high level idea I think we are pursuing is: 1. @adriangb / @sunchao / @zhuqi-lucas: arrow-rs lets the ParquetPushDecoder communicate what it will do so the caller can plan I/O - done: https://github.com/apache/arrow-rs/pull/10555 (@adriangb, @sunchao) - done: https://github.com/apache/arrow-rs/pull/10158 (@zhuqi-lucas) 2. @adriangb / @HippoBaro: arrow-rs lets the decoder decode a row group incrementally, batch by batch, as its pages arrive - epic: https://github.com/apache/arrow-rs/issues/11234 (@HippoBaro, @adriangb), 3. @alamb / @adriangb / @Dandandan: make the I/O policy in DataFusion extensible, so the scan can choose between one request per row group, prefetch with a byte budget, or batch-granular fetch depending on the storage system - in progress: https://github.com/apache/datafusion/pull/26387 - in progress: https://github.com/apache/datafusion/pull/23492 - proof of concept: https://github.com/apache/datafusion/pull/24086 ### Additional context DataFusion: - #20324 - #24393 - #24527 - #23978 - #18362 - #19983 - #26387 - #24086 arrow-rs: - apache/arrow-rs#8000 - apache/arrow-rs#11234 - apache/arrow-rs#11431 - apache/arrow-rs#7456 ### Willingness to contribute None -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
