Node buffers & spill

Most operators stream: a record comes in, a record goes out. But a blocking operator, like a sort or a large aggregation, has to hold records while it works: you can’t emit the smallest row until you’ve seen them all. If the held set is bigger than memory allows, it must spill to disk. Clinker models that held set as a three-state enum (in memory, spilled, or mixed), and the enum is the whole state machine. This is one of the cleanest “an enum is a state machine” examples in the engine, and it’s how the engine keeps memory bounded even when a buffer outgrows RAM.

Prerequisites

• The worker pool

→

Leads to

• The memory arbitrator

What you’ll be able to do

• Name the three NodeBuffer states and say what each one means for where the records physically live.
• Trace the Spilled → Mixed transition and explain why it uses std::mem::replace rather than mutating in place.
• Write the push and spill methods of a three-state buffer enum, with the move on the spilled edge.
• Explain how a spilled buffer drains back in a fixed order, and why a tempfile-backed SpillFile leaves no garbage behind.

New terms in this lesson (each is also expanded inline at the point you first need it):

• buffer: the records a blocking operator holds while it works.
• spill: push held rows to disk to free RAM.
• bounded memory: keeping working memory under a budget regardless of input size.
• backpressure: the channel-bound pacing from the previous lesson (re-used here).

Predict first

✋ Predict first

A NodeBuffer that has already spilled all its rows to disk is in the Spilled state. Now one more record arrives and gets pushed onto it. Which state is the buffer in after that push, and where do the rows physically live? Write down your answer before revealing.

Decide on your answer before you open the reveal — committing a guess is what makes it stick.

Reveal the answer

The buffer is now Mixed, not Spilled and not back to Memory.

A push can’t go to disk on the spot (spilling is a separate, deliberate step), so the new record lands in a fresh in-memory tail that rides on top of the existing on-disk chunks. That split, old rows on disk, new rows in RAM, is exactly what Mixed encodes.

The trap is assuming a push always appends to wherever the buffer already is, so a Spilled buffer would stay Spilled (and somehow grow on disk). It doesn’t: the type changes, because the physical location of the records changed. Pushing onto Spilled is the one edge that has to build a brand-new variant and carry the old variant’s owned chunks across, and that move is what the rest of this lesson is about. Hold that prediction; the worked example earns it.

A buffer with three states

A node’s held records are a NodeBuffer, and its variants are the three places those records can live:

clinker-exec ·node_buffer.rs ·NodeBuffer type @19acdcb4

pub(crate) enum NodeBuffer {
    /// All events live in memory — records and punctuations in arrival order.
    Memory(Vec<StreamEvent>),
    /// Every record lives on disk, as spill files paired with row counts.
    /// Punctuations never spill — they wait in the `pending_puncts` sidecar.
    Spilled {
        chunks: Vec<(SpillFile<u64>, u64)>,
        pending_puncts: Vec<Punctuation>,
    },
    /// A memory tail accumulated after a partial spill.
    Mixed {
        mem: Vec<StreamEvent>,
        spills: Vec<(SpillFile<u64>, u64)>,
        pending_puncts: Vec<Punctuation>,
    },
}

The enum makes the illegal states unrepresentable: a buffer is in exactly one of these shapes, and every reader matches all three. Memory is the cheap default; Spilled means the records have been pushed to disk so they no longer count against the memory budget; Mixed is what you get when new records arrive after a spill: a fresh in-memory tail riding on top of the on-disk chunks.

 Memory               Spilled                  Mixed
 ┌──────────┐         ┌───────────────┐        ┌─────────────────────────────┐
 │ Vec in   │         │ spill files   │        │ mem tail  │  spill files     │
 │ RAM      │         │ on disk       │        │ in RAM    │  on disk         │
 └──────────┘         └───────────────┘        └─────────────────────────────┘
       │  spill (under memory pressure)  ▲            ▲
       └─────────────────────────────────┘            │ push after a spill
                                                       │ (new mem tail)

This is the engine’s answer to bounded memory: a Memory buffer can grow only until the budget says spill; after that its bulk lives on disk and only the new tail counts against RAM. In Threads & channels, a bounded channel paced a fast reader so it couldn’t pile records into memory: that’s backpressure. Spill is the matching trick one layer in: when a holding operator can’t be paced (it genuinely needs every row), the engine moves the bulk to disk instead of letting it grow without bound.

Transitions: std::mem::replace moves a state forward

