Polars' New Streaming Engine Works in Morsels

TL;DR: A morsel is a slice of rows plus a monotonically increasing sequence number, and it is the only thing that moves between operators in the new streaming engine. A source node decodes a file, cuts the decoded frame into morsels of about POLARS_IDEAL_MORSEL_SIZE rows, and pushes them into capacity-one channels. A distributor then hands each morsel to whichever pipeline is free, and a “consume token” attached to the morsel turns the channel’s handoff into real backpressure: the producer cannot run more than one morsel ahead.

I went looking for the thing that makes Polars’ new streaming engine different from the old one, expecting to find a scheduler or a task runtime. What I found was a much smaller idea: the engine has no notion of a “batch” that a thread loops over. It has a morsel — a fixed number of rows — and everything else in the engine is about getting morsels from place to place without letting memory run away.

That sounds like an implementation detail. It is not. The choice of morsel-as-the-unit-of-work is what makes backpressure, ordering, and parallelism fall out naturally instead of needing three separate mechanisms.

What a morsel actually is

The whole definition is 15 lines in crates/polars-stream/src/morsel.rs:95:

#[derive(Debug)]
pub struct Morsel {
    /// The data contained in this morsel.
    sf: SpillFrame,

    /// The sequence number of this morsel. May only stay equal or increase
    /// within a pipeline.
    seq: MorselSeq,

    /// A token that indicates which source this morsel originates from.
    source_token: SourceToken,

    /// Used to notify someone when this morsel is consumed, to provide backpressure.
    consume_token: Option<WaitToken>,
}

Three things are worth noticing.

First, the payload is a SpillFrame, not a DataFrame. A SpillFrame is a frame that can be pushed to disk if the memory manager decides it has to be, and pulled back when it is needed again. The streaming engine does not assume it has enough RAM to hold what it is processing.

Second, seq is a MorselSeq, which is deliberately a logical sequence number with room to spare (crates/polars-stream/src/morsel.rs:35):

#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug, Default)]
pub struct MorselSeq(u64);

impl MorselSeq {
    // TODO: use least significant bit to indicate 'last morsel with this
    // sequence number'.
    pub fn new(seq: u64) -> Self {
        Self(seq.checked_mul(2).unwrap())
    }

    // The morsel sequence id which comes after this morsel.
    pub fn successor(self) -> Self {
        // We increment by two because in the future we want to use the least
        // significant bit to indicate the final morsel with that sequence id.
        Self(self.0.checked_add(2).unwrap())
    }

The sequence numbers are multiplied by two on the way in. The low bit is reserved for a future “this is the last morsel with this sequence id” flag. The cost of that reservation is one bit and a shift; the benefit is that the ordering machinery does not have to be rewritten when the flag lands.

Third, consume_token is the backpressure handle. It is a WaitToken from a WaitGroup, and dropping it wakes whoever is waiting on the group. We will come back to it.

Sizing a morsel

There is exactly one knob, and it is a global config value (crates/polars-stream/src/morsel.rs:13):

pub fn get_ideal_morsel_size() -> usize {
    polars_config::config().ideal_morsel_size() as usize
}

The default is 100,000 rows (crates/polars-config/src/lib.rs:36):

const IDEAL_MORSEL_SIZE: &str = "POLARS_IDEAL_MORSEL_SIZE";
const DEFAULT_IDEAL_MORSEL_SIZE: u64 = 100_000;

100k rows is a compromise. Too small and the per-morsel overhead (channel handoff, sequence bookkeeping, a gather per partition) dominates. Too large and the pipeline stops being incremental — a single morsel sitting in a channel is already a lot of memory replicated across all workers.

Note the word ideal. The read path does not hit it exactly. What the read path does is a size calculation, and the write path does a different one; both live in crates/polars-utils/src/calc_morsel_split.rs, and the sink-side one is where calc_n_parts earns its keep (crates/polars-utils/src/calc_morsel_split.rs:69):

/// Number of parts to split `size` to minimize the average part size difference to `target_part_size`.
pub fn calc_n_parts(size: u64, target_part_size: NonZeroU64) -> u64 {
    if size <= target_part_size.get() {
        return if size == 0 { 0 } else { 1 };
    }

    let n_parts = size / target_part_size;

    (n_parts..=n_parts.saturating_add(1))
        .min_by_key(|n_parts| (size / *n_parts).abs_diff(target_part_size.get()))
        .unwrap()
}

It considers n and n+1 parts and picks whichever gives a part size closest to the target. This is the trick that lets the sink side decide how many rows go into a Parquet row group; on the source side the equivalent logic is simpler and lives with the readers.

How a source splits a file into morsels

The Parquet reader is the most interesting source because it does real work — prefetching byte ranges, decoding row groups — before it has anything to slice. The pipeline has three stages (crates/polars-stream/src/nodes/io_sources/parquet/init.rs:110):

