Repository navigation
Decouple IO and CPU operations in the Parquet Reader (push decoder) #7983
Description
(This is based on discussions with @crepererum and @XiangpengHao over the last few days)
Is your feature request related to a problem or challenge?
After working with the Parquet Reader, I have concluded it is not realistic to implement smart prefetching of row groups or pages because the knowledge of what data is needed next is not available from the reader.
This means in practice, users either have to
- buffer the entire file in memory or local SSD (which is what we do in InfluxDB)
- read the data serially with multiple smaller requests while decoding.
Buffering the entire file consumes significant resource and avoids some of the benefits inherent in columnar IO formats.
Reading the data with multiple smaller requests is ok on a low latency local file system or when you have a bunch of outstanding requests (where the latency can be hidden by other work), but it is pretty bad for remote object stores or remote filesystems (e.g. S3, HDFS, etc) where the latency of each request is high.
There are also several other things which I find annoying about the current design which i will rant about:
- Row filters are evaluated eagerly (before the reader is created) see docs ("Note: this will eagerly evaluate any RowFilter before returning") which can potentially "front load" significant IO and CPU operations before client code might expect
- There are different sync and async readers, which makes it hard (impossible?) to write tests that cover both without duplicating code (see Add option to control predicate cache, documentation,
ArrowReaderMetricsand tests XiangpengHao/arrow-rs#7) - The internal implementation (e.g. for evaluating filters) is duplicated between the sync and async readers, which makes it hard to maintain and. Example: The initial predicate cache in Speed up Parquet filter pushdown v4 (Predicate evaluation cache for async_reader) #7850 does not work for the sync reader 😬
- The
ArrowReaderBuilderis a templated typedef . So the innocent soundingParquetRecordBatchReaderBuilderis actuallyArrowReaderBuilder<SyncReader<T>>;and to actually use that you need to figure outT: ChunkReader. The multiple template levels makes it challenging to use (why is the the underlying type of the IO reader needed to to configure things like aRowSelection?) - I found the async API confusing: you have to implement
AsyncFileReaderwhich combines IO and optional Parquet metadata caching. It is quite tricky to implement if you want to read from some different remote source. - It is also a pain to test the async reader (as you can't easily do so with in memory
Bytes). You can see this pain as there are at least two implementations of AsyncFileReader for Bytes in the codebase. (I also had to add a third in Add option to control predicate cache, documentation,ArrowReaderMetricsand tests XiangpengHao/arrow-rs#7)
Root Cause: Pull Based Decoder ?
I think the core reason for many of the above challenges is the current design of the Parquet Reader as a "pull" based decoder where the CPU operations (decoding) drives the IO operations (e.g reading from the file).
The way the reader works is:
- the caller asks for the next batch,
- the reader internally (may) make an I/O call to the underyling reader (either async or non async) if it needs needs more data
- the reader then decodes from buffered data
let reader = ParquetRecordBatchBuilder::new(io_reader)
// more configuration here
.build();
// The call to `reader.next()` MAY do IO, depending on the readers state, but
// the outer loop has no way to know what IO will be needed next
while let Some(batch) = reader.next() {
// process
}This design makes it challenging to add more sophisticated strategies. For example, @masonh22 proposed to add pre-fetching of row groups to the reader #6676. Among other challenges with that PR is that automatically pre-fetching row group data helps for some cases (e.g. reading remote object store), but can be pure cost on others (e.g. increases memory usage when the data is already in memory)
Given the current pull based design I do not think it is feasible to implement different strategies in a reasonable way with the current API.
Describe the solution you'd like
I would like a way for advanced users of the parquet crate to more carefully optimize the I/O patterns
Describe alternatives you've considered
Push Based Decoder
One potential solution (again, thanks to @crepererum and @XiangpengHao), is to make a "push based" decoder, similarly to the JSON decoder and the CSV decoder.
A (simplified) example of how the JSON Decoder works shows illustrates the difference:
let mut decoder = ReaderBuilder::new(schema).build_decoder()?;
let mut next = move || {
loop {
// load some more bytes from somewhere
let buf = read_input();
if buf.is_empty() {
break; // Input exhausted
}
let read = buf.len();
let decoded = decoder.decode(buf)?;
// If the decoder couldn't read the whole buffer, it made an output batch
if decoded != read {
break;
}
}
decoder.flush() // retrieve completed batch
};So the idea is that we could make a ParquetDecoder that is "push" based, and did not have an underlying IO reader, but instead would be given a stream of data to decode. We could then implement the current sync and async APIs using that underlying Decoder
The challenge is likely with the API design: unlike streams of JSON / CSV, the data that the parquet reader needs next will need is not easy to predict as it depends on the filters, the row groups, which columns are requested, etc.
Here is a high level straw man design:
// Create a decoder for decoding parquet data (note it does NOT have any IO / readers)
let mut decoder = ParquetDecoderBuilder::new();
// provide metadata, if known (avoid having to read it each time)
.with_file_metadata(metadata)
.with_row_groups(row_groups)
.with_projection(projection)
.build();
// In a loop, ask the decoder what it needs next, and provide it with the required data
while let Some(next_request) = decoder.next_request() {
// next_request will tell us what data ranges it needs (details TBD)
let data: Bytes = fetch_data(next_request);
// provide the data to the decoder
decoder.push_data(next_request, data);
// Now the decoder may be able to decode a batch
// maybe it will return one or more batches, or it will ask for more data
while let Some(batch) = decoder.next_batch() {
// process the batch
}
}Note that the above example is not async, but could be easily be used by an async API
With such a ParquetDecoder I think we could then add things like prefetching of row groups, etc, with new APIs on the decoder
// Create a decoder for decoding parquet data as above
let mut decoder: ParquetDecoderBuilder = ...;
// As the decoder up from what data it will need, start prefetching data if desired
while let Some(pre_request) = decoder.peek_next_requests() {
// note that this is a peek and if we call peek again in the
// future, we may get a different set of pre_requests (for example
// if the decoder has applied a row filter and ruled out
// some row groups or data pages)
start_prefetch(pre_request);
}
// push data to the decoder as before, but hopefully the reader
// will have already prefetched some of the dataAdditional context
I have now hit this while trying to write tests end to end for the parquet reader (also see #7971), and it was very annoying to have to duplicate the code for the sync and async readers.
- @tustvold 's suggestion Push-Based Parquet Reader #1605
- Here is another proposal for a low level API Low-Level Arrow Parquet Reader #5522
Activity
tustvold commentedon Jul 24, 2025
Broadly speaking I agree with this, in fact my original proposal was for such a reader #1605 however the realities of the current code and various aspects of the parquet format meant we ended up with the current situation as a pragmatic hack. I'd love to see something better in this space.
That being said, there are various aspects of the parquet format that make this rather difficult, and there's a fair amount of subtlety to how things like RecordReader work that rely on being able to see if there may be more pages, etc... Ultimately parquet was not designed for this sort of reader, and whilst stuff like the offset index has been bolted on, we can't mandate the existence of these.
Fortunately we have relatively good test coverage of these quirks, so provided any rework was able to reuse these, we should avoid regressions.
alamb commentedon Jul 24, 2025
Thank you -- I am sorry I should have mentioned your previous work in this space -- and the fact that it inspired this writeup.
Indeed -- what I think would be ideal is if both the ParquetRecordBatchReader (sync) and ParquetRecordBatchStream (async) were just "IO wrappers" around this
ParquetDecoderNow we just need to find someone with enough time to do the work -- lol 🎣
alamb commentedon Jul 24, 2025
FYI @etseidl I think this topic might also be interesting to you (it is somewhat similar to your work on the ParquetMetadataDecoder and how that returns information about what data is needed to decode the next chunk of the metadata)
alamb commentedon Jul 24, 2025
I also wonder if this decoder does the IO as well is part of why we are having so much trouble getting filter pushdown enabled in
alamb commentedon Jul 25, 2025
I wrote up my plan for parquet here: #8000
JigaoLuo commentedon Jul 28, 2025
This is really interesting (even for GPU tablescan). I’m wondering: how do other Parquet-based tablescans/readers typically handle the execution model —are they using a pull model, a push model, or something different?
alamb commentedon Jul 28, 2025
I do not know to be honest -- I am pretty sure the java reader is using a pull model
mapleFU commentedon Aug 4, 2025
Just curious that if a push-based decoder means there should be a throttle, which limits the memory for consumer using finally?
alamb commentedon Aug 4, 2025
I think a push decoder will eventually allow much more control over how much memory is used -- there will always be the need to provide page data to the decoder. Once the code for determining what pages are needed is detangled from the code to fetch and buffer them, I think it will be far more tractable to imagine doing more I/O operations and buffering less
But I don't envision that as part of the initial decoder
I hope to have the push decoder up for review later this week
mapleFU commentedon Aug 5, 2025
@alamb Just curious, previously I worked on a system like presto/impala, which has metrics for parquet decompress time/decode time/io time. Currently in arrow-rs, seems we don't have these wrappers, how can we get the time-spent here? I know flamegraph would help but is there some offical way to acquire time spending on these? And how do datafusion get these metrics?
alamb commentedon Aug 5, 2025
TLDR is there are no metrics reported by the arrow-rs parquet reader directly.
I also very much would like more fine grained metrics, I actually have a PR up to propose adding such an API here
Perhaps you can take a look at it and let me know if that is what you had in mind
DataFusion gets its metrics by observing the reader from the outside, so it can't do things like separate the decompress from decode times.
alamb commentedon Aug 7, 2025
Update here -- the push decoder is looking quite nice. Here is a preview of its API:
My next steps:
lidavidm commentedon Aug 7, 2025
The parquet-cpp "row group coalescing" optimizations may be helpful as some prior art. This is the original thread which has a Google doc outlining the kinds of optimizations we were trying to enable (only coalescing actually made it in to parquet-cpp): https://lists.apache.org/thread/cdfkm8oflm2zvd25yn4k6gh2o7pc9z88
They had to be hacked into the C++ reader rather roughly. I've always wanted a push-based decoder like the one you have here; the ranges API looks perfect.
6 remaining items
lidavidm commentedon Aug 9, 2025
I guess: you push bytes into the decoder (by polling its state), vs the decoder pulls bytes from a stream
arrow-cpp calls its equivalent IPC reader a "push-style stream decoder": https://github.com/apache/arrow/blob/0d72f7e8f33a79ced28f42017bf07910c85c6d74/cpp/src/arrow/ipc/reader.h#L384-L393
[-][DISCUSS] Decouple IO and CPU operations in the Parquet Reader (push decoder?)[/-][+]Decouple IO and CPU operations in the Parquet Reader (push decoder?)[/+][-]Decouple IO and CPU operations in the Parquet Reader (push decoder?)[/-][+]Decouple IO and CPU operations in the Parquet Reader (push decoder)[/+]alamb commentedon Aug 11, 2025
Yes, exactly, I think this is the intention.
The Json
Decoderin this crate https://docs.rs/arrow-json/56.0.0/arrow_json/reader/struct.Decoder.html also uses a similar terminology:alamb commentedon Aug 19, 2025
Update:
ParquetRecordBatchStreamin terms of the PushDecoder #8159 (comment)) . I am moving on to some more cleanups and then will try the same test with the sync readerBukhtawar commentedon Sep 8, 2025
Is io_uring also on the plans(similar to https://github.com/tokio-rs/io-uring), sorry I am still catching up on the discussion.
crepererum commentedon Sep 9, 2025
I think using
io_uringis a larger task. For file-system-based I/O, we would at least wire that intoobject_store. For network-based I/O, we would also need to wire it throughhyper,reqwestsand frieds. I don't see this happening any time soon.alamb commentedon Sep 9, 2025
In my mind the PaquetPushDecoder could also be used with io_uring (though someone would have to write the code to translate the io needs of the decoder into io_uring requests and manage the queue, push the data, etc)
The whole point is to remove all IO from the actual parquet decoder to enable this kind of integration
[Parquet] Add ParquetMetadataPushDecoder (#8080)
Implement Push Parquet Decoder (#7997)
Rewrite `ParquetRecordBatchStream` in terms of the PushDecoder (#8159)