Pushing a record onto a Spilled buffer promotes it to Mixed, carrying the existing spill chunks across. You can’t merely mutate in place; you need to move the old variant’s owned fields into the new variant, and std::mem::replace is the standard Rust move-out-then-replace trick:

crates/clinker-exec/src/executor/node_buffer.rs

pub(crate) fn push_event(&mut self, event: StreamEvent) {
    match self {
        Self::Memory(v) => v.push(event),
        Self::Mixed { mem, .. } => mem.push(event),
        Self::Spilled { .. } => {
            // move the chunks + puncts out of the old `Spilled`, leaving a temporary
            let (chunks, puncts) = match std::mem::replace(self, Self::Memory(Vec::new())) {
                Self::Spilled { chunks, pending_puncts } => (chunks, pending_puncts),
                _ => unreachable!(),
            };
            *self = Self::Mixed { mem: vec![event], spills: chunks, pending_puncts: puncts };
        }
    }
}

Why std::mem::replace? self is &mut, so you only have a borrow: you can’t move chunks out of *self and leave a hole, because a borrowed value must always stay valid. std::mem::replace swaps in a placeholder (Self::Memory(Vec::new())) and hands you the old value by ownership, so you can take its chunks apart and build Mixed. The Memory → Spilled transition happens elsewhere, at the admission boundary, when the memory arbitrator (the next lesson) says to spill. The point for now is that the type enforces the rules: each transition consumes the old state and produces a valid new one.

Spilling to disk, and draining back

When a buffer spills, its rows are written through a SpillWriter:

clinker-exec ·node_buffer_spill.rs ·spill_node_buffer fn @19acdcb4

pub(crate) fn spill_node_buffer(
    rows: Vec<(Record, u64)>,
    spill_dir: Option<&Path>,
    compress: bool,
) -> Result<Option<(SpillFile<u64>, u64)>, PipelineError> {
    // writes each (record, row_number) pair through a SpillWriter, returns the file + count
}

clinker-exec ·spill.rs ·SpillWriter type @19acdcb4

The on-disk format is a leading tag byte (so the reader knows whether the rest is LZ4-compressed), a JSON schema header, then length-prefixed postcard record frames. A SpillFile is backed by a tempfile::TempPath, so it auto-deletes when dropped. The RAII cleanup pattern (Data & Representation’s Drop) means a spilled buffer leaves no garbage behind even if the run aborts.

Draining unifies all three states into one iterator, in a fixed order: memory events first, then each spill file streamed back via a SpillReader, then the trailing punctuations. A downstream operator consumes a Mixed buffer exactly as it would a Memory one, so the spill is invisible at the drain interface. That uniformity is what lets the rest of the engine ignore whether a buffer fit in RAM.

Worked → completion → faded

Now you read the state machine, complete it, and finally write the whole thing; each rung adds exactly one idea over the last.

Worked: walk all three states

Here is a self-contained buffer that mirrors the real NodeBuffer shape, minus the disk IO. Run it and watch the buffer walk Memory → Spilled → Mixed. The std::mem::replace move on the Spilled → Mixed edge is the same trick the real push_event uses.

rust // editable

reset ▶ run

#[derive(Debug)]
enum Buffer {
  Memory(Vec<u32>),
  Spilled { on_disk: usize },               // rows "written to disk"
  Mixed { mem: Vec<u32>, on_disk: usize },  // a memory tail after a spill
}

impl Buffer {
  // Memory -> Spilled: pretend we wrote the Vec out, recording its length.
  fn spill(&mut self) {
      if let Buffer::Memory(v) = self {
          *self = Buffer::Spilled { on_disk: v.len() };
      }
  }

  fn push(&mut self, row: u32) {
      match self {
          Buffer::Memory(v) => v.push(row),         // append in place
          Buffer::Mixed { mem, .. } => mem.push(row), // append to the tail
          Buffer::Spilled { .. } => {
              // can't append to disk on the spot: build a fresh Mixed tail.
              // move the disk count OUT of the old state via replace, then rebuild.
              let on_disk = match std::mem::replace(self, Buffer::Memory(Vec::new())) {
                  Buffer::Spilled { on_disk } => on_disk,
                  _ => unreachable!(),
              };
              *self = Buffer::Mixed { mem: vec![row], on_disk };
          }
      }
  }
}