   prefetch task                decode tasks              distribute task
   (tokio)                      (computational)            (computational)
  ┌──────────────┐            ┌──────────────┐           ┌──────────────┐
  │ byte ranges  │  channel   │ row group -> │  channel  │ frame ->     │
  │ for row grps │ ─────────► │ DataFrame    │ ────────► │ morsels      │
  └──────────────┘  (bounded) └──────────────┘  (bounded)└──────────────┘

The prefetch stage is bounded by a byte budget, the decode stage by the number of pipelines, and the distribute stage is a pure dispatch loop that does no CPU or I/O work of its own. The distribute stage is where morsels are born, and it is remarkably short (crates/polars-stream/src/nodes/io_sources/parquet/init.rs:403):

            while let Some((df, permits)) = next.take() {
                // Try to decode the next non-empty morsel first, so we know
                // whether the df is the last morsel.

                // Important: Drop this before awaiting the next one, or could
                // deadlock if the permit limit is 1.
                drop(permits);

                loop {
                    let Some((decode_fut, permit)) = decode_recv.recv().await else {
                        break;
                    };
                    let next_df = decode_fut.await?;
                    if next_df.height() == 0 {
                        continue;
                    }
                    next = Some((next_df, permit));
                    break;
                }

                for df in split_to_morsels(
                    &df,
                    ideal_morsel_size,
                    next.is_none(),
                    last_morsel_pipelines,
                ) {
                    if morsel_sender
                        .send_morsel(Morsel::new_unregistered(
                            df,
                            morsel_seq,
                            source_token.clone(),
                        ))
                        .await
                        .is_err()
                    {
                        return Ok(());
                    }
                    morsel_seq = morsel_seq.successor();
                }
            }

The interesting decision is that the distributor decodes one row group ahead so it knows whether the frame it is holding is the last one. If it is not the last, the frame can be split freely. If it is the last, the split has to be spread across the pipelines so that the final morsels actually run in parallel instead of leaving all but one worker idle.

The splitting itself is split_to_morsels (crates/polars-stream/src/nodes/io_sources/shared/morsels.rs:9):

pub(crate) fn split_to_morsels(
    df: &DataFrame,
    ideal_morsel_size: usize,
    last_morsel: bool,
    last_morsel_pipelines: usize,
) -> impl Iterator<Item = DataFrame> + '_ {
    let mut n_morsels = if df.height() > 3 * ideal_morsel_size / 2 {
        // num_rows > (1.5 * ideal_morsel_size)
        (df.height() / ideal_morsel_size).max(2)
    } else {
        1
    };

    if last_morsel {
        n_morsels = n_morsels.max(last_morsel_pipelines);
    }

    let rows_per_morsel = df.height().div_ceil(n_morsels).max(1);

    (0..i64::try_from(df.height()).unwrap())
        .step_by(rows_per_morsel)
        .map(move |offset| df.slice(offset, rows_per_morsel))
        .filter(|df| df.height() > 0)
}

Two thresholds are hiding in the first block. Below 1.5× the ideal size the frame is left whole — splitting a 120k-row frame into two 60k morsels buys nothing and costs a handoff. Above it, the frame is split into roughly ideal-sized pieces. And the “last morsel” case bumps the count up to at least last_morsel_pipelines, which is how the tail of a scan keeps all the workers busy.

