[Parquet] PushDecoder: Add a peek API to support pre-fetching

[alamb]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

• part of [Epic] Parquet Reader Improvement Plan / Proposal - July 2025 #8000

Unlike streams of JSON / CSV, the data that the parquet reader needs next i is not easy to predict as it depends on the filters, the row groups, which columns are requested, etc.

Now that we have the initial PushDecoder in this PR

• Implement Push Parquet Decoder #7997

We will be in the position to add an API for the decoder to communicate what data will be needed next

Describe the solution you'd like
I would like an API that allows users of the Parquet decoder to have more fine grained control over peeking

Describe alternatives you've considered

Here is an idea from @adriangb on https://github.com/apache/arrow-rs/pull/7997/files#r2444922393

a method along the lines of try_peek()? It'd be cool if it returned some structure that allowed fine grained control of the peeking:

let max_ranges = 32;
let max_bytes = 1024 * 1024 * 32;
let mut current_bytes = 0;
let mut ranges = Vec::new();
let mut peek = decoder.peek()
loop {
    match peek.next() {
        PeekResult::Range(range) => {
            ranges.push(range);
            current_bytes += range.end - range.start;
            if ranges.len() > max_ranges { break }
            if current_bytes > max_bytes { break }
        PeekResult::End { break }
    }
}

Here is another potential API from the original ticket:

// 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 data
Additional context

[friendlymatthew]

Hi hi, I think the push based decoder looks super cool. I had a couple of questions:

1. Peek behavior
   I'm trying to understand when peek() should return ranges. Should it only return ranges when the decoder is in WaitingOnFilterData or WaitingOnData states? For other states (like Start, Filters, or StartData), it would return empty because determining needed ranges would require running the state machine

This form of behavior would look something like:

loop {
    // Peek returns ranges only when decoder is blocked waiting for data
    let ranges = decoder.peek().take(32).collect::<Vec<_>>();

    if !ranges.is_empty() {
        fetch_and_push(ranges);
    }

    match decoder.try_decode() {
        DecodeResult::NeedsData(_) => { /* already handled above */ }
        DecodeResult::Data(batch) => process(batch),
        DecodeResult::Finished => break,
    }
}

Or should peek() try to look ahead and predict future ranges (which would require simulating state transitions)

2. API simplification

PeekResult has 2 variants (Range(Range<u64>) and End), which is basically reinventing Option<Range<u64>>. I wonder if we should just simplify peek to:

fn peek(&self) -> impl Iterator<Item = Range<u64>>

[adriangb]

Should peek() try to look ahead and predict future ranges (which would require simulating state transitions)

I think that's the idea / is necessary. E.g. it will "simulate" filters filtering nothing out, the worst case scenario. If you ask it for a little bit of data it's probably accurate. If you ask it for a lot it might be quite an over-estimate.

[alamb]

Thank you @friendlymatthew and @adriangb

What I would suggest for this ticket is starting out by writing a test / usecase and then figuring out what API you need.

For example try to write a Parquet reader that

1. takes an ObjectStore instance as input
2. tries to prefetch pages for the next row group as it is decoding the current row group.

I am not sure exactly how this would work -- some ideas:

1. You could implement a peek API and start a separate task for prefetch
2. Maybe we could have a function like PushDecoder::take_next_row_group() that would return a new PushDecoder for evaluating only a single row group which could be processed in parallel?

The more I write this the more I like the PushDecoder::take_next_row_group idea...

Something like

let decoder = ParquetPushDecoder::new(...);
// start decoding row groups in parallel:
let rg1_decoder = decoder.take_next_row_group()
// now fetch / decode data from rg1 in a thread
task::run(... rg1_decocder ...);
let rg2_decoder  =decoder.take_next_row_group();
// read data from rg2 ...