fn main() {
  let mut b = Buffer::Memory(vec![1, 2, 3]);
  println!("{b:?}");
  b.spill();          // Memory -> Spilled
  println!("{b:?}");
  b.push(4);          // Spilled -> Mixed (fresh mem tail)
  b.push(5);
  println!("{b:?}");
}

#[derive(Debug)] enum Buffer { Memory(Vec<u32>), Spilled { on_disk: usize }, // rows "written to disk" Mixed { mem: Vec<u32>, on_disk: usize }, // a memory tail after a spill } impl Buffer { // Memory -> Spilled: pretend we wrote the Vec out, recording its length. fn spill(&mut self) { if let Buffer::Memory(v) = self { *self = Buffer::Spilled { on_disk: v.len() }; } } fn push(&mut self, row: u32) { match self { Buffer::Memory(v) => v.push(row), // append in place Buffer::Mixed { mem, .. } => mem.push(row), // append to the tail Buffer::Spilled { .. } => { // can't append to disk on the spot: build a fresh Mixed tail. // move the disk count OUT of the old state via replace, then rebuild. let on_disk = match std::mem::replace(self, Buffer::Memory(Vec::new())) { Buffer::Spilled { on_disk } => on_disk, _ => unreachable!(), }; *self = Buffer::Mixed { mem: vec![row], on_disk }; } } } } fn main() { let mut b = Buffer::Memory(vec![1, 2, 3]); println!("{b:?}"); b.spill(); // Memory -> Spilled println!("{b:?}"); b.push(4); // Spilled -> Mixed (fresh mem tail) b.push(5); println!("{b:?}"); }

The compiler guarantees you handled every state in push; std::mem::replace lets you move owned data out of the old variant safely. That’s the whole spill state machine, minus the disk IO.

Completion: finish the spilled edge

Same buffer, but the Spilled arm of push has a gap. The replace-and-rebuild is the one move on the whole edge: fill in the line that constructs the new Mixed variant, carrying the disk count across and starting a fresh memory tail with the new row.

rust // editable

reset ▶ run

#[derive(Debug)]
enum Buffer {
  Memory(Vec<u32>),
  Spilled { on_disk: usize },
  Mixed { mem: Vec<u32>, on_disk: usize },
}

impl Buffer {
  fn spill(&mut self) {
      if let Buffer::Memory(v) = self {
          *self = Buffer::Spilled { on_disk: v.len() };
      }
  }

  fn push(&mut self, row: u32) {
      match self {
          Buffer::Memory(v) => v.push(row),
          Buffer::Mixed { mem, .. } => mem.push(row),
          Buffer::Spilled { .. } => {
              let on_disk = match std::mem::replace(self, Buffer::Memory(Vec::new())) {
                  Buffer::Spilled { on_disk } => on_disk,
                  _ => unreachable!(),
              };
              // TODO: build the new state. The old rows are on disk (on_disk),
              // and `row` is the first record of a fresh in-memory tail.
              *self = todo!();
          }
      }
  }
}

fn main() {
  let mut b = Buffer::Spilled { on_disk: 3 };
  b.push(4); // Spilled -> Mixed
  b.push(5);
  println!("{b:?}"); // expect: Mixed { mem: [4, 5], on_disk: 3 }
}

#[derive(Debug)] enum Buffer { Memory(Vec<u32>), Spilled { on_disk: usize }, Mixed { mem: Vec<u32>, on_disk: usize }, } impl Buffer { fn spill(&mut self) { if let Buffer::Memory(v) = self { *self = Buffer::Spilled { on_disk: v.len() }; } } fn push(&mut self, row: u32) { match self { Buffer::Memory(v) => v.push(row), Buffer::Mixed { mem, .. } => mem.push(row), Buffer::Spilled { .. } => { let on_disk = match std::mem::replace(self, Buffer::Memory(Vec::new())) { Buffer::Spilled { on_disk } => on_disk, _ => unreachable!(), }; // TODO: build the new state. The old rows are on disk (on_disk), // and `row` is the first record of a fresh in-memory tail. *self = todo!(); } } } } fn main() { let mut b = Buffer::Spilled { on_disk: 3 }; b.push(4); // Spilled -> Mixed b.push(5); println!("{b:?}"); // expect: Mixed { mem: [4, 5], on_disk: 3 } }

💡 Hint 1

The new tail holds exactly one row to start: the one just pushed. The disk count is unchanged; you’re carrying it across, not adding to it. So mem is vec![row] and on_disk is the count you moved out.