last_morsel_pipelines is computed by the multi-scan layer, not by the reader, so that the tail of several files does not each spawn a full set of morsels (crates/polars-stream/src/nodes/io_sources/multi_scan/pipeline/initialization.rs:490):

    // Share the last-morsel split budget across files in the scan: divide it by the number
    // of files that can be in flight at once, so the total morsel count at end-of-file
    // boundaries stays bounded by `num_pipelines`.
    let last_morsel_pipelines =
        num_pipelines.div_ceil(n_effective_sources.min(max_concurrent_scans).max(1));

That is a nice example of a policy decision living above the operator that implements it.

The in-memory source is the degenerate case where the “file” is already a DataFrame in memory (crates/polars-stream/src/nodes/in_memory_source.rs:56):

        if self.morsel_size == 0 {
            let len = self.source.as_ref().unwrap().height();
            let ideal_morsel_count = (len / get_ideal_morsel_size()).max(1);
            let morsel_count = ideal_morsel_count.next_multiple_of(state.num_pipelines);
            self.morsel_size = len.div_ceil(morsel_count).max(1);
            self.seq = AtomicU64::new(0);
        }

Here the morsel count is rounded up to a multiple of the pipeline count, so each of the N workers gets the same number of morsels and no worker is left holding one extra at the end.

The channel: capacity one, by design

Morsels travel over a PortSender/PortReceiver pair (crates/polars-stream/src/pipe.rs:29):

impl PortSender {
    #[inline]
    pub async fn send(&mut self, morsel: Morsel) -> Result<(), Morsel> {
        let rows = morsel.height() as u64;
        self.0.send(morsel).await?;
        if let Some(metrics) = self.0.shared() {
            metrics.morsels_sent.fetch_add(1);
            metrics.rows_sent.fetch_add(rows);
            metrics.largest_morsel_sent.fetch_max(rows);
        }
        Ok(())
    }
}

The underlying primitive is deliberately the most minimal thing that works — a single-producer, single-consumer, capacity-one channel (crates/polars-async/src/primitives/connector.rs:16):

