DataFusion's Group Values Are Columnar: How Multi-Column GROUP BY Keys Skip the Row Copy

TL;DR: DataFusion never stores a group key as a row. It stores a u64 hash in a hashbrown table and keeps the values in one builder per key column. Interning is therefore two-stage: find-or-create a group id from the hash, then append the row’s value to column 0, column 1, … column N. A single-column GROUP BY gets a dedicated specialization; nested types get a row-format fallback per column, not per table.

The problem: `GROUP BY a, b` needs a group id per row

Every hash aggregation starts with the same question: given a row, which group does it belong to? The answer must be a dense usize id, because that id is what indexes into the accumulators (counts[group_id] += 1). The grouping key itself is a tuple of Arrow columns, and the tuple has to be interned: the same (a, b) must map to the same id, and different tuples to different ids.

There are two obvious layouts.

Row-per-group (what Arrow’s RowConverter gives you): one opaque byte buffer per group containing all key columns concatenated. This is compact and compares fast, but every comparison decodes bytes, and the values have to be re-encoded when you emit them back as Arrow arrays.

Column-per-group (what DataFusion actually does): keep N separate Arrow-ish builders, one per key column. Group g’s value for column c is row g of builder c. Comparison against an input row becomes N typed comparisons, and emission is essentially free — the builder’s buffer is the output array.

DataFusion picks the columnar layout for the common case, and says so:

rust
// datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs:219-227
    /// The actual group by values, stored column-wise. Compare from
    /// the left to right, each column is stored as [`GroupColumn`].
    ///
    /// Performance tests showed that this design is faster than using the
    /// more general purpose [`GroupValuesRows`]. See the ticket for details:
    /// <https://github.com/apache/datafusion/pull/12269>
    group_values: Vec<Box<dyn GroupColumn>>,

The dispatch happens once, at stream construction, in new_group_values (datafusion/physical-plan/src/aggregates/group_values/mod.rs:157-211):

rust
pub fn new_group_values(
    schema: SchemaRef,
    group_clustering: &GroupClustering,
) -> Result<Box<dyn GroupValues>> {
    if matches!(group_clustering, GroupClustering::Full(_))
        && GroupValuesClustered::supports_schema(&schema)
    {
        return Ok(Box::new(GroupValuesClustered::try_new(schema)?));
    }
    if schema.fields.len() == 1 {
        // ... single-column specializations: primitive / bytes / bytes_view / bool
    }
    if multi_group_by::supported_schema(schema.as_ref()) {
        Ok(Box::new(GroupValuesColumn::<false>::try_new(schema)?))
    } else {
        Ok(Box::new(GroupValuesRows::try_new(schema)?))
    }
}

Note the ordering: fully clustered input wins over everything (that is the “no hash table at all” path — see the ordered-aggregation article), then the single-column specializations, then the multi-column columnar path, then the row-format fallback.

The two-part structure

GroupValuesColumn is deliberately split into two halves that never touch:

text
        group id 0        group id 1        group id 2
     ┌───────────┐    ┌───────────┐    ┌───────────┐
col 0│    10     │    │    20     │    │    10     │   <- Vec<i32>
     ├───────────┤    ├───────────┤    ├───────────┤
col 1│   "foo"   │    │   "bar"   │    │   "baz"   │   <- Vec<u8> + offsets
     └───────────┘    └───────────┘    └───────────┘

hashbrown HashTable<(u64, GroupIndexView)>:
     0x9f2c... -> (group 0)      0x4411... -> (group 1)

The hash table maps hash → group id. It never stores key bytes.
The column builders store the key values, indexed by group id.

The GroupIndexView is a small trick worth reading (multi_group_by/mod.rs:136-174):

rust
/// The view(a u64) format is like:
///   +---------------------+---------------------------------------------+
///   | inlined flag(1bit)  | group index / index to group indices(63bit) |
///   +---------------------+---------------------------------------------+
struct GroupIndexView(u64);

The top bit says “this hash bucket is a chain”. In the common case a hash maps to exactly one group and the 63-bit payload is the group id itself — no indirection, no extra allocation. Only when two distinct keys collide on the same 64-bit hash does DataFusion allocate a Vec<usize> of candidate group ids and flip the flag:

rust
// multi_group_by/mod.rs:831-843
                let list_offset = self.group_index_lists.len();
                // Create new `group_index_list` including
                // `exist group index` + `new group index`.
                let exist_group_index = group_index_view.value() as usize;
                let new_group_index_list = vec![exist_group_index, group_idx];
                self.group_index_lists.push(new_group_index_list);
                // Update the `group_index_view` to non-inlined
                let new_group_index_view =
                    GroupIndexView::new_non_inlined(list_offset as u64);
                *group_index_view = new_group_index_view;

