You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "alamb (via GitHub)" <gi...@apache.org> on 2026/10/23 21:07:14 UTC
[I] [EPIC] Improve I/O patterns for Parquet scans [datafusion]
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: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] [EPIC] Improve I/O patterns for Parquet scans [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #26682:
URL: https://github.com/apache/datafusion/issues/26682#issuecomment-6374777426
> @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
I don't think there is yet an epic for that describes how this would work end to end (aka what APIs would be exposed to the user)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] [EPIC] Improve I/O patterns for Parquet scans [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on issue #26682:
URL: https://github.com/apache/datafusion/issues/26682#issuecomment-6416879301
I found we didn't mention morsel in the epic issue. I believe the first question the design document for Layer 3 should address is: Should the strategy be placed in the decoder stream or in the morsel scheduler?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] [EPIC] Improve I/O patterns for Parquet scans [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on issue #26682:
URL: https://github.com/apache/datafusion/issues/26682#issuecomment-6416794894
> 3\. [@alamb](https://github.com/alamb) / [@adriangb](https://github.com/adriangb) / [@Dandandan](https://github.com/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: [WIP: Add parquet `progressive_io` option using `ParquetPushDecoder::scan_plan` (one-shot row group I/O with pushdown filters) #26387](https://github.com/apache/datafusion/pull/26387)
> * in progress: [Prefetch Parquet row groups with a bounded I/O budget #23492](https://github.com/apache/datafusion/pull/23492)
> * proof of concept: [POC: batch-granular Parquet scans with the push decoder and scan_plan read-ahead #24086](https://github.com/apache/datafusion/pull/24086)
These three ongoing PRs each added a configuration option:
#26387 Added `progressive_io` (bool), which ensures that only one request is made per row group when pushdown is enabled.
#23492 Added `prefetch_size` (bytes), which enables prefetching at the row group granularity.
#24086 adds `read_ahead_bytes` (bytes), which enables read-ahead at the batch level.
The semantics of the latter two are nearly identical—both involve “reading ahead by a specified number of bytes”—but they differ in granularity. If merged in the order they were submitted, users would be faced with four separate toggles: `pushdown`, `progressive`, `prefetch`, and `read_ahead`. It would be impossible to test all possible combinations.
My recommendation:
#26387 Merge first. This is essentially a bug fix: `pushdown` should not alter the I/O mode. Only after merging will `pushdown` have the opportunity to be enabled by default.
Merge #23492 and #24086 into a single read-ahead budget parameter. The granularity is determined by the decoder’s capabilities: before `arrow-rs#11234` was merged, it was based on row groups; after the merge, it is based on batches.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org