Repository navigation
Support Morsel output for Parquet known to be non blocking #21598
Description
Is your feature request related to a problem or challenge?
- Follow-on from Introduce morsel-driven Parquet scan #20529
- And themorsel API work in Rewrite FileStream in terms of Morsel API #21342
In this review comment, @adriangb raised a good question about whether Morsel should return an iterator rather than a stream:
Is this another case where we could better encode the behavior constraints into the type system?
I tried it out with codex, and found that switching Morsel from Stream to Iterator was straightforward for FileStream but was very non trivial for Parquet and the ParquetPushDecoder, which can discover NeedsData only after some batches have already been produced from the current decode state.
This makes creating a synchronous morsel boundary difficult to express cleanly in ParquetOpenState, PushDecoderState, and ParquetStreamMorsel because there is no way to know with the current APIs much data will be produced prior to the next need for IO
Describe the solution you'd like
I think it would be useful if Parquet exposed one or both of:
- a way to ask the decoder whether more output can be produced synchronously before the next I/O boundary
- a cheap way to split or clone the reader / read context so the next planner I/O can be prepared independently of the currently active morsel iterator
Two likely enablers are:
- an API on
ParquetPushDecoderthat can tell the caller how much already-buffered output remains, or otherwise expose a clearer “this many rows/batches are ready before the nextNeedsData” boundary
That would let DataFusion produce a Morsel that is CPU-only, while keeping future I/O in PendingMorselPlanner rather than hiding async work behind a blocking iterator bridge.
Describe alternatives you've considered
- Keep
Morselas a stream for now. - Use
futures::executor::block_onorblock_on_streaminside Parquet morsels. This works mechanically but blocks executor threads and is not acceptable in my mind - Use a channel handoff where the morsel iterator returns batches until
NeedsData, then sends decoder state to a waiting planner future. This is conceptually closer to the desired model, but the prototype still had lifecycle/handoff complexity and hung in some Parquet tests.
Activity
[-]Support iterator-based Morsel output for Parquet without blocking[/-][+]Support Morsel output for Parquet known to be non blocking[/+]alamb commentedon Apr 16, 2026
Actually, I was poking around the API in the context of apache/arrow-rs#9697 and I actually think we have the API we needed for this:
https://docs.rs/parquet/58.0.0/parquet/arrow/push_decoder/struct.ParquetPushDecoder.html#method.try_next_reader
So the idea is that a morsel planner could call try_next_reader and retrn that as the morsel 🤔
alamb commentedon Apr 20, 2026
FYI I started messing around with this locally -- hopefully I will have something to share soon
zhuqi-lucas commentedon Apr 22, 2026
Thanks @alamb! The
try_next_reader/peek_next_row_group/skip_next_row_groupAPIs in arrow-rs are exactly what we need for dynamic RG pruning.In #21580 we've built the statistics-driven TopK optimization chain (file reorder + RG reorder + stats init + cumulative prune) which gives 17-60x improvement on sorted data by skipping RGs at planning time. But for the WHERE case and overlapping RGs, we need runtime dynamic pruning — checking the TopK threshold against the next RG's statistics between reads.
The morsel split via
try_next_readerwould let us do this naturally: when a morsel planner callstry_next_reader, it can checkpeek_next_row_group()statistics against the currentDynamicFilterPhysicalExprthreshold andskip_next_row_group()if prunable.This is tracked in #21399. Happy to help implement the dynamic pruning logic on top of the morsel split once the API is wired up.
Dandandan commentedon Apr 22, 2026
Cool @alamb .
I've a PoC / bench to show the impact of going from file -> row group morsel using the similar approach earlier of using 2 queues (files / morsels).
#21766
Using
try_next_readerlooks great, I think it might simplify the implementation a bitI didn't try to delay morsel splitting to the query tail, just split always into row groups.