Show solution

*self = Buffer::Mixed { mem: vec![row], on_disk };

The new row opens a fresh in-memory tail (vec![row]), and on_disk rides across unchanged: the spilled rows are still on disk; nothing moved them. This is the Spilled → Mixed promotion exactly as the real push_event does it.

Faded: write the buffer yourself

Now with much less scaffolding. The Buffer enum and main are given; write both methods. spill turns a Memory buffer into Spilled (recording the row count) and leaves the other states alone. push appends in place for Memory/Mixed, and promotes Spilled → Mixed with the std::mem::replace move. Make main’s expected output match.

rust // editable

reset ▶ run

#[derive(Debug)]
enum Buffer {
  Memory(Vec<u32>),
  Spilled { on_disk: usize },
  Mixed { mem: Vec<u32>, on_disk: usize },
}

impl Buffer {
  // TODO: Memory -> Spilled (record the Vec's length as on_disk).
  // Other states: leave unchanged.
  fn spill(&mut self) {
      todo!()
  }

  // TODO: Memory/Mixed append in place; Spilled promotes to Mixed
  // by moving on_disk out with std::mem::replace.
  fn push(&mut self, row: u32) {
      todo!()
  }
}

fn main() {
  let mut b = Buffer::Memory(vec![1, 2]);
  b.spill();   // -> Spilled { on_disk: 2 }
  b.push(3);   // -> Mixed { mem: [3], on_disk: 2 }
  b.push(4);   // -> Mixed { mem: [3, 4], on_disk: 2 }
  println!("{b:?}"); // expect: Mixed { mem: [3, 4], on_disk: 2 }
}

#[derive(Debug)] enum Buffer { Memory(Vec<u32>), Spilled { on_disk: usize }, Mixed { mem: Vec<u32>, on_disk: usize }, } impl Buffer { // TODO: Memory -> Spilled (record the Vec's length as on_disk). // Other states: leave unchanged. fn spill(&mut self) { todo!() } // TODO: Memory/Mixed append in place; Spilled promotes to Mixed // by moving on_disk out with std::mem::replace. fn push(&mut self, row: u32) { todo!() } } fn main() { let mut b = Buffer::Memory(vec![1, 2]); b.spill(); // -> Spilled { on_disk: 2 } b.push(3); // -> Mixed { mem: [3], on_disk: 2 } b.push(4); // -> Mixed { mem: [3, 4], on_disk: 2 } println!("{b:?}"); // expect: Mixed { mem: [3, 4], on_disk: 2 } }

Show solution

fn spill(&mut self) {
    if let Buffer::Memory(v) = self {
        *self = Buffer::Spilled { on_disk: v.len() };
    }
}

fn push(&mut self, row: u32) {
    match self {
        Buffer::Memory(v) => v.push(row),
        Buffer::Mixed { mem, .. } => mem.push(row),
        Buffer::Spilled { .. } => {
            let on_disk = match std::mem::replace(self, Buffer::Memory(Vec::new())) {
                Buffer::Spilled { on_disk } => on_disk,
                _ => unreachable!(),
            };
            *self = Buffer::Mixed { mem: vec![row], on_disk };
        }
    }
}

spill only fires from Memory (the if let leaves Spilled/Mixed untouched). push is the exhaustive match you completed above: append for the two in-memory cases, replace-and-rebuild for Spilled. Drop any arm of that match and the compiler refuses to build; exhaustiveness is doing the work of guaranteeing you handled every state.

Real-source grounding

The buffer you just wrote is the shape of the engine’s real one. Read these in order:

clinker-exec ·node_buffer.rs ·NodeBuffer type @19acdcb4

What the compiler enforces: a NodeBuffer is in exactly one of Memory / Spilled / Mixed, and every consumer matches all three. There is no way to read records out without first deciding which state you’re in; the on-disk case can’t be silently skipped.

What a junior might misread: assuming a push onto Spilled stays Spilled (it promotes to Mixed), or assuming Mixed means the rows are duplicated in RAM and on disk. They aren’t: Mixed partitions the rows (old ones on disk, new tail in memory) with no overlap.

clinker-exec ·node_buffer_spill.rs ·spill_node_buffer fn @19acdcb4

The real spill writes through a SpillWriter into a tempfile-backed SpillFile. Because the file is owned by an RAII guard, dropping the buffer unlinks the file, the same Drop-based cleanup you met in Data & Representation, now doing the work of “no leftover spill garbage even if the run aborts.”

