Repository navigation
[EPIC] Improve I/O patterns for Parquet scans #26682
Description
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:
- 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.
- Peak memory per partition is at least one full row group (often 100MB+), regardless of
batch_sizeor the memory pool limit, and is not reclaimable. - 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.
- Enabling
pushdown_filterschanges 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 [EPIC] Fix performance regressions when enabling parquet filter pushdown (late materialization) #20324, Parquet Pushdown Filters: Way to keep the I/O pattern the same when pushdown filters are enabled #24393, Filtered Parquet scans issue additional object-store requests after ParquetPushDecoder migration #24527). - 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 [EPIC] Batch-granular fetch and decode in ParquetPushDecoder arrow-rs#11234.
- Item 3 is already tracked by [Parquet] Pre-fetch the next row group when reading parquet files #18470
Visually, here is how the DataFusion reader read two row groups, RG1 and RG2 today:
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
ParquetPushDecoderscan plan so DataFusion, not arrow-rs, decides which byte ranges to fetch and when (WIP: Add parquetprogressive_iooption usingParquetPushDecoder::scan_plan(one-shot row group I/O with pushdown filters) #26387, [EPIC] Batch-granular fetch and decode in ParquetPushDecoder 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 (POC: batch-granular Parquet scans with the push decoder and scan_plan read-ahead #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_filtersis enabled, so filter pushdown never issues more requests than the default path (Parquet Pushdown Filters: Way to keep the I/O pattern the same when pushdown filters are enabled #24393, Filtered Parquet scans issue additional object-store requests after ParquetPushDecoder migration #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 (EPIC: Tuning guide for DataFusion #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 ([Parquet] Add tests for IO/CPU access in parquet reader arrow-rs#7971).
Describe alternatives you've considered
The high level idea I think we are pursuing is:
-
@adriangb / @sunchao / @zhuqi-lucas: arrow-rs lets the ParquetPushDecoder communicate what it will do so the caller can plan I/O
-
@adriangb / @HippoBaro: arrow-rs lets the decoder decode a row group incrementally, batch by batch, as its pages arrive
-
@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: WIP: Add parquet
progressive_iooption usingParquetPushDecoder::scan_plan(one-shot row group I/O with pushdown filters) #26387 - in progress: Prefetch Parquet row groups with a bounded I/O budget #23492
- proof of concept: POC: batch-granular Parquet scans with the push decoder and scan_plan read-ahead #24086
- in progress: WIP: Add parquet
Additional context
DataFusion:
- [EPIC] Fix performance regressions when enabling parquet filter pushdown (late materialization) #20324
- Parquet Pushdown Filters: Way to keep the I/O pattern the same when pushdown filters are enabled #24393
- Filtered Parquet scans issue additional object-store requests after ParquetPushDecoder migration #24527
- Deferred page-index load in the Parquet opener bypasses FileMetadataCache, causing repeated uncached I/O when the skip heuristic doesn't fire #23978
- Allow support for switching off metadata cache #18362
- Reuse file descriptors for LocalFileSystem object storage #19983
- WIP: Add parquet
progressive_iooption usingParquetPushDecoder::scan_plan(one-shot row group I/O with pushdown filters) #26387 - POC: batch-granular Parquet scans with the push decoder and scan_plan read-ahead #24086
arrow-rs:
- [Epic] Parquet Reader Improvement Plan / Proposal - July 2025 arrow-rs#8000
- [EPIC] Batch-granular fetch and decode in ParquetPushDecoder arrow-rs#11234
- [Parquet] ScanPlan docs: document the read-ahead buffering contract, and deprecate redundant ParquetPushDecoder APIs arrow-rs#11431
- [EPIC] Faster performance for parquet predicate evaluation for non selective filters arrow-rs#7456
Willingness to contribute
None
Reacted by Hippolyte Barraud, Xiangpeng Hao, Huaijin, Qi Zhu and RIchard Baah
Activity
alamb commentedon Oct 23, 2026
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)
xudong963 commentedon Oct 26, 2026
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, andread_ahead. It would be impossible to test all possible combinations.My recommendation:
#26387 Merge first. This is essentially a bug fix:
pushdownshould not alter the I/O mode. Only after merging willpushdownhave 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#11234was merged, it was based on row groups; after the merge, it is based on batches.xudong963 commentedon Oct 26, 2026
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?
alamb commentedon Oct 26, 2026
Sounds good -- I agree.
I think once arrow 60.1 is out (eta in a few days) we'll have that one ready to merge
I think the ideal budget is not based on rows / row groups, but instead on "target_buffer_memory" as that is the limiting resource that prevents infinite buffering and what will run out earlier.
Things like "num_row_groups" is just a proxy for buffer / memory usage.
It may be that "read_ahead_bytes" in #24086 is the right measure
alamb commentedon Oct 26, 2026
that is an excellent question and I don't have any idea -- it might make sense to try both to see what they look like
My intuition is that it is the morsel scheduler -- the decode stream should try and read data for target_batch size chunks as that is what it wants to decode. However we would need to find some way to coalesce multiple requests maybe
adriangb commentedon Oct 26, 2026
👍🏻, yep this is why I chose read ahead bytes in #24086.
It might be reasonable to say that we only read ahead within a row group and thus setting
read_ahead_bytes = 32Gior something large (could be0or-1) is equivalent toprogressive_io = falsealamb commentedon Oct 26, 2026
I am thinking progressive_io is sort of a policy choice rather than a resource cap (and the ideal setting is based on how rows that match the predicate are distributed over the input pages).
In general I think the parquet reader should be reading ahead to the next row group if there is memory budget for it, rather than being restricted to just a single row group
XiangpengHao commentedon Oct 28, 2026
fwiw, I'm working on a very related thing as part of my work at SpiralDB: https://github.com/XiangpengHao/t4
t4 is not ready to use yet, will make blog posts & announcements when it's ready in a few weeks.
The high level idea is that
object_storetrait already give us a lot of opportunity to do io optimizations, and t4 especially focus on super fast SSD caching (using io_uring), with good enough SSDs (which you can rent from AWS), t4 can serve Parquet/Vortex data from disk as fast as from memory.