/// Single-producer, single-consumer capacity-one channel.
pub fn connector<T>() -> (Sender<T>, Receiver<T>) {

Capacity one means a send blocks until the receiver has taken the previous value. This is the entire backpressure mechanism for a one-to-one pipe. There is no separate credit system, no queue of pending work, no risk of the producer racing ahead and filling memory with morsels nobody has asked for.

The cost is that a one-to-one pipe cannot overlap producer and consumer work: the producer is blocked while the consumer works on the previous morsel. That is why the engine prefers parallel pipes with a distributor whenever the operator can handle unordered input.

Distributing morsels over threads

When the downstream operator wants parallel input, the pipe gets a distributor instead of a single channel. The send side spawns a task per receiver (crates/polars-stream/src/pipe.rs:323):

            State::NeedsDistributor { mut recv, senders } => {
                let num_pipelines = senders.len();
                let (mut distributor, distr_receivers) = morsel_distributor(
                    num_pipelines,
                    *DEFAULT_DISTRIBUTOR_BUFFER_SIZE,
                    self.seq_offset.clone(),
                );

                handles.push(scope.spawn_task(TaskPriority::High, async move {
                    while let Ok(morsel) = recv.0.recv().await {
                        if distributor.send((morsel, ())).await.is_err() {
                            break;
                        }
                    }

                    Ok(())
                }));

                for (mut send, mut recv) in senders.into_iter().zip(distr_receivers) {
                    handles.push(scope.spawn_task(TaskPriority::High, async move {
                        let wait_group = WaitGroup::default();
                        while let Ok((mut morsel, ())) = recv.recv().await {
                            morsel.set_consume_token(wait_group.token());
                            if send.0.send(morsel).await.is_err() {
                                break;
                            }
                            wait_group.wait().await;
                        }

                        Ok(())
                    }));
                }
            },

Read the last six lines carefully. The forwarder task takes a morsel out of the distributor, attaches a fresh consume token to it, sends it on, and then waits for that token to be dropped before taking the next morsel. The token is dropped by the downstream operator when it is done with the morsel. So a forwarder can be at most one morsel ahead of its worker, and it blocks entirely until the worker consumes.

The distributor itself is the interesting bit. It is not a round-robin queue; it tries to send to the hungriest receiver (crates/polars-async/src/primitives/distributor_channel.rs:132):

impl<T: Send> Sender<T> {
    /// Note: This intentionally takes `&mut` to ensure it is only accessed in a single-threaded
    /// manner.
    pub async fn send(&mut self, mut value: T) -> Result<(), T> {
        let num_receivers = self.inner.receivers.len();
        loop {
            // Fast-path.
            self.round_robin_idx += 1;
            if self.round_robin_idx >= num_receivers {
                self.round_robin_idx -= num_receivers;
            }

            let mut hungriest_idx = self.round_robin_idx;
            let mut shortest_len = self.upper_bound_len(self.round_robin_idx);
            for _ in 0..4 {
                let idx = ((self.rng.random::<u32>() as u64 * num_receivers as u64) >> 32) as usize;
                let len = self.upper_bound_len(idx);
                if len < shortest_len {
                    shortest_len = len;
                    hungriest_idx = idx;
                }
            }

            match unsafe { self.try_send(hungriest_idx, value) } {
                Ok(()) => return Ok(()),
                Err(SendError::Full(v)) => value = v,
                Err(SendError::Closed(v)) => value = v,
            }

Round-robin for the fast path, then four random probes looking for the receiver with the shortest queue, then a full scan of all receivers before parking. Each receiver has its own fixed-size ring buffer (bufsize, default 4, from POLARS_DEFAULT_DISTRIBUTOR_BUFFER_SIZE), so “hungriest” is well-defined and the work naturally balances out: a worker that is stuck on an expensive morsel stops draining its buffer, its buffer fills, and the distributor starts routing around it.

The morsel distributor in polars-stream wraps this with one more concern: the sequence numbers (crates/polars-stream/src/utils/morsel_distributor.rs:38):

impl<T: Send> MorselDistributorSender<T> {
    pub async fn send(&mut self, (mut morsel, payload): (Morsel, T)) -> Result<(), (Morsel, T)> {
        let orig_seq = morsel.seq();
        if Some(orig_seq) == self.prev_orig_seq {
            self.offset += 1;
            self.seq_offset.store(self.offset);
        }
        self.prev_orig_seq = Some(orig_seq);
        morsel.set_seq(orig_seq.offset_by_u64(self.offset));

        // Important: we have to drop the consume token before
        // going into the buffered distributor.
        drop(morsel.take_consume_token());
        self.sender.send((morsel, payload)).await
    }
}

When a single source’s rows are split across several pipes, two morsels can end up with the same sequence number on different pipes. Their relative order is then undefined, and any downstream operator that cares about order has no way to recover it. The distributor therefore bumps a global offset whenever it sees a repeated sequence number, so that all morsels derived from one source morsel get distinct, increasing sequence ids. The comment about dropping the consume token first is the other half of the backpressure story: the token from the previous stage is released before the morsel enters the buffered distributor, so the distributor’s own buffer is the only thing in flight.

Putting the order back

If the downstream operator does care about order, the pipe grows a linearizer instead of a distributor (crates/polars-stream/src/pipe.rs:266):

            State::NeedsLinearizer {
                receivers,
                mut send,
                maintain_order,
            } => {
                let num_pipelines = receivers.len();
                let (mut linearizer, inserters) =
                    Linearizer::<Priority<Reverse<MorselSeq>, Morsel>>::new_with_maintain_order(
                        num_pipelines,
                        *DEFAULT_LINEARIZER_BUFFER_SIZE,
                        maintain_order,
                    );

                let seq_offset = self.seq_offset.load();
                handles.push(scope.spawn_task(TaskPriority::High, async move {
                    while let Some(Priority(_, mut morsel)) = linearizer.get().await {
                        morsel.set_seq(morsel.seq().offset_by_u64(seq_offset));
                        if send.0.send(morsel).await.is_err() {
                            break;
                        }
                    }

                    Ok(())
                }));

The linearizer keeps exactly one item per sender in a binary heap and always emits the smallest sequence number available, refilling from the sender it just drained (crates/polars-async/src/primitives/linearizer.rs:95). This is a k-way merge, and it is what makes maintain_order possible without giving up parallelism.

The subtlety is that a linearizer can only emit when every sender has produced at least one item, or has closed. It therefore introduces a latency floor: the slowest worker gates the whole pipe. The maintain_order flag on the receiver decides whether the linearizer keeps per-sender ordering (heap) or just merges the streams as they arrive.

Notice also the backpressure comment in the linearizer’s input tasks (crates/polars-stream/src/pipe.rs:284):

                for (mut recv, mut inserter) in receivers.into_iter().zip(inserters) {
                    handles.push(scope.spawn_task(TaskPriority::High, async move {
                        while let Ok(mut morsel) = recv.0.recv().await {
                            // Drop the consume token, but only after the send has succeeded. This
                            // ensures we have backpressure, but only once the channel fills up.
                            let consume_token = morsel.take_consume_token();
                            if inserter
                                .insert(Priority(Reverse(morsel.seq()), morsel))
                                .await
                                .is_err()
                            {
                                break;
                            }
                            drop(consume_token);
                        }

The token is dropped only after the insert succeeded. Since the inserter’s channel is bounded (channel(buffer_size) with buffer_size = 4 by default), the insert blocks once the buffer is full, and only then does the consume token get released. The result is that backpressure appears exactly when the buffer fills, and not before.

Who decides what runs when

Morsels flow; nodes are scheduled separately. The graph is walked and runnable subgraphs are chosen (crates/polars-stream/src/execution_order.rs:13):

/// Chooses which of the given accumulating nodes to run, in order of preference:
/// 1. nodes whose outputs are ready to be consumed, so the result is not held;
/// 2. nodes that read from the most nodes holding data (see [`num_held_upstream`]), so that data
///    can be released;
/// 3. nodes with the most non-bounded nodes downstream (see [`num_non_bounded_downstream`]), so
///    a short branch does not hold its result while a long branch is processed;
/// 4. nodes with more outputs, so in-memory sinks (which have none) run last.
pub fn choose_accumulating_node(graph: &Graph, candidates: &[GraphNodeKey]) -> GraphNodeKey {

An “accumulating” node is one that holds state — a group-by, a join build side, a sort. Those cannot all run at once without blowing up memory, so the engine picks one (and at most one) per phase, preferring the one whose output can be drained soonest, then the one that is holding the most upstream data, then the one on the longest downstream path.

num_pipelines is simply the thread count (crates/polars-stream/src/execute.rs:293):

    let state = StreamingExecutionState {
        num_pipelines: polars_config::config().max_threads(),

Every morsel, every channel, every distributor receiver count is a multiple of this number. It is the one number that describes the parallelism of the whole engine.

The bottom line

The morsel is a small idea with a lot of leverage:

• The payload is spillable. SpillFrame means the engine can run queries larger than RAM, and that the memory manager — not the operator — decides what gets evicted.
• The channel is capacity one. Backpressure is not a policy bolted on top; it is the default behaviour of the primitive. The consume_token extends that guarantee across the distributor and the linearizer.
• The distributor balances by queue length. Workers that fall behind get skipped rather than blocking everyone.
• Sequence numbers are logical, not physical. They are doubled to leave room for a flag, and they are re-based when a source’s rows are split across pipes so that global order survives the split.

The trade-offs are real, though. A capacity-one channel means a one-to-one pipe cannot overlap producer and consumer work. A linearizer gates on the slowest worker. And a morsel of 100k rows is a fixed unit of memory that is replicated per worker at the boundaries of a stage.

If you want to see these decisions for yourself, the knobs are environment variables: POLARS_IDEAL_MORSEL_SIZE sets the morsel size, POLARS_DEFAULT_DISTRIBUTOR_BUFFER_SIZE and POLARS_DEFAULT_LINEARIZER_BUFFER_SIZE set the buffer depths. Set POLARS_VERBOSE=1 and the readers print their configuration, including the ideal morsel size they are using.