alamb commented on code in PR #23492:
URL: https://github.com/apache/datafusion/pull/23492#discussion_r3724117458
##########
datafusion/common/src/config.rs:
##########
@@ -1133,6 +1133,13 @@ config_namespace! {
/// the hint, two reads will still be performed.
pub metadata_size_hint: Option<usize>, default = Some(512 * 1024)
+ /// (reading) If specified, the parquet reader will prefetch data for
+ /// subsequent row groups when the projected column chunks fit within
+ /// this many bytes. The required ranges for the current row group are
+ /// always read, even when they exceed this value. None disables data
+ /// prefetching.
+ pub prefetch_size: Option<usize>, default = None
Review Comment:
I agree this is an obvious and common strategy for reading from object store
and we should add it
One thing I am wondering is if we can make this IO strategy more generic
(for example, I can imagine a user wanting to support racing reads like the
craziness described in
https://github.com/apache/datafusion/pull/23492#issuecomment-5197454577
Where I am heading is "can we add an API / trait" that lets user customize
the I/O behavior more?
For example, maybe the trait kind of like `madvise` that could have methods
like
```rust
/// The reader will need the bytes in `range` at some point in the future
/// it will request the bytes in the same order as called by this function
/// though it may not read all of them
fn advise_bytes_needed(&self, range: Range<usize>)
```
And then we could include a default implementation that implemented memory
limited prefetch 🤔 but users could provide their own implementations that did
other things (like racing reads, etc)
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -423,6 +446,68 @@ impl PushDecoderStreamState {
}
}
+/// Append projected column chunks from subsequent row groups to a decoder
+/// request while staying within `prefetch_size`.
+///
+/// The first entry in `rg_plan` is the row group responsible for `ranges`.
+/// Complete projected ranges for later row groups are added in scan order so
+/// the push decoder can stage them for future calls to `try_next_reader`.
+fn prefetch_row_group_ranges(
+ mut ranges: Vec<Range<u64>>,
+ buffered_bytes: u64,
+ prefetch_size: Option<usize>,
+ rg_plan: &VecDeque<RgPlanEntry>,
+ projection: &ProjectionMask,
+ metadata: &ParquetMetaData,
+ prefetched_row_groups: &mut HashSet<usize>,
+) -> Vec<Range<u64>> {
+ let Some(prefetch_size) = prefetch_size.filter(|size| *size > 0) else {
+ return ranges;
+ };
+
+ let requested_bytes = ranges
+ .iter()
+ .map(|range| range.end - range.start)
+ .sum::<u64>();
+ let mut staged_bytes = buffered_bytes.saturating_add(requested_bytes);
+ let budget = prefetch_size as u64;
+ if staged_bytes >= budget {
+ return ranges;
+ }
+
+ for entry in rg_plan.iter().skip(1) {
+ if prefetched_row_groups.contains(&entry.rg_index) {
+ continue;
+ }
+
+ let row_group = metadata.row_group(entry.rg_index);
Review Comment:
this would be a nice API to add upstream in the parquet crate probably
--
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]
