TL;DR: SELECT * FROM t LIMIT 10 is the classic example of a query that looks cheap and is not. In a batched volcano engine, every operator produces batches of 8,192 rows, so a limit of 10 makes the engine read 8,192 rows, build a hash table of 8,192 rows, and throw away 8,182 of them. DataFusion’s answer has two halves: limit pushdown (move the limit down the plan so the scan itself stops early) and limit absorption (let an operator take the limit as a parameter and stop producing output once it is satisfied).

The problem is structural. DataFusion, like every vectorised query engine, uses a batched volcano model: operators pull RecordBatches of up to datafusion.execution.batch_size rows (8,192 by default) and push them up. The batching is what makes the engine fast — it is what lets Arrow kernels do their work with SIMD and tight loops — but it has a sharp edge at the tail of a query.

The physical optimizer’s LimitPushdown rule describes it exactly:

Background: vectorized volcano execution model

DataFusion uses a batched volcano model. For most operators, output is produced in batches of datafusion.execution.batch_size (default 8192), so the batch sizes typically look like:

text Copy

8192, 8192, ..., 8192, 100 (the final batch may be partial)

Example

For a join with an expensive, selective predicate:

text Copy

GlobalLimitExec: skip=0, fetch=10
-- NestedLoopJoinExec(on=expr_expensive_and_selective)
--- DataSourceExec()
--- DataSourceExec()

Under this model, NestedLoopJoinExec would keep working until it can emit a full batch (8192 rows), even though the query only needs 10.

That is 819 times more work than the query asked for, in the operator that is usually the most expensive one in the plan.

Two mechanisms

The rule’s documentation names them:

Plan Limit Absorption

In addition to pushing down GlobalLimitExec and LocalLimitExec nodes in the plan, some operators can “absorb” a limit and stop early during execution.

Pushdown removes the limit node from the plan and hands the fetch value to an operator that can honour it. Absorption is the operator using that value to stop producing output.

The interface is small:

// datafusion/physical-plan/src/execution_plan.rs
fn supports_limit_pushdown(&self) -> bool {
    false
}

and

// datafusion/datasource/src/source.rs
/// Return a copy of this DataSource with a new fetch limit
fn with_fetch(&self, _limit: Option<usize>) -> Option<Arc<dyn DataSource>>;

with_fetch returns None if the operator cannot honour the limit, and the optimizer then leaves a LimitExec in place instead. Note that the default with_fetch returns None, so an operator opts in to limit pushdown.

Pushing into the scan

The most valuable target is the file scan, because a scan that knows it only needs 10 rows can stop after 10 rows — no more pages fetched, no more row groups decoded.

The Parquet implementation threads the limit all the way down to the decoder:

// datafusion/datasource-parquet/src/push_decoder.rs
pub(crate) struct DecoderBuilderConfig<'a> {
    ...
    pub(crate) decoder_limit: Option<usize>,
    ...
}

if let Some(limit) = self.decoder_limit {
    builder = builder.with_limit(limit);
}

But before the decoder even starts, the row group plan is pruned:

// Prune by limit if limit is set and limit order is not sensitive
if let (Some(limit), false) = (prepared.limit, prepared.preserve_order) {
    row_groups.prune_by_limit(limit, rg_metadata, &prepared.file_metrics);
}

and prune_by_limit is a careful little function:

pub fn prune_by_limit(
    &mut self,
    limit: usize,
    rg_metadata: &[RowGroupMetaData],
    metrics: &ParquetFileMetrics,
) {
    let mut fully_matched_row_group_indexes: Vec<usize> = Vec::new();
    let mut fully_matched_rows_count: usize = 0;

    // Iterate through the currently accessible row groups and try to
    // find a set of matching row groups that can satisfy the limit
    for &idx in self.access_plan.row_group_indexes().iter() {
        if self.access_plan.is_fully_matched(idx) {
            let row_group_row_count = match &self.access_plan.inner()[idx] {
                RowGroupAccess::Skip => continue,
                RowGroupAccess::Scan => rg_metadata[idx].num_rows() as usize,
                RowGroupAccess::Selection(selection) => selection.row_count(),
            };
            fully_matched_row_group_indexes.push(idx);
            fully_matched_rows_count += row_group_row_count;
            if fully_matched_rows_count >= limit {
                break;
            }
        }
    }

    // If we can satisfy the limit with fully matching row groups,
    // rewrite the plan to do so
    if fully_matched_rows_count >= limit {
        ...
        metrics.limit_pruned_row_groups.add_pruned(pruned_count);
        ...
    }
}

Two things to notice.

First, it only counts fully matched row groups — row groups where the predicate was statically proven to select every row. A row group with a partial selection has an unknown output size, so it cannot be used to guarantee that the limit is satisfied; but it can still be scanned.

