Repository navigation
Push-Based Parquet Reader #1605
Description
TLDR
This proposal will allow reading data from parquet files in a "streaming" fashion that reduces the IO required, the resources to locally buffer parquet files, and potentially reduces the latency of first data. It will help the usecase of reading parquet files from remote object stores such as AWS S3.
Problem
SerializedFileReader is currently created with a ChunkReader which looks like
pub trait ChunkReader: Length + Send + Sync {
type T: Read + Send;
/// get a serialy readeable slice of the current reader
/// This should fail if the slice exceeds the current bounds
fn get_read(&self, start: u64, length: usize) -> Result<Self::T>;
}
The process for reading a file is then
SerializedFileReader::newwill callfooter::parse_metadataparse_metadatawill- Call
ChunkReader::get_readwith the final 64 KB byte range, and read this to a buffer - Determine the footer length
- Potentially call
ChunkReader::get_readto read the remainder of the footer, and read this to a buffer
- Call
SerializedFileReader::get_row_iterwill return aRowIterwhich for each row group- Call
SerializedRowGroupReader::newwhich will- Call
ChunkReader::get_readwith the byte range of each column chunk
- Call
There are two major options to apply this to files in object storage
- Fetch the entire file to local disk or memory and pass it to
SerializedFileReader - Convert
ChunkReader::get_readto a range request to object storage
The first option is problematic as it cannot use pruning logic to reduce the amount of data fetched from object storage.
The second option runs into two problems:
- The interface is not async and blocking a thread on network IO is not ideal
- Lots of small range requests per file adding cost and latency
Proposal
I would like to decouple the parquet reader entirely from IO concerns, allowing downstreams complete freedom to decide how they want to handle this. This will allow the reader to support a wide variety of potentially data access patterns:
- Sync/Async Disk IO
- Sync/Async Network IO
- In-memory/mmapped parquet files
- Interleaved row group decode with fetching the next row group
Footer Decode
Introduce functions to assist parsing the parquet metadata
/// Parses the 8-bytes parquet footer and returns the length of the metadata section
pub fn parse_footer(footer: [u8; 8]) -> Result<usize> {}
/// Parse metadata payload
pub fn parse_metadata(metadata: &[u8]) -> Result<ParquetMetaData> {}
This will allow callers to obtain ParquetMetaData regardless of how they choose to fetch the corresponding bytes
ScanBuilder / Scan
Next introduce a ScanBuilder and accompanying Scan.
/// Build a [`Scan`]
///
/// Eventually this will support predicate pushdown (#1191)
pub struct ScanBuilder {}
impl ScanBuilder {
pub fn new(metadata: Arc<ParquetMetaData>) -> Self {}
pub fn with_projection(self, projection: Vec<usize>) -> Self {}
pub fn with_row_groups(self, groups: Vec<usize>) -> Self {}
pub fn with_range(self, range: Range<usize>) -> Self {}
pub fn build(self) -> Scan {}
}
/// Identifies a portion of a file to read
pub struct Scan {}
impl Scan {
/// Returns a list of byte ranges needed
pub fn ranges(&self) -> &[Range<usize>] {}
/// Perform the scan returning a [`ParquetRecordBatchReader`]
pub fn execute<R: ChunkReader>(self, reader: R) -> Result<ParquetRecordBatchReader> {}
}
Where ParquetRecordBatchReader is the same type returned by the current ParquetFileArrowReader::get_record_reader, and is just an Iterator<Item=ArrowResult<RecordBatch>> with a Schema.
This design will only support the arrow use-case, but I couldn't see an easy way to add this at a lower level without introducing API inconsistencies when not scanning the entire file
Deprecate Async
Once implemented, I would propose deprecating and then removing the async API added by #1154
Alternatives Considered
#1154 added an async reader that uses the AsyncRead and AsyncSeek traits to read individual column chunks into memory from an async source. This is the approach taken by arrow2, with its range_reader abstraction. This was not found to perform particularly well (#1473).
#1473 proposed an async reader with prefetch functionality, and was also suggested by @alamb in apache/datafusion#2205 (comment). This is similar to the new FSDataInputStream vectored IO API in the Hadoop ecosystem. This was implemented in #1509 and found to perform better, but still represented a non-trivial performance regression on local files.
Additional Context
The motivating discussion for this issue can be found apache/datafusion#2205
@mateuszkj clearly documented the limitations of the current API datafusion-contrib/datafusion-objectstore-s3#53
Activity
alamb commentedon Apr 24, 2022
All in all I like this proposal. Thank you for writing it down. I know it is not based on @jorgecarleitao https://github.com/jorgecarleitao/parquet2 but it is not a dissimilar API where the metadata reading / management is handled outside the main parquet decoding logic -- see the example. I see this similarity as a good sign. 👍
I think it is important to sketch out what the interface for existing users of the
ParquetRecordBatchReaderwould be. Not just for helping migration, but to ensure that all use cases are satisfied (I am happy to help with this).Maybe we can provide functions like the following for basic use to both ease migration and to demonstrate how to use this API:
What about offering an function from
Scanthat sends backSerializedRowGroupReader?https://docs.rs/parquet/12.0.0/parquet/file/serialized_reader/struct.SerializedRowGroupReader.html
alamb commentedon Apr 24, 2022
Is the idea that the calling code would create a
Scanfor each row group and the execute thoseScans in whatever order was desired?tustvold commentedon Apr 24, 2022
Adding some free functions to assist migration makes sense to me. It should be pretty much a drop-in replacement 👍
I am somewhat apprehensive about providing an async version, as the whole intent is to let users handle what level of buffering/pre-fetch makes sense for their use case, but I guess with sufficient disclaimers...
As for SerializedRowGroupReader the challenge is the RowGroupReader trait exposes APIs for column projection, reading individual pages, etc... which then gets rather confusing if you have projection or predicate pushdown concepts on the Scan. I'd rather a clean break, than trying to shoehorn an existing API
W.r.t to users constructing per-row group, entirely up to their use-case, if they want to they can, if they don't want to, they don't need to. It was more me trying to demonstrate the flexibility afforded, than prescribing anything
alamb commentedon Apr 24, 2022
👍 makes sense
Yeah I am thinking about the "I want to try out reading parquet from s3 just to see if it works" rather than "I am set on using parquet library and I am now trying to squeeze the most performance"
Right -- I agree -- I was just confirming my understanding of what would be possible with the API
tustvold commentedon Oct 6, 2022
The desired IO decoupling has been achieved through AsyncFileReader and its integration with ParquetRecordBatchStream
alamb commentedon Jul 30, 2025
BTW we are discussing (and I am prototyping) a proper push decoder to fully decouple things.
It is getting close. See this for more details