Morsel-Driven IO: How DataFusion Reads Files in Pieces So One Slow File Cannot Stall the Scan

TL;DR: A file scan is usually a fixed assignment: partition 0 gets files 0..k, partition 1 gets files k..2k, and if the files have wildly different sizes, one partition takes ten times as long as the others and the query waits for it. DataFusion’s answer is a shared work queue — whoever finishes first takes the next file — combined with morsels, chunks of a file that are ready to decode without further I/O. The result is a scan that balances itself.

The classic parallel file scan has a load-balancing bug. You have 1,000 files and 16 workers. You split the files into 16 contiguous groups, one per worker. Then it turns out 900 of the files are 1 MB and 100 are 100 MB. Worker 3 got three of the 100 MB files and takes 300 MB of I/O while worker 7 got 62 MB of small files and finishes in a tenth of the time and then idles.

The total work is the same and the wall clock is 5× worse than it needs to be, because the query cannot finish until the slowest worker does.

The fix: a shared queue

The obvious fix is a work queue:

rust
/// Shared source of work for sibling `FileStream`s
///
/// The queue is created once per execution and shared by all reorderable
/// sibling streams for that execution. Whichever stream becomes idle first may
/// take the next unopened file from the front of the queue.
///
/// It uses a [`Mutex`] internally to provide thread-safe access
/// to the shared file queue.
#[derive(Debug, Clone)]
pub(crate) struct SharedWorkSource {
    inner: Arc<SharedWorkSourceInner>,
}

#[derive(Debug, Default)]
pub(super) struct SharedWorkSourceInner {
    files: Mutex<VecDeque<PartitionedFile>>,
}

A Mutex<VecDeque<PartitionedFile>> behind an Arc. Each scan task pops from the front. Whoever finishes first gets the next file. That is work stealing at file granularity, and it is exactly what you want.

But there is a catch, and the code models it explicitly:

rust
/// Source of work for `ScanState`.
///
/// Streams that may share work across siblings use [`WorkSource::Shared`],
/// while streams that can not share work (e.g. because they must preserve file
/// order) use [`WorkSource::Local`].
#[derive(Debug, Clone)]
pub(super) enum WorkSource {
    /// Files this stream will plan locally without sharing them.
    Local(VecDeque<PartitionedFile>),
    /// Files shared with sibling streams.
    Shared(SharedWorkSource),
}

A stream that must preserve file order — because the query has an ORDER BY on a partition column, or the file format requires it — cannot take files out of order, so it gets a private queue. The Local variant exists for that.

And the accounting differs too:

rust
/// Return how many queued files should be counted as already processed
/// when this stream stops early after hitting a global limit.
pub(super) fn skipped_on_limit(&self) -> usize {
    match self {
        Self::Local(files) => files.len(),
        Self::Shared(_) => 0,
    }
}

With a local queue, the unprocessed files are yours and count as skipped when you hit a limit. With a shared queue, they are not yours — another stream will get them — so the count is zero. Getting this wrong would make the “files skipped by LIMIT” metric lie.

Morsels: the second level

The shared queue balances at file granularity. But what if there is one enormous file? Then the queue does not help at all: one worker gets the 1 GB file and the other 15 idle.

That is where morsels come in. The idea is borrowed from a specific paper, and the source cites it:

Morsel Driven IO is a technique for parallelizing the reading of large files by dividing them into smaller “morsels” that are processed independently.
It is inspired by the paper [Morsel-Driven Parallelism: A NUMA-Aware Query Evaluation Framework for the Many-Core Age].

A morsel is a chunk of a file that is ready to decode — all its bytes are already in memory:

rust
/// A Morsel of work ready to resolve to a stream of [`RecordBatch`]es.
///
/// This represents a single morsel of work that is ready to be processed. It
/// has all data necessary (does not need any I/O) and is ready to be turned
/// into a stream of [`RecordBatch`]es for processing by the execution engine.
pub trait Morsel: Send + Debug {
    /// Consume this morsel and produce a stream of [`RecordBatch`]es for processing.
    ///
    /// Note: This may do CPU work to decode already-loaded data, but should not
    /// do any I/O work such as reading from the file.
    fn into_stream(self: Box<Self>) -> BoxStream<'static, Result<RecordBatch>>;
}

“does not need any I/O” is the whole contract. A morsel is data in memory, waiting for a CPU.

Separating planning from I/O

The other idea in the morsel design is that planning a file and reading from it are different kinds of work, and they should not be confused.

