Coalesce Batches, Don't Shrink Them: How DataFusion Rebuilds 8K-Row Batches Out of 3-Row Ones

TL;DR: Arrow kernels amortize their fixed cost over the rows in a batch; a 3-row batch pays the full per-batch overhead for 3 rows. concat_batches is the obvious fix and the wrong one: it holds the input and the output simultaneously (2x peak memory) and copies the data twice. arrow-rs’s BatchCoalescer builds output batches incrementally, column by column, so there is exactly one copy. DataFusion wraps it in LimitedBatchCoalescer, which additionally understands LIMIT (slicing at the boundary instead of copying) and can fuse “filter + coalesce” into one pass. As of DataFusion 52, CoalesceBatchesExec is deprecated: coalescing moved inside the operators (filter, repartition, async-func) instead of being a node in the plan.

The problem: fixed cost per batch

Arrow is a vectorized execution engine. arrow::compute::filter on an 8192-row batch uses SIMD and a single allocation; on a 3-row batch it does the same setup work for 3 rows. The per-batch overhead is not small: kernel dispatch, validity-buffer handling, allocator traffic, and — in DataFusion specifically — the bookkeeping that every operator does per batch.

DataFusion’s own config comment states the motivation plainly:

// datafusion/common/src/config.rs:903-907
/// Default batch size while creating new batches, it's especially useful for
/// buffer-in-memory batches since creating tiny batches would result in too much
/// metadata memory consumption
pub batch_size: ConfigNonZeroUsize, default = non_zero_usize_default(8192)

Two things are being conflated there, and both are real:

1. Throughput. Fixed per-batch cost divided by rows per batch.
2. Memory. Every RecordBatch carries a schema Arc, a column list of ArrayRefs, and per-array metadata. A million 3-row batches cost far more in metadata than the data.

And tiny batches are not an exotic input; they are what selective operators produce. A filter that keeps 0.1% of rows emits batches whose size is a random variable with mean 8 rows at a batch_size of 8192. Same for the probe side of a hash join, and for an aggregation that happens to have many distinct groups. Any operator whose output cardinality is data-dependent produces an unpredictable batch size, and the operators downstream have to cope.

The naive fix, and why it is slow

The obvious implementation is: accumulate batches in a Vec, and when the row count crosses the threshold, call concat_batches and emit the result. That works. It is also the wrong shape, for reasons that arrow-rs documents on BatchCoalescer:

If we use concat_batches to implement the same functionality, there are 2 potential issues:

1. At least 2x peak memory (holding the input and output of concat)
2. 2 copies of the data (to create the output of filter and then create the output of concat)

The 2x peak memory is fatal for a query engine that has a memory pool and a spill-to-disk path: the coalescer is exactly where you would not want to double your resident set. And the double copy is pure waste — filter writes a new array, then concat writes it again.

The actual mechanism: incremental, column-wise construction

BatchCoalescer (in arrow-select, re-exported as arrow::compute::BatchCoalescer) is a small state machine: you push batches in, and it hands you finished batches of exactly target_batch_size rows out, with the tail buffered until more input arrives or you call finish_buffered_batch().

The key property is that it does not hold “the input” and “the output”. It holds partially-built output columns. Pushing a batch into a column builder is one copy; the completed batch is that builder’s output, moved out. Peak memory is approximately the size of the batch being assembled, not twice it.

DataFusion’s wrapper adds the LIMIT semantics, in datafusion/physical-plan/src/coalesce/mod.rs:25-35:

// datafusion/physical-plan/src/coalesce/mod.rs:25-35
#[derive(Debug)]
pub struct LimitedBatchCoalescer {
    /// The arrow structure that builds the output batches
    inner: BatchCoalescer,
    /// Total number of rows returned so far
    total_rows: usize,
    /// Limit: maximum number of rows to fetch, `None` means fetch all rows
    fetch: Option<usize>,
    /// Indicates if the coalescer is finished
    finished: bool,
}

The constructor is where the first real trick lives:

// datafusion/physical-plan/src/coalesce/mod.rs:57-77 (abridged)
pub fn new(schema: SchemaRef, target_batch_size: usize, fetch: Option<usize>) -> Self {
    Self {
        inner: BatchCoalescer::new(schema, target_batch_size)
            .with_biggest_coalesce_batch_size(Some(target_batch_size / 2)),
        total_rows: 0,
        fetch,
        finished: false,
    }
}

