apache / apache/datafusion

Support Morsel output for Parquet known to be non blocking

Open
#21,598 4 comments 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

## Is your feature request related to a problem or challenge?

- Follow-on from https://github.com/apache/datafusion/issues/20529
- And themorsel API work in https://github.com/apache/datafusion/pull/21342

In [this review comment](https://github.com/apache/datafusion/pull/21342#discussion_r3070184104), @adriangb raised a good question about whether [`Morsel`](https://github.com/apache/datafusion/blob/04dbbbf6694a4b162f76aee0091fdc3a47d2f9f0/datafusion/datasource/src/morsel/mod.rs#L52) 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`](https://docs.rs/parquet/latest/parquet/arrow/push_decoder/struct.ParquetPushDecoder.html), 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`](https://github.com/apache/datafusion/blob/04dbbbf6694a4b162f76aee0091fdc3a47d2f9f0/datafusion/datasource-parquet/src/opener.rs#L202), [`PushDecoderState`](https://github.com/apache/datafusion/blob/04dbbbf6694a4b162f76aee0091fdc3a47d2f9f0/datafusion/datasource-parquet/src/opener.rs#L411), and [`ParquetStreamMorsel`](https://github.com/apache/datafusion/blob/04dbbbf6694a4b162f76aee0091fdc3a47d2f9f0/datafusion/datasource-parquet/src/opener.rs#L507) 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 [`ParquetPushDecoder`](https://docs.rs/parquet/latest/parquet/arrow/push_decoder/struct.ParquetPushDecoder.html) that can tell the caller how much already-buffered output remains, or otherwise expose a clearer “this many rows/batches are ready before the next `NeedsData`” boundary

That would let DataFusion produce a [`Morsel`](https://github.com/apache/datafusion/blob/04dbbbf6694a4b162f76aee0091fdc3a47d2f9f0/datafusion/datasource/src/morsel/mod.rs#L52) that is CPU-only, while keeping future I/O in [`PendingMorselPlanner`](https://github.com/apache/datafusion/blob/04dbbbf6694a4b162f76aee0091fdc3a47d2f9f0/datafusion/datasource/src/morsel/mod.rs#L188) rather than hiding async work behind a blocking iterator bridge.

## Describe alternatives you've considered

- Keep [`Morsel`](https://github.com/apache/datafusion/blob/04dbbbf6694a4b162f76aee0091fdc3a47d2f9f0/datafusion/datasource/src/morsel/mod.rs#L52) as a stream for now.
- Use `futures::executor::block_on` or `block_on_stream` inside 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.

## Additional context

Contributor guide

Open the contributing guide

Research direction

Read datafusion/datasource/src/morsel/mod.rs, especially Morsel and PendingMorselPlanner, alongside datafusion/datasource-parquet/src/opener.rs at ParquetOpenState, PushDecoderState, and ParquetStreamMorsel. Start by examining ParquetPushDecoder's current output and NeedsData behavior; done means a clear decoder boundary lets DataFusion produce CPU-only Morsels while future I/O remains with the planner.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Needs clarification
Newbie friendliness
30/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.