rust
/// A Morsel Planner is responsible for creating morsels for a given scan.
///
/// The [`MorselPlanner`] is the unit of I/O. There is only ever a single I/O
/// outstanding for a specific planner. DataFusion may run
/// multiple planners in parallel, which corresponds to multiple parallel
/// I/O requests.
///
/// It is not a Rust `Stream` so that it can explicitly separate CPU bound
/// work from I/O work.
pub trait MorselPlanner: Send + Debug {
    /// Attempt to plan morsels. This may involve CPU work, such as parsing
    /// parquet metadata and evaluating pruning predicates.
    ///
    /// It should NOT do any I/O work, such as reading from the file. If I/O is
    /// required, the returned [`MorselPlan`] should contain a pending planner
    /// future that the caller polls to drive the I/O work to completion.
    ///
    /// Note this function is **not async** to make it explicitly clear that if
    /// I/O is required, it should be done in the returned `io_future`.
    fn plan(self: Box<Self>) -> Result<Option<MorselPlan>>;
}

Three properties worth highlighting:

plan is not async. That is a deliberate API choice, and the doc says why: “Note this function is not async to make it explicitly clear that if I/O is required, it should be done in the returned io_future.” Making it async would invite implementations to just .await inside it, and the whole point is that planning is CPU-only.

The I/O is a separate future. A MorselPlan can carry a pending I/O future:

rust
pub struct MorselPlan {
    /// Morsels ready for CPU work
    morsels: Vec<Box<dyn Morsel>>,
    /// Planners that are ready for CPU work.
    ready_planners: Vec<Box<dyn MorselPlanner>>,
    /// A future with planner I/O that resolves to a CPU ready planner.
    pending_planner: Option<PendingMorselPlanner>,
}

So the state machine is: plan (CPU) → maybe get an I/O future → poll it → plan again (CPU) → get morsels → decode (CPU).

The design mirrors the push decoder. The doc says so: “The design is similar to ParquetPushDecoder: when plan is called, it should do CPU work to produce the next morsels or discover the next I/O phase.”

The scan state machine

The ScanState in datafusion/datasource/src/file_stream/scan_state.rs is the machinery that ties it together, and its doc comment is a nice little pipeline diagram:

text
unopened files, CPU-ready planners, pending planner I/O, ready morsels,
...
morselizer.plan_file(file)
...
ready_planners ---> plan() ---> ready_morsels ---> into_stream() ---> reader ---> RecordBatches

So: an unopened file is turned into a planner by the Morselizer (morselizer.plan_file(part_file)), a planner produces morsels and child planners, a morsel becomes a reader, and the reader produces batches.

And the ordering of the state machine matters — it prioritises having something to decode over starting new work:

rust
// No active reader, but a morsel is ready to become the reader.
if let Some(morsel) = self.ready_morsels.pop_front() {
    ...
    self.reader = Some(morsel.into_stream());
}

// No reader or morsel, so try to produce more work via CPU planning.

Ready morsels first, then planning. That keeps the CPU busy with decoding before it goes off to do more planning.

The ordering contract

Morsels have to come out in a defined order, or the scan’s output order would be undefined:

Logical Ordering
For plans where the output order of rows is maintained, the output order of a [MorselPlanner] is logically defined as follows:
All morsels that are directly produced
Recursively, all morsels produced by the returned planners

So a planner’s output is: its own morsels first, then its children’s, recursively. That is a depth-first order, and it is what lets a planner’s children be planned lazily and still produce a correct order.

The catch

It is experimental. The module says so: “NOTE: As of DataFusion 54.0.0, these are experimental APIs that may change substantially.”
Not every scan uses it. There is a config flag, enable_file_stream_work_stealing, default true, and a test that verifies the disabled path keeps each partition’s files local:
rust
```
/// Verifies that disabling `enable_file_stream_work_stealing` keeps each
```
So the shared queue is opt-out, and the opt-out has a defined behaviour.
The shared queue breaks order guarantees. Any plan that depends on the order of files must use the Local work source. This is a real constraint on which optimisations can be combined.
Morsels require the format to support them. A file format that cannot produce independently-decodable chunks cannot be morselised. The Morselizer trait is the extension point, and the Parquet implementation is the one that exists.
The two levels interact. A shared queue over files and morsels within a file is a two-dimensional balancing problem, and the code handles it by keeping the file-level queue shared and the morsel-level work local to whoever opened the file.

Why it matters

Load imbalance is the difference between a query that scales and one that does not. It is invisible in benchmarks with uniform files and brutal in production, where files come in whatever sizes the upstream system produced.

The shared queue is a five-line change with a large effect, and the morsel machinery is a more substantial change with an even larger effect for the single-huge-file case. Together they turn “the scan is as slow as its slowest worker” into “the scan is as slow as the total work divided by the number of workers”.

And the idea is not new — it comes from a 2010 paper about NUMA-aware query evaluation. But seeing it inside a modern vectorised engine, in an API that explicitly separates CPU planning from I/O, is a nice illustration of how much of database systems engineering is re-deriving the same handful of good ideas.