with_biggest_coalesce_batch_size(Some(target / 2)) says: a batch that is already more than half the target does not need coalescing. Per arrow’s documentation, “batches that are already large are costly to coalesce and are efficient enough to process directly without coalescing” — they bypass coalescing when there is no buffered data, or when the buffered data already exceeds the limit. Without this, a source that hands you 100k-row batches (very common for Parquet row groups) would be copied, row by row, into an 8192-row buffer: a full copy of the data for zero benefit. With it, big batches pass straight through and only the small ones get merged.

The second trick is LIMIT-awareness. A naive coalescer would happily buffer 8192 rows and then throw 8191 of them away because the query said LIMIT 10. LimitedBatchCoalescer::push_batch avoids that by slicing the incoming batch at the limit boundary:

// datafusion/physical-plan/src/coalesce/mod.rs:90-125 (abridged)
pub fn push_batch(&mut self, batch: RecordBatch) -> Result<PushBatchStatus> {
    ...
    if let Some(fetch) = self.fetch {
        // limit previously reached
        if self.total_rows >= fetch {
            return Ok(PushBatchStatus::LimitReached);
        }
        // limit now reached
        if self.total_rows + batch.num_rows() >= fetch {
            let remaining_rows = fetch - self.total_rows;
            debug_assert!(remaining_rows > 0);

            let batch_head = batch.slice(0, remaining_rows);
            self.total_rows += batch_head.num_rows();
            self.inner.push_batch(batch_head)?;
            return Ok(PushBatchStatus::LimitReached);
        }
    }
    ...
}

batch.slice(0, remaining_rows) is zero-copy (Arrow arrays are just a buffer plus an offset), so trimming to the limit costs nothing. The PushBatchStatus::LimitReached signal then tells the caller to stop pulling from the input stream entirely — which matters because “stop early” is exactly how LIMIT gets its performance win in a distributed engine.

The third trick: fusing filter and coalesce

push_batch_with_filter (coalesce/mod.rs:132-180) goes further: instead of “filter into a new batch, then coalesce”, it hands the predicate mask to the coalescer so the selection happens during construction:

// datafusion/physical-plan/src/coalesce/mod.rs:169-180
let selected_count = filter.true_count();
if self.total_rows + selected_count >= fetch {
    let remaining = fetch - self.total_rows;
    let mask = match filter.null_count() {
        0 => filter.clone(),
        _ => prep_null_mask_filter(filter),
    };
    let end = mask
        .values()
        .set_indices()
        .nth(remaining - 1)
        .map_or(0, |i| i + 1);
    self.total_rows += remaining;
    self.inner
        .push_batch_with_filter(batch.slice(0, end), &mask.slice(0, end))?;
    return Ok(PushBatchStatus::LimitReached);
}

Note nth(remaining - 1): the mask’s true-index bitmap is walked to find the input row index of the last row the LIMIT allows, so the batch can be sliced at the source rather than after the fact. prep_null_mask_filter normalises nulls (a null in a boolean filter is “not selected”) so true_count() and set_indices() agree.

But the fused path has a trap, and DataFusion handles it explicitly:

// datafusion/physical-plan/src/coalesce/mod.rs:132-142
/// Pushes the next [`RecordBatch`] into the coalescer after applying `filter`,
/// avoiding a separate materialization pass compared to calling
/// [`filter_record_batch`] followed by [`Self::push_batch`].
///
/// Batches with `Utf8View` or `BinaryView` columns that reference data
/// buffers are filtered first and then pushed with [`Self::push_batch`]:
/// arrow's fused path keeps the whole input data buffers alive for the few
/// selected strings, while [`Self::push_batch`] copies them into compact
/// buffers.

Utf8View/BinaryView are Arrow’s “view” layouts: short strings are stored inline in the array, long strings point into shared data buffers. If you select 3 rows out of 8192 and the fused path keeps a reference to the entire data buffer, you have retained the whole buffer to keep 3 strings — a classic memory-retention bug. So for view types with data buffers, DataFusion deliberately takes the slower path (has_view_data_buffers, coalesce/mod.rs:205-210):

fn has_view_data_buffers(batch: &RecordBatch) -> bool {
    batch.columns().iter().any(|array| match array.data_type() {
        DataType::Utf8View => !array.as_string_view().data_buffers().is_empty(),
        DataType::BinaryView => !array.as_binary_view().data_buffers().is_empty(),
        _ => false,
    })
}