So a 64-bit hash collision — which sounds like it should be a non-event — costs you one extra indirection only for the colliding bucket, and the common case stays a flat 16-byte table entry.

The vectorized intern: four phases

The hot path is vectorized_intern (multi_group_by/mod.rs:476-528), used when the caller does not care about the order of group ids relative to input rows:

Hash the whole batch at once with create_hashes — one vectorized pass over all key columns.
Look every row’s hash up in the table (collect_vectorized_process_context). Misses become new groups immediately; hits become “candidate rows to compare”.
Compare candidates vectorized across all key columns (vectorized_equal_to).
Any row that hashed equal but compared unequal (a real hash collision) falls back to a scalarized path (scalarized_intern_remaining).

The comparison step is where the columnar layout pays off. Each column builder implements a vectorized equality that writes into a bitmask rather than returning a bool per row:

rust
// multi_group_by/primitive.rs:82-116
        // Build a packed comparison bitmask, then AND it into equal_to_results
        let num_bytes = n.div_ceil(8);
        let mut cmp_buf = vec![0u8; num_bytes];

        for (i, (&lhs_row, &rhs_row)) in lhs_rows.iter().zip(rhs_rows.iter()).enumerate()
        {
            if !equal_to_results.get_bit(i) {
                continue;
            }
            let left = ... self.group_values[lhs_row] ...;
            let right = ... array_values[rhs_row] ...;
            if left.is_eq(right.canonicalize()) {
                cmp_buf[i / 8] |= 1 << (i % 8);
            }
        }

        // AND the comparison result into the existing equal_to_results bitmask
        apply_bitwise_binary_op(
            equal_to_results.as_slice_mut(),
            0,
            &cmp_buf,
            0,
            n,
            |a, b| a & b,
        );

Two things happen here that are worth calling out.

Short-circuit across columns. equal_to_results starts all-true and each column ANDs its result in. Once a column has rejected a row, the next column’s loop skips it (if !equal_to_results.get_bit(i) { continue; }). So a GROUP BY (a, b, c) where a already discriminates never pays for comparing b and c on the rows a rejected. The comment at multi_group_by/mod.rs:646-648 says it plainly: “Usually, such rows having same hash but different value with exists rows are very few.”

Floats are canonicalized, not compared bitwise. canonicalize() folds -0.0 into +0.0 (single_group_by/primitive.rs:73-96), and it is applied on both sides — the stored value at append time and the incoming value at compare time. Otherwise -0.0 and +0.0 would land in different groups, because the hash is over to_bits().

Nullability is a compile-time constant. The builder is generic over NULLABLE:

rust
// multi_group_by/primitive.rs:47-52
pub struct PrimitiveGroupValueBuilder<T: ArrowPrimitiveType, const NULLABLE: bool> {
    data_type: DataType,
    group_values: Vec<T::Native>,
    nulls: NullBufferBuilder,
}

With NULLABLE = false the null buffer is never built and equal_to skips the null check entirely — the compiler deletes it. That is a per-query decision made once, at schema time, not a branch in the inner loop.

Why the columnar layout is faster

Three concrete wins, all visible in the code:

Emission is free. PrimitiveGroupValueBuilder::build (primitive.rs:268-283) moves the Vec<T::Native> straight into a ScalarBuffer — no copy, no re-encoding. The row-format path has to decode rows back into arrays.
Comparison is typed. is_eq on native ints is a single instruction; comparing two Arrow rows is a byte-wise memcmp over a varint-encoded buffer.
The hash table is tiny. It stores (u64, u64) per group — 16 bytes — not the key. The key bytes live in the column builders where they are contiguous and typed.

The single-column specialization goes further. GroupValuesPrimitive (group_values/single_group_by/primitive.rs:102-118) memoizes the hash alongside the group id:

rust
    /// Stores the `(group_index, hash)` based on the hash of its value
    ///
    /// We also store `hash` is for reducing cost of rehashing. Such cost
    /// is obvious in high cardinality group by situation.
    map: HashTable<(usize, u64)>,

Rehashing on every probe would be the dominant cost at high cardinality, so the hash is computed once per row and cached next to the group id. The collision check is then a u64 compare first, and only then a value compare:

rust
// single_group_by/primitive.rs:155-171
                    let insert = self.map.entry(
                        hash,
                        |&(g, h)| unsafe {
                            hash == h && self.values.get_unchecked(g).is_eq(key)
                        },
                        |&(_, h)| h,
                    );

The fallback: one row per group

When the schema contains a type with no column builder, DataFusion falls back to GroupValuesRows (group_values/row.rs:45-83), which stores the classic (u64, usize) raw table plus an Arrow Rows buffer:

rust
    /// keys: u64 hashes of the GroupValue
    /// values: (hash, group_index)
    map: HashTable<(u64, usize)>,

    /// The actual group by values, stored in arrow [`Row`] format.
    /// `group_values[i]` holds the group value for group_index `i`.
    group_values: Option<Rows>,

Notice the same “hash first, compare second” discipline, with a reference to the upstream PR that introduced it:

rust
// row.rs:152-163
            let entry = self.map.find_mut(target_hash, |(exist_hash, group_idx)| {
                // Somewhat surprisingly, this closure can be called even if the
                // hash doesn't match, so check the hash first with an integer
                // comparison first avoid the more expensive comparison with
                // group value. https://github.com/apache/datafusion/pull/11718
                target_hash == *exist_hash
                    && group_rows.row(row) == group_values.row(*group_idx)
            });

A newer refinement keeps the columnar fast path even for mixed schemas: if a single key column is a nested type that has no builder, only that column falls back to a row-encoded builder (RowsGroupColumn), and the other columns stay native:

rust
// multi_group_by/mod.rs:1054-1065
    match builder {
        Some(b) => Ok(b),
        // Generic fallback for nested types (Struct / List / LargeList /
        // FixedSizeList, recursively) that lack a type-specialized builder but
        // can be encoded by arrow's row format. This is what lets a mixed
        // schema keep the column-wise fast path for its native columns instead
        // of dropping the whole key onto `GroupValuesRows`.
        None if data_type.is_nested() && RowsGroupColumn::supports_type(data_type) => {
            Ok(Box::new(RowsGroupColumn::try_new(data_type.clone())?))
        }
        None => not_impl_err!("{data_type} not supported in GroupValuesColumn"),
    }

Memory accounting

The hash table is grown under the memory pool’s nose. insert_accounted (datafusion/common/src/utils/proxy.rs:144-185) bumps an accounting counter by the capacity delta before reserving, so a resize is visible to the MemoryReservation before the allocation happens:

rust
        if self.len() == self.capacity() {
            // need to request more memory
            let bump_elements = self.capacity().max(16);
            let bump_size = bump_elements * size_of::<T>();
            *accounting = (*accounting).checked_add(bump_size).expect("overflow");
            self.reserve(bump_elements, &hasher);
        }

GroupValuesColumn::size() (multi_group_by/mod.rs:1086-1106) then reports group_values + map_size + group_index_lists + the scratch buffers, so the aggregate stream’s reservation tracks the real footprint of the key store.

Trade-offs

The vectorized path pays a fixed cost per batch. It allocates and fills three scratch index buffers (append_row_indices, equal_to_row_indices, equal_to_group_indices) plus a BooleanBufferBuilder for the mask. For a batch of 8 KiB rows that is fine; for a stream of tiny batches it is overhead. The STREAMING const parameter exists precisely to opt out of the vectorized path when group-id order must match input order (the scalarized_intern path, multi_group_by/mod.rs:375-463).
Hash collisions are handled by a scalar fallback. The vectorized comparison can only answer “equal / not equal” per hash bucket. Rows that hash-equal but compare-unequal are re-processed one at a time (scalarized_intern_remaining). This is rare, but it is not free: a pathological hash distribution turns the vectorized path back into the scalar one, plus the overhead of having tried.
EmitTo::First(n) is destructive. Emitting the first n groups requires renumbering every surviving group id and rewriting the hash table (multi_group_by/mod.rs:1145-1180). This is why the clustered/ordered path emits completed groups in bulk rather than a few at a time.
Columnar does not mean free for wide keys. A GROUP BY on 50 columns means 50 builders, 50 hash-combining passes per row (create_hashes folds column hashes with combine_hashes, datafusion/common/src/hash_utils.rs:100-103), and 50 comparisons per candidate row. The row format wins when the key is genuinely wide and heterogeneous — but the mixed-schema fallback above keeps the damage local.
Every specialization has a domain. Floats, nested types and dictionary-encoded types are deliberately excluded from the fully-clustered path and from some single-column specializations. Each exclusion is a correctness decision (does Arrow’s == agree with GROUP BY equality?) made conservatively, and it is written down in the code rather than discovered at runtime.

The takeaway

The group key store is the hottest data structure in hash aggregation, and DataFusion’s bet is that the key should never be copied into a hash table at all. Hash it, keep 16 bytes per group in a hashbrown table, and keep the real values in typed, columnar builders that are also the output. Everything else in this file — the inlined group index view, the bitmask comparisons, the NULLABLE const parameter, the per-column row fallback — exists to protect that one decision.

The problem: GROUP BY a, b needs a group id per row