Second, the if fully_matched_rows_count >= limit guard. If the fully matched row groups cannot satisfy the limit on their own, the function does nothing — the plan is left alone, because pruning the rest of the row groups would be wrong (the limit might be satisfied by the partially matched ones). This is a case where the conservative behaviour is also the correct one.

The preserve_order condition

Note the false in if let (Some(limit), false) = (prepared.limit, prepared.preserve_order).

preserve_order is set when the file must be read in its natural order — for example, when the query is SELECT * FROM t ORDER BY partition_col LIMIT 10 and the ordering of the files matters. In that case you cannot skip row groups, because the first rows in file order are the ones the query wants, and those are in the first row groups — but the pruning logic here counts row groups by accumulated row count, not by position, so it would be wrong.

The comment in the source makes the intent explicit:

// `select * from t order by partition_col limit 10`.

The limit and the row filter

The other half of the Parquet story is that the limit also affects how the per-row filter is installed. The RgPlanEntry struct carries a fully_matched flag precisely so the decoder knows it can skip the per-row RowFilter for that row group:

#[derive(Debug, Clone)]
pub(crate) struct RgPlanEntry {
    pub(crate) rg_index: usize,
    /// `true` when static pruning proved every row of this RG satisfies the
    /// predicate, so the per-row `RowFilter` can be skipped as a no-op.
    pub(crate) fully_matched: bool,
    /// On-disk size of this row group, credited to the `bytes_processed` metric
    /// documented on `ParquetFileMetrics` once the scan is done with it.
    pub(crate) bytes: u64,
}

So a limit can turn “evaluate a filter on every row” into “evaluate a filter on no rows”, for the row groups that are entirely selected.

The optimizer rule

The physical rule that drives all of this is datafusion/physical-optimizer/src/limit_pushdown.rs (487 lines). Its state machine is a small struct tracking the global skip/fetch as it descends the plan:

if pushdown_plan.fetch().is_some() {
    if global_state.skip == 0 {
        global_state.satisfied = true;
    }
    (global_state.skip, global_state.fetch) = combine_limit(
        global_state.skip,
        global_state.fetch,
        0,
        pushdown_plan.fetch(),
    );
}

combine_limit is where two limits are merged: the tighter one wins, and skips add up. It is the standard “limit algebra” — LIMIT 10 OFFSET 5 is LIMIT 15 OFFSET 0, and two nested limits compose to the tighter one.

The rule also handles the case where the limit is already satisfied by what it found:

if let Some(fetch) = fetch
    && limit_satisfied_by_input(&limit_info.input, skip, fetch, stats_ctx)?
{
    // The input already produces at most `fetch` rows, so no new limit
    // node is needed. Mark satisfied so downstream won't re-add one,
    // but preserve skip/fetch so any nested limit nodes (e.g. an inner
    // GlobalLimitExec) can still be merged with the outer constraint.
    global_state.satisfied = true;
    ...

This uses statistics — if the input’s estimated row count is below the limit, the limit node is not needed at all and can be deleted from the plan. The plan gets smaller and the limit disappears entirely.

The catch

• A limit is not always pushable. UnionExec advertises supports_limit_pushdown() == true but combines partitions, so the rule has to be careful:

  rust Copy

  // Checking `supports_limit_pushdown()` alone is not enough. For example:
  // UnionExec advertises `supports_limit_pushdown() == true` because it can
  

• Skip is harder than fetch. The rule notes: “Execution plans can’t (yet) handle skip, so if we have one, we still need to add a global limit”. A LIMIT 10 OFFSET 100000 cannot be pushed into a scan, because the scan would have to count and discard 100,000 rows — which it could do, but the machinery is not there.

• The limit changes the meaning of the plan for the operators above it. A SortExec with a limit produces the top-k rows, not the first-k rows of a full sort. DataFusion handles this with a separate TopKExec and by pushing the limit into the sort as a “fetch” parameter, but the distinction between “limit” and “top-k” is a real semantic difference that the plan has to preserve.

• Pushing a limit into a scan changes the metrics. A scan that stopped after 10 rows did not process the whole file, and the bytes_processed metric has to reflect that. The pruned_row_groups metric exists for exactly this.

• Limits interact with the dynamic filters. A TopK that tightens its threshold and a pushed-down LIMIT are two different ways of saying “we do not need much more data”, and both can be active at once.

The general point

The batched volcano model is a performance decision: batching is what lets Arrow’s kernels shine. But it creates a tail-latency problem for queries that need few rows. The fix is not to abandon batching — it is to teach every operator about the limit, so that the last operator in the chain can tell the first one to stop.

That is the same shape as the dynamic-filter idea: an operator learns something at runtime and passes it upstream so the work upstream of it can be avoided. The limit is just the simplest possible instance of it.