This is the kind of detail that only shows up in production: the “clever” fused path is disabled for exactly the layout where it would blow up memory.

Where the coalescing actually happens today

Here is the part that surprises people: CoalesceBatchesExec is deprecated.

// datafusion/physical-plan/src/coalesce_batches.rs:66-69
#[deprecated(
    since = "52.0.0",
    note = "We now use BatchCoalescer from arrow-rs instead of a dedicated operator"
)]
#[derive(Debug, Clone)]
pub struct CoalesceBatchesExec {

The old design inserted a plan node between every pair of operators. That is architecturally ugly: a node that does nothing but buffer, with its own metrics, its own partition, and an extra poll_next hop for every batch. The modern design pushes the coalescer into the operators that produce small batches:

• FilterExec builds a coalescer with its own batch size and fetch limit:

  rust Copy

  // datafusion/physical-plan/src/filter.rs:731-740
  let in_stream = self.input.execute(partition, context)?;
  ...
  batch_coalescer: LimitedBatchCoalescer::new(
      self.schema(),
      self.batch_size,
      self.fetch,
  ),

  The default is 8192 (filter.rs:81, FILTER_EXEC_DEFAULT_BATCH_SIZE), and the physical planner overrides it with the session’s datafusion.execution.batch_size (datafusion/core/src/physical_planner.rs:1198).

• RepartitionExec coalesces on the producer side, before the channel’s gate:

  rust Copy

  // datafusion/physical-plan/src/repartition/mod.rs:545-560 (abridged)
  // Coalesce on the producer side, before the channel's gate, so
  // the consumer never sees the per-input-task small batches.
  // Skip in preserve-order mode, where `StreamingMergeBuilder`
  // handles batching, and for unbounded inputs, where a residual
  // batch could otherwise be withheld indefinitely.
  let shared_coalescer = coalesce_batches.then(|| {
      SharedCoalescer::new(
          input.schema(),
          coalescer_batch_size,
          num_input_partitions,
      )
  });

  This is the interesting one: repartitioning inherently splits batches (one input batch becomes N output batches, one per destination partition), so it must re-merge them or the whole downstream pipeline degrades to one-row batches. The coalescer is shared per output partition and protected by a mutex (repartition/mod.rs:270-330), with an “active senders” counter so the last input task to finish is the one that flushes the residual batch.

  The config datafusion.execution.coalesce_batches (default true, datafusion/common/src/config.rs:928-932) gates it, and it is deliberately skipped for unbounded inputs — a streaming source must not have its rows withheld waiting for a batch to fill.

• AsyncFuncExec coalesces its input before invoking the user’s async function, so the UDF sees batch-sized work (datafusion/physical-plan/src/async_func.rs:237-243).

     BEFORE (CoalesceBatchesExec as a plan node)

     Scan ─► Filter ─► [CoalesceBatchesExec] ─► HashJoin ─► Agg
              │            (buffers, copies)     │
              └─ 3 rows     └─ 8192 rows         └─ 11 rows

     AFTER (coalescer inside the operator)

     Scan ─► Filter ─► HashJoin ─► Agg
              │  │         │         │
              │  └─ coalesces internally
              │     (no extra node, no extra poll hop)
              └─ 3 rows ──► 8192 ──► 11 rows

The same trick in DataFusion’s own code: block-doubling builders

There is a second, smaller version of the “grow the buffer amortized” idea inside DataFusion itself, in the StringView builders. Because Arrow’s Utf8View layout stores short strings inline (≤ 12 bytes) and long strings in shared data buffers, building one means producing a new set of data buffers plus a new set of views — the same “assemble a column incrementally” problem.

DataFusion’s builders solve it with block doubling:

// datafusion/functions/src/strings.rs:536-539
pub(crate) const STRING_VIEW_INIT_BLOCK_SIZE: u32 = 8 * 1024;
pub(crate) const STRING_VIEW_MAX_BLOCK_SIZE: u32 = 2 * 1024 * 1024;

// datafusion/functions/src/string/common.rs:613-621 (abridged)
// Long values are packed into `in_progress`; when full it is sealed into
// `completed` and a new, larger block is started — same block-doubling
// scheme as Arrow's `GenericByteViewBuilder`.
let mut in_progress: Vec<u8> = Vec::new();
let mut completed: Vec<Buffer> = Vec::new();
let mut block_size: u32 = STRING_VIEW_INIT_BLOCK_SIZE;

and the growth itself:

// datafusion/functions/src/string/common.rs:650-668 (abridged)
let required_cap = in_progress.len() + len;
if in_progress.capacity() < required_cap {
    if !in_progress.is_empty() {
        completed.push(Buffer::from_vec(std::mem::take(&mut in_progress)));
    }
    if block_size < STRING_VIEW_MAX_BLOCK_SIZE {
        block_size = block_size.saturating_mul(2);
    }
    let to_reserve = len.max(block_size as usize);
    #[expect(
        clippy::disallowed_methods,
        reason = "StringView's block size bounds growth, so reserve cannot overflow capacity arithmetically. This hot loop intentionally avoids the extra `try_reserve` checks. It remains subject to allocator failure/OOM, which must be managed externally."
    )]
    in_progress.reserve(to_reserve);
}

