DataFusion Skips the Hash Table When the Input Is Already Sorted by the Group Key

TL;DR: AggregateExec asks the input’s equivalence properties for the longest permutation of the GROUP BY expressions that the existing sort order already satisfies. If that covers every key, the plan enters “full clustering” mode: a state machine knows when a group is complete, and GroupValuesClustered interns keys with no hash table, using Arrow’s partition kernel to find equal-key runs inside each batch. The same family of ideas ends in TopK, where the group set is bounded by the LIMIT.

Sorting is a hint, not a requirement

GROUP BY does not care about input order. But if the input happens to be sorted by the grouping key, then all rows of a group are contiguous, and contiguity is enormously valuable:

You can emit a group the moment you see a different key. No group is ever revisited.
You never need to look up “have I seen this key before?” — the answer is always “the previous row”.
Your memory footprint is one group, not N groups.

DataFusion does not require you to sort. It notices when the input is already sorted, and switches to a completely different execution path.

Detecting the clustering

The decision happens at plan construction, in AggregateExec::try_new_with_schema:

rust
// datafusion/physical-plan/src/aggregates/mod.rs:1304-1360
        let input_eq_properties = input.equivalence_properties();
        // Get GROUP BY expressions:
        let groupby_exprs = group_by.input_exprs();
        // If existing ordering satisfies a prefix of the GROUP BY expressions,
        // prefix requirements with this section. In this case, aggregation will
        // work more efficiently.
        // Copy the `PhysicalSortExpr`s to retain the sort options.
        let (new_sort_exprs, indices) =
            input_eq_properties.find_longest_permutation(&groupby_exprs)?;

        ...
        // Constant expressions never change, so they cannot mark a completed group.
        // Exclude them from both the ordering indices and the group expression count.
        let indices: Vec<usize> = indices
            .into_iter()
            .filter(|idx| {
                group_by.groups.iter().all(|group| !group[*idx])
                    && input_eq_properties
                        .is_expr_constant(&groupby_exprs[*idx])
                        .is_none()
            })
            .collect();

        let num_non_constant_groupby_exprs = groupby_exprs
            .iter()
            .filter(|expr| input_eq_properties.is_expr_constant(expr).is_none())
            .count();
        let mut group_clustering_mode = if indices.len() == num_non_constant_groupby_exprs
            && !indices.is_empty()
            && group_by.groups.len() == 1
        {
            GroupClusteringMode::Full
        } else if !indices.is_empty() {
            GroupClusteringMode::Partial(indices)
        } else {
            GroupClusteringMode::None
        };

Three outcomes:

rust
// datafusion/physical-plan/src/aggregates/order/mod.rs:47-62
pub enum GroupClusteringMode {
    /// No clustering guarantee is available to complete groups before the input ends.
    None,
    /// Rows with the same values at these grouping-expression indices form one
    /// contiguous range. When those values change, every group in the previous
    /// range is complete and can be emitted.
    ///
    /// For example, with `GROUP BY (a, b)`, `Partial(vec![0])` means all rows
    /// for each value of `a` are contiguous, while an `(a, b)` tuple may recur
    /// within that range.
    Partial(Vec<usize>),
    /// Rows with the same complete grouping tuple form one contiguous range.
    /// When the tuple changes, the previous group can be emitted.
    Full,
}

Full — every group key tuple is contiguous. GROUP BY a, b on input sorted by (a, b).
Partial — some keys are contiguous. GROUP BY a, b on input sorted by a only: all rows for a given a are contiguous, but (a, b) tuples may recur inside that range.
None — no guarantee; use the hash table.

Note the two filters applied to indices: grouping sets are excluded (they rewrite the key expression per set, so the ordering argument does not transfer), and constant group expressions are excluded (a constant never changes, so it cannot mark a boundary — but it also must not be counted towards the “all keys covered” test).

The mode then selects the stream implementation:

rust
// datafusion/physical-plan/src/aggregates/mod.rs:1723-1765
        let stream = match (self.mode, &self.group_clustering_mode) {
            (Partial, GroupClusteringMode::None) => StreamType::PartialHash(
                PartialHashAggregateStream::new(self, context, partition)?,
            ),
            (Partial, GroupClusteringMode::Partial(_) | GroupClusteringMode::Full) => {
                StreamType::ClusteredPartialAggregate(
                    ClusteredPartialAggregateStream::new(self, context, partition)?,
                )
            }
            ...
            (Final | FinalPartitioned, GroupClusteringMode::None) => {
                StreamType::FinalHash(FinalHashAggregateStream::new(
                    self, context, partition,
                )?)
            }
            (
                Final | FinalPartitioned,
                GroupClusteringMode::Partial(_) | GroupClusteringMode::Full,
            ) => StreamType::ClusteredFinalAggregate(ClusteredFinalAggregateStream::new(
                self, context, partition,
            )?),
            (Single | SinglePartitioned, GroupClusteringMode::None) => {
                StreamType::SingleHash(SingleHashAggregateStream::new(
                    self, context, partition,
                )?)
            }
            (
                Single | SinglePartitioned,
                GroupClusteringMode::Partial(_) | GroupClusteringMode::Full,
            ) => StreamType::ClusteredSingleAggregate(
                ClusteredSingleAggregateStream::new(self, context, partition)?,
            ),
        };

Knowing when a group is over

The “completion tracker” is deliberately tiny. For the full-clustering case it is a three-state machine that remembers the id of the group currently being accumulated:

rust
// datafusion/physical-plan/src/aggregates/order/full.rs:56-96
#[derive(Debug)]
pub struct GroupClusteringFull {
    state: State,
}

#[derive(Debug)]
enum State {
    /// Seen no input yet
    Start,

    /// Data is in progress. `current` is the current group for which
    /// values are being generated. Can emit `current` - 1
    InProgress { current: usize },

    /// Seen end of input: all groups can be emitted
    Complete,
}

impl GroupClusteringFull {
    // How many groups be emitted, or None if no data can be emitted
    pub fn emit_to(&self) -> Option<EmitTo> {
        match &self.state {
            State::Start => None,
            State::InProgress { current, .. } => {
                if *current == 0 {
                    // Can not emit if still on the first row
                    None
                } else {
                    // otherwise emit all rows prior to the current group
                    Some(EmitTo::First(*current))
                }
            }
            State::Complete => Some(EmitTo::All),
        }
    }

That is the whole trick: the number of the current group id is the number of groups that are complete. Because group ids are assigned densely in first-seen order and the input is clustered, group g is complete as soon as group g+1 appears. EmitTo::First(current) therefore means “everything before the group I am currently filling is done”.

For the partial-clustering case the tracker is a little richer, because the boundary is defined by a subset of the keys, and it has to compare the key values themselves:

rust
// datafusion/physical-plan/src/aggregates/order/partial.rs:88-100
    /// Data is in progress.
    InProgress {
        /// Smallest group index in the current run.
        current_run_start: usize,
        /// The key values of the current run.
        group_key: Vec<ScalarValue>,
        /// index of the current group for which values are being
        /// generated
        current: usize,
    },

rust
// order/partial.rs:192-207
    fn updated_group_key(
        current_run_start: usize,
        group_key: Option<Vec<ScalarValue>>,
        range_current_run_start: usize,
        range_group_key: Vec<ScalarValue>,
    ) -> Result<(usize, Vec<ScalarValue>)> {
        if let Some(group_key) = group_key {
            let sort_options = vec![SortOptions::new(false, false); group_key.len()];
            let ordering = compare_rows(&group_key, &range_group_key, &sort_options)?;
            if ordering == Ordering::Equal {
                return Ok((current_run_start, group_key));
            }
        }

        Ok((range_current_run_start, range_group_key))
    }

The boundary comparison uses Arrow’s compare_rows on the subset of key columns that define the run.

No hash table at all

Now the interesting part. When the input is fully clustered, the group key store doesn’t need a hash table. GroupValuesClustered interns keys using adjacency:

rust
// datafusion/physical-plan/src/aggregates/group_values/multi_group_by/clustered.rs:30-45
/// Columnar group keys for input clustered by the complete grouping tuple.
///
/// Equal keys must be contiguous across input batches. Within a batch Arrow's
/// partition kernel finds those runs. Only the first run can continue the last
/// buffered group; compare it with the stored columns, without retaining the
/// input batch. New group representatives use the existing column builders.
///
/// This removes hashing and hash-table storage, but does not change the dense
/// group-id or emission contracts. In particular, `First(n)` shifts the remaining
/// keys and their ids together. Inputs clustered by only a subset of the keys
/// must use a hash table.
pub(crate) struct GroupValuesClustered {
    schema: SchemaRef,
    columns: Vec<Box<dyn GroupColumn>>,
    new_groups: Vec<usize>,
}

The intern is four steps:

rust
// clustered.rs:102-134
impl GroupValues for GroupValuesClustered {
    fn intern(&mut self, cols: &[ArrayRef], groups: &mut Vec<usize>) -> Result<()> {
        groups.clear();
        let ranges = partition(cols)?.ranges();
        if ranges.is_empty() {
            return Ok(());
        }

        let old_len = self.len();
        let continues = old_len > 0
            && self
                .columns
                .iter()
                .zip(cols)
                .all(|(stored, input)| stored.equal_to(old_len - 1, input, 0));

        self.new_groups.clear();
        self.new_groups.extend(
            ranges
                .iter()
                .skip(usize::from(continues))
                .map(|range| range.start),
        );
        for (stored, input) in self.columns.iter_mut().zip(cols) {
            stored.vectorized_append(input, &self.new_groups)?;
        }

        let first_group = old_len - usize::from(continues);
        for (index, range) in ranges.iter().enumerate() {
            groups.resize(range.end, first_group + index);
        }
        Ok(())
    }

arrow_ord::partition::partition(cols) finds the runs of equal keys inside the batch — one vectorized kernel, no hashing.
The first run might be a continuation of the last group of the previous batch, so the stored column builders are compared against input row 0. If they match, that run is not a new group.
One representative row per new run is appended to the column builders — the same GroupColumn machinery as the hash-based path.
The group id for every row is first_group + run_index, computed with a resize (fill-with-value) rather than a per-row loop.

The memory layout is still columnar, but the “hash table” is gone entirely:

text
   before (hash-based)                     after (clustered)