Why-bridge: spill is how a holding operator stays within budget

Why model the buffer as a state machine at all, instead of just letting a Vec grow? Because a blocking operator can’t apply backpressure to its own input the way the bounded channel does: a sort genuinely needs every row before it can emit the first one. If the held set outgrows RAM, the only options are abort or spill. The three-state enum is what makes spill invisible to the rest of the engine: a downstream operator drains a Mixed buffer exactly as it drains a Memory one, so nothing above the buffer has to know whether the run fit in memory.

Solid on the buffer state machine and spill? Skip ahead →

Retrieval checkpoint

✓ Checkpoint — node buffers & spill

💡 Hint 1

Read push_event’s Spilled arm. Why does it need std::mem::replace rather than mutating the fields directly? (Hint: self is a &mut borrow, so it has to move the owned chunks out of the old variant, leaving a valid placeholder behind.)

What the tour establishes

NodeBuffer is a three-state enum: Memory, Spilled, Mixed. push_event promotes Spilled → Mixed by moving the chunks/puncts out via std::mem::replace. Spilling writes rows through a SpillWriter (tagged, optionally LZ4, postcard frames) into a tempfile-backed SpillFile that auto-deletes on drop. Draining streams memory, then disk, then puncts, so the rest of the engine never has to know a buffer spilled.

// quick check

What does the Mixed variant of NodeBuffer represent?

• A buffer split evenly between two threads
• A memory tail of records that arrived after the buffer had already spilled, so some rows are in RAM and the rest are on disk
• A buffer holding both records and errors

// quick check

Pushing a record onto a Spilled buffer needs std::mem::replace. What goes wrong if you instead try to move chunks straight out of *self?

• Nothing: it compiles and is simpler
• self is a &mut borrow, so moving chunks out would leave the borrowed value half-empty (invalid). replace swaps in a placeholder and hands you the old value by ownership so you can take it apart
• std::mem::replace is faster but otherwise identical to a plain move

// quick check

Misconception check: an engineer says "once a buffer is Mixed, the same rows exist both in memory and on disk, so it uses double the memory." What's the correction?

• Correct: Mixed keeps a full copy in both places for safety
• Wrong. Mixed partitions the rows with no overlap: the older rows live only on disk, and only the new tail is in memory. It does not double-store anything
• Wrong. Mixed actually keeps everything in memory and the disk copy is a backup

// quick check

Spaced recall from Threads & channels: the previous lesson paced a fast source reader with a bounded channel. How does that backpressure relate to spilling a node buffer?

• They are the same mechanism: spilling is just a bounded channel with a bigger capacity
• Both keep memory bounded, but they apply in different places: backpressure paces an *input* reader by blocking its send; spill handles a *holding* operator that can't be paced (it needs every row) by moving the bulk to disk
• Backpressure replaces spilling: a bounded channel means a buffer never has to spill

You should be able to:

• You can name the three NodeBuffer states and where the records physically live in each
• You can trace the Spilled -> Mixed transition and explain why std::mem::replace is used
• You wrote spill and push for a three-state buffer, with the replace-and-rebuild on the spilled edge
• You can describe how a spilled buffer drains back in order and why SpillFile cleans itself up

Verify in the checkout:

grep -n 'enum NodeBuffer' crates/clinker-exec/src/executor/node_buffer.rs
grep -n 'fn push_event' crates/clinker-exec/src/executor/node_buffer.rs
grep -n 'fn spill_node_buffer' crates/clinker-exec/src/executor/node_buffer_spill.rs
grep -n 'struct SpillWriter' crates/clinker-exec/src/pipeline/spill.rs

Connections

You can now name the three NodeBuffer states, trace the Spilled → Mixed move, write the buffer yourself, and explain how it drains and cleans up. A buffer spills when memory gets tight. But who decides it’s tight, across all the operators running at once, and how do they coordinate without stepping on each other? That’s the memory arbitrator, and it’s the engine’s sharpest lesson in interior mutability.

Prerequisites

• The worker pool

→

Leads to

• The memory arbitrator

Go deeper on the Rust (optional, one-directional, the enum-as-state-machine idea from first principles, with no engine framing):

• Want enums-with-data and 'an enum is a state machine' built from scratch? Read the docs →

Glossary terms used: buffer, spill, bounded memory, backpressure.

Mark this lesson complete