Three things to notice:

• Amortized growth. The block size doubles (8 KiB → 16 KiB → … → 2 MiB cap), so the number of reallocations for n bytes is O(log n), and each sealed block becomes an immutable Buffer that is shared rather than copied.
• The blocks are the output format. StringViewArray is defined as a list of data buffers plus a view per row; sealing a block into completed is literally producing one of the array’s buffers. There is no final “copy everything into one big buffer” pass.
• A deliberate lint escape. This hot loop uses Vec::reserve, not Vec::try_reserve — the workspace lint (clippy.toml:4) normally forbids that so that allocation failures are reported instead of panicking. Here the escape is justified in the reason string: the block size bounds the growth so it cannot overflow arithmetically, and the remaining risk (allocator OOM) is “managed externally” by the memory pool.

That is the same engineering taste as the coalescer: never materialise the whole thing twice; build the output structure as you go.

The opposite direction: splitting

Coalescing has a symmetric problem — if you merge everything to 8192 rows, you destroy parallelism downstream, because a single 8192-row batch cannot be processed by two threads. DataFusion handles that with BatchSplitStream (datafusion/physical-plan/src/stream.rs:626-690), which slices oversized batches back down to batch_size:

// datafusion/physical-plan/src/stream.rs:668-680 (abridged)
let remaining = batch.num_rows() - self.offset;
let to_take = remaining.min(self.batch_size);
let out = batch.slice(self.offset, to_take);

self.metrics.batches_split.add(1);
self.offset += to_take;

Slicing is zero-copy, so splitting is cheap in a way that merging is not. That asymmetry is the whole reason the coalescer had to be built so carefully.

Trade-offs and failure modes

• Latency. A coalescer delays the first row until the target is reached. For an interactive LIMIT 1 query over a slow source, that means waiting for batch_size rows to arrive. The fetch/LIMIT path mitigates this by slicing at the boundary, but it can only help once rows have arrived.
• Memory. The coalescer holds up to target_batch_size rows of input plus the output it is building. With batch_size = 8192 and wide rows, that is a real allocation held per partition per operator. It is bounded, but it is not free, and it interacts with the memory pool.
• Unbounded inputs. As repartition/mod.rs:549-551 notes, a residual batch on an unbounded stream may never be flushed. Coalescing must be skipped there.
• Preserve-order mode. Also skipped, because the ordering machinery (StreamingMergeBuilder) does its own batching; two mechanisms would fight.
• The view-type fallback. The fused path is slower for Utf8View/BinaryView batches with data buffers, and DataFusion accepts that cost deliberately.
• CoalesceBatchesExec still exists and is still used by the topk_repartition optimizer rule (datafusion/physical-optimizer/src/topk_repartition.rs:56), but it is deprecated and its behaviour is now just a thin wrapper over LimitedBatchCoalescer — the same primitive, exposed as a node for compatibility.

The takeaway

The interesting lesson is not “merge small batches”. It is that the merge has to be incremental to avoid the 2x-memory trap, limit-aware so that LIMIT stays cheap, fused with the producing operator so no plan node is needed, and careful about layout-specific retention (Arrow views). DataFusion solved this once, moved the primitive into arrow-rs, and then dissolved the operator into the places that actually produce small batches. The plan you EXPLAIN today no longer shows coalescing at all — which is exactly what a well-integrated optimization looks like when it stops being a special case.