   HashTable<(u64, GroupIndexView)>        (no hash table)
      0x9f2c -> group 0
      0x4411 -> group 1
              ...                            column builders hold one
   columns: [10, 20, 10]                    representative row per group:
             ["a","b","c"]                  [10, 20]
             ...                            ["a", "b"]

The correctness caveat is spelled out in the type’s own docs: the only reason this works is that the caller promised the input is clustered. If it is not, the same key would appear in two runs and become two groups. The promise comes from GroupClusteringMode::Full, which was derived from the input’s ordering properties.

The supported types are constrained:

rust
// clustered.rs:47-51
impl GroupValuesClustered {
    /// Types for which adjacent Arrow equality agrees with GROUP BY equality.
    /// Keep single-column specializations and floats/nested/encoded keys on the
    /// established path until their semantics and performance are validated.
    pub(crate) fn supports_schema(schema: &Schema) -> bool {
        schema.fields().len() > 1

“Adjacent Arrow equality agrees with GROUP BY equality” is the invariant that makes partition usable as a group detector. Floats are excluded (because GROUP BY folds -0.0 into +0.0 but Arrow’s equality does not), as are nested and dictionary-encoded types.

Note also fields().len() > 1: the single-column clustered case does not use the hash-free path. It uses GroupValuesColumn::<true> — the streaming variant of the hash-based columnar store, whose scalarized_intern preserves the property that group ids are assigned in input order.

Streaming out completed groups

The clustered partial stream materializes the completed prefix in bulk rather than dribbling groups out one at a time:

rust
// datafusion/physical-plan/src/aggregates/clustered_partial_stream.rs:62-72
/// # Group Clustering Optimization
///
/// For the aggregation work, the hash aggregation implementation is reused.
///
/// After each input batch, the group-clustering mode determines whether any
/// groups can be emitted eagerly to improve memory efficiency. For example, if
/// the input is ordered by `k` and the last group key seen is `k = 100`, all
/// groups with keys less than 100 are complete. Materialize that entire completed
/// prefix once, then emit slices of it before reading more input. This avoids
/// repeatedly removing small batches of groups and shifting the remaining hash
/// table and accumulator state.

“Materialize the entire completed prefix once” matters: emitting groups one at a time would require renumbering the surviving group ids and rewriting the hash table on every batch. Bulk emission amortizes that.

The memory story is the headline:

rust
// clustered_partial_stream.rs:74-90
/// ## Full group clustering
///
/// Every complete grouping tuple is contiguous. Ordering by every group key is
/// one way to establish this mode ...
///
/// Completed groups can be emitted as soon as the next group is observed. Thus,
/// only the current group remains active after completed groups are emitted, and
/// memory usage does not grow with the total number of groups.
///
/// If a memory reservation nevertheless fails, the stream returns the error
/// directly, indicating an unexpected behavior.
///
/// ## Partial group clustering
///
/// Rows are contiguous for a subset of the group keys ...
///
/// If one `a` value contains many distinct `b` values, the table may accumulate
/// enough groups to exceed the memory limit.
///
/// - `ClusteredPartialAggregateStream`: On reservation failure, it emits all current
///   intermediate states downstream and resets the table. The final stage can
///   merge repeated `(a, b)` state rows, so no disk spill is required.
/// - `ClusteredFinalAggregateStream`: It cannot emit incomplete final results. On
///   reservation failure, it sorts the current intermediate states by the complete
///   group key and spills them as one run. ...

Partial clustering is the interesting middle ground: the runs are complete, but the individual (a, b) groups inside a run are not, so the hash table still lives — but it only has to hold the groups of the current run, and on memory pressure the stream can flush and reset without spilling to disk, because the final stage can merge duplicate (a, b) state rows.

The other way to bound the group set: TopK

There is a second, orthogonal way to avoid an unbounded group set: don’t compute all the groups at all, compute only the ones that can survive the LIMIT.

AggregateExec::try_optimize_topk recognizes two shapes:

rust
// datafusion/physical-plan/src/aggregates/mod.rs:949-1002
    /// ## TopK MIN/MAX
    ///
    /// ```txt
    /// SELECT k, MAX(v) AS score FROM t GROUP BY k
    /// ORDER BY score DESC NULLS LAST LIMIT 10;
    ///
    /// Before:
    /// Sort: score DESC, fetch=10
    ///   Aggregate: k, MAX(v)
    ///     input
    ///
    /// After (Added info in `AggregateExec` to trigger heap optimization):
    /// Sort: score DESC, fetch=10
    ///   Aggregate: k, MAX(v), TopKMinMax(10)
    ///     input
    /// ```
    /// ...
    /// ## TopK DISTINCT
    ///
    /// ```txt
    /// SELECT DISTINCT k FROM t ORDER BY k ASC NULLS LAST LIMIT 10;
    ///
    /// Before:
    /// Sort: k ASC, fetch=10
    ///   Aggregate: k, no aggregates
    ///     input
    ///
    /// After:
    /// Sort: k ASC, fetch=10
    ///   Aggregate: k, TopKDistinct(10)
    ///     input
    /// ```

The optimizer keeps the Sort/Limit above (the aggregate may still produce more than K rows); it just annotates the aggregate so the execution can use a bounded heap instead of a full hash table.

GroupedTopKAggregateStream drives a PriorityMap, which is a hash table and a heap in one:

rust
// datafusion/physical-plan/src/aggregates/topk/priority_map.rs:27-37
/// A `Map<K, V>` / `PriorityQueue` combo that evicts the worst values after reaching `capacity`
pub struct PriorityMap {
    map: Box<dyn ArrowHashTable + Send>,
    heap: Box<dyn ArrowHeap + Send>,
    capacity: usize,
    mapper: Vec<(usize, usize)>,
    /// Mirror of the map's all-NULL group count, kept as a plain field so the
    /// per-row `insert` path can check it without a `dyn` call (measured to
    /// regress the topk_aggregate benchmarks when read through the trait)
    null_count: usize,
}

The eviction test is one comparison against the heap’s worst element:

rust
// topk/priority_map.rs:60-70
    pub fn insert(&mut self, row_idx: usize) -> Result<()> {
        debug_assert_eq!(self.null_count, 0);

        // if we're full, and the new val is worse than all our values, just bail
        if self.heap.is_worse(row_idx) {
            return Ok(());
        }
        assert!(self.map.len() <= self.capacity, "Overflow");

        self.insert_eligible(row_idx)
    }

rust
// topk/heap.rs:179-186
    fn is_worse(&self, row_idx: usize) -> bool {
        if !self.heap.is_full() {
            return false;
        }
        let new_val = self.batch.value(row_idx);
        let worst_val = self.heap.worst_val().expect("Missing root");
        (!self.desc && new_val > *worst_val) || (self.desc && new_val < *worst_val)
    }

The hash table underneath is not a HashMap but an index-addressed store, so the heap can refer to entries by index instead of by hash:

rust
// topk/hash_table.rs:41-52
/// A custom wrapper around `hashbrown::HashTable` that:
/// 1. limits the number of entries to the top K
/// 2. Allocates a capacity greater than top K to maintain a low-fill factor and prevent resizing
/// 3. Tracks indexes to allow corresponding heap to refer to entries by index vs hash
struct TopKHashTable<ID> {
    map: HashTable<usize>,
    // Store the actual items separately to allow for index-based access
    store: Vec<HashTableItem<ID>>,
    // Free indexes in the store for reuse
    free_indices: Vec<usize>,
    // Pool of reusable value locations, usually Strings
    free_slots: Vec<ID>,
    // The maximum number of entries allowed
    limit: usize,

Note the free-list for string slots: evicted strings are recycled rather than dropped, which matters when the key is a String and the workload is high-cardinality.

The group-by-only (TopKDistinct) variant uses the group key as both the key and the priority value, so SELECT DISTINCT k ... LIMIT 10 keeps the 10 smallest keys rather than the 10 arbitrary first-seen keys.

Trade-offs

You still pay for the sort. The clustering optimization only removes the hash table; the SortExec that produced the sorted input is still in the plan. For a query that would not otherwise be sorted, adding a sort to enable clustering is usually a loss — which is why DataFusion only uses the clustering guarantee when the input already has it.
The guarantee is per-partition. The clustering mode is derived from the input’s ordering properties, which are per-partition. If the plan has a RepartitionExec that does not preserve order, the guarantee is gone, and the mode reverts to None for the downstream aggregate.
Partial clustering still needs the hash table. Only full clustering removes it. With GROUP BY a, b on input sorted by a only, the (a, b) groups inside an a run still require interning, and the memory bound is “the number of groups in the current run”, not “one group”.
The type whitelist is conservative. Floats, nested types and dictionary-encoded keys stay on the hash path, because Arrow’s adjacent equality does not agree with GROUP BY equality for them. The comment says so: “Keep single-column specializations and floats/nested/encoded keys on the established path until their semantics and performance are validated.”
The hash-free path is multi-column only. Single-column clustering uses the streaming variant of the hash-based store, because the single-column primitive specialization was already faster than a partition-kernel-based approach.
TopK changes the answer’s completeness, not its correctness. The aggregate may produce more than K rows (hence the Limit stays above), and all-NULL groups are tracked separately so they are not silently dropped:

rust
// topk/priority_map.rs:128-137
    /// Track a group whose aggregate values are all NULL, so it can be emitted
    /// with a NULL value. MIN/MAX ignore NULL inputs, but an all-NULL group
    /// must still appear in the aggregation output; such groups all tie on the
    /// sort key, so tracking up to `capacity` of them preserves top-k semantics.
    pub fn insert_null(&mut self, row_idx: usize) {

TopK only applies to a narrow shape. One group key, one MIN/MAX aggregate or no aggregate at all, no filter, no grouping sets:

rust
// aggregates/mod.rs:1010-1016
        if limit == 0
            || matches!(self.kind, AggregateKind::DistinctLimit { .. })
            || self.group_by().has_grouping_set()
            || self.filter_expr().iter().any(Option::is_some)
        {
            return None;
        }

The takeaway

DataFusion treats input ordering as a capability, not as a requirement. The plan asks “what does the input already guarantee?”, and the answer picks one of three execution paths. In the strongest case — the input is clustered by the complete grouping tuple — the group hash table is not merely optimized, it is absent, replaced by Arrow’s partition kernel and a three-state machine that knows when a group is finished. In the TopK case, the group set is bounded by the query’s own LIMIT, and the hash table becomes an index-addressed store feeding a heap that evicts as it goes.

Both are the same insight: the number of live groups is the thing to minimize, and every input property that bounds it is worth a specialized code path.