COUNT(*) Is a Sum of Ones: How DataFusion Makes Counting the Cheapest Aggregate

TL;DR: DataFusion rewrites COUNT(*) into COUNT(1) where the 1 is a non-null literal, not a column. That single fact unlocks a branchless counting path: no value loads, no null checks, no per-group “did I see a value” bitmap. The same “sum of ones” idea reappears in the partial-aggregation skip optimization, where convert_to_state materializes “an array of 1s for each row in the input batch”.

The rewrite that actually exists

Let’s be precise, because the folklore version of this story (“DataFusion rewrites COUNT(*) to SUM(1)”) is not what the code does. What the code does is the first half of that idea, and it is the half that matters.

COUNT(*) is expanded at planning time into a count over a sentinel literal:

rust
// datafusion/common/src/utils/expr.rs:22-24
///  The value to which `COUNT(*)` is expanded to in
///  `COUNT(<constant>)` expressions
pub const COUNT_STAR_EXPANSION: ScalarValue = ScalarValue::Int64(Some(1));

And the aggregate planner performs the expansion:

rust
// datafusion/functions-aggregate/src/planner.rs:82-112
        // handle count() and count(*) case
        // convert to count(1) as "count()"
        // or         count(1) as "count(*)"
        #[expect(deprecated)]
        if raw_expr.func.name() == "count"
            && (raw_expr.args.len() == 1
                && matches!(raw_expr.args[0], Expr::Wildcard { .. })
                || raw_expr.args.is_empty())
        {
            ...
            let new_expr = Expr::AggregateFunction(AggregateFunction::new_udf(
                func,
                vec![Expr::Literal(COUNT_STAR_EXPANSION, None)],
                distinct,
                filter,
                order_by,
                null_treatment,
            ));

Note COUNT_STAR_EXPANSION is Int64(Some(1)) — not a null. That is deliberate. COUNT(1) counts non-null values, and a non-null literal is counted once per row, which is exactly COUNT(*). Had the sentinel been Int64(None), the count would be zero.

The consequence is enormous: the argument to the accumulator is a constant column of 1s with no null buffer. Every downstream fast path that keys off “does this input have nulls?” takes the cheap branch.

Where the ones actually get added

COUNT has a purpose-built groups accumulator:

rust
// datafusion/functions-aggregate/src/count.rs:649-698
/// An accumulator to compute the counts of [`PrimitiveArray<T>`].
/// Stores values as native types, and does overflow checking
///
/// Unlike most other accumulators, COUNT never produces NULLs. If no
/// non-null values are seen in any group the output is 0. Thus, this
/// accumulator has no additional null or seen filter tracking.
#[derive(Debug)]
struct CountGroupsAccumulator {
    /// Count per group.
    ///
    /// Note this is an i64 and not a u64 (or usize) because the
    /// output type of count is `DataType::Int64`. Thus by using `i64`
    /// for the counts, the output [`Int64Array`] can be created
    /// without copy.
    counts: Vec<i64>,
}

impl GroupsAccumulator for CountGroupsAccumulator {
    fn update_batch(
        &mut self,
        values: &[ArrayRef],
        group_indices: &[usize],
        opt_filter: Option<&BooleanArray>,
        total_num_groups: usize,
    ) -> Result<()> {
        assert_eq!(values.len(), 1, "single argument to update_batch");
        let values = &values[0];

        // Add one to each group's counter for each non null, non
        // filtered value
        self.counts.resize(total_num_groups, 0);
        accumulate_indices(
            group_indices,
            values.logical_nulls().as_ref(),
            opt_filter,
            |group_index| {
                // SAFETY: group_index is guaranteed to be in bounds
                let count = unsafe { self.counts.get_unchecked_mut(group_index) };
                *count += 1;
            },
        );

        Ok(())
    }

Note what is not here. There is no NullState. There is no SeenValues bitmap. There is no “have I seen a value for this group” state, because COUNT never produces NULL — the output for an empty group is 0, not NULL. The doc comment on NullState spells out the cost that COUNT avoids:

rust
// datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/accumulate.rs:30-36
/// If the input has nulls, then the accumulator must potentially
/// handle each input null value specially (e.g. for `SUM` to mark the
/// corresponding sum as null)
///
/// `NullState` tracks if it has seen *any* value for each group when filters or
/// sparse group indices may omit input for a registered group.

SUM needs all of that. SUM’s groups accumulator is the generic primitive op accumulator:

rust
// datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/prim_op.rs:41-63
pub struct PrimitiveGroupsAccumulator<T, F>
where
    T: ArrowPrimitiveType + Send,
    F: Fn(&mut T::Native, T::Native) + Send + Sync + 'static,
{
    /// values per group, stored as the native type
    values: Vec<T::Native>,
    ...
    /// Track nulls in the input / filters
    null_state: NullState,
    /// Function that computes the primitive result
    prim_fn: F,
}

and NullState::accumulate has to conditionally maintain a per-group “seen” bitmask:

rust
// accumulate.rs:195-212
        // Skip per-value null handling when every input value is valid and all
        // newly registered groups are represented. Pre-filtered inputs can have
        // sparse group indices despite not passing a filter to the accumulator.
        if opt_filter.is_none()
            && values.null_count() == 0
            && let SeenValues::All { num_values } = &mut self.seen_values
            && new_groups_are_dense(group_indices, *num_values, total_num_groups)
        {
            accumulate(group_indices, values, None, value_fn);
            *num_values = total_num_groups;
            return;
        }

        let seen_values = self.seen_values.get_builder(total_num_groups);
        accumulate(group_indices, values, opt_filter, |group_index, value| {
            seen_values.set_bit(group_index, true);
            value_fn(group_index, value);
        });

Three conditions must hold to skip the bookkeeping: no filter, no nulls, and the new group ids are dense. For SUM(x) where x is a nullable column, the second condition fails on any batch containing a null, and you fall into the bitmask path plus a value load per row.

COUNT sidesteps all of it, because it does not use NullState at all. It calls the lower-level accumulate_indices directly, which for a null-free input is a bare loop:

rust
// accumulate.rs:591-604
pub fn accumulate_indices<F>(
    group_indices: &[usize],
    nulls: Option<&NullBuffer>,
    opt_filter: Option<&BooleanArray>,
    mut index_fn: F,
) where
    F: FnMut(usize) + Send,
{
    match (nulls, opt_filter) {
        (None, None) => {
            for &group_index in group_indices.iter() {
                index_fn(group_index)
            }
        }

The per-row work for COUNT(*) is therefore: read group_indices[row], increment counts[group_index]. No array dereference, no null bit test, no branch. It is the tightest possible aggregation loop.

Compare the SUM inner loop, which must load a value and add it:

rust
// prim_op.rs:104-116
        self.null_state.accumulate(
            group_indices,
            values,
            opt_filter,
            total_num_groups,
            |group_index, new_value| {
                // SAFETY: group_index is guaranteed to be in bounds
                let value = unsafe { self.values.get_unchecked_mut(group_index) };
                (self.prim_fn)(value, new_value);
            },
        );

The literal column is built once and sliced forever

One more piece of engineering: the 1 column does not get rebuilt per batch.

AggregateArgument recognizes literals and caches the materialized array:

rust
// datafusion/physical-plan/src/aggregates/aggregate_argument.rs:35-53
/// One argument of an aggregate function, such as `x` in `SUM(x)`.
///
/// Accumulators take one value per input row, so an argument that evaluates to
/// a single value must be expanded to an array for every batch. A literal
/// argument, such as the `1` in `COUNT(1)` (which is how `COUNT(*)` is planned)
/// or the `','` in `STRING_AGG(x, ',')`, has the same value for every batch of
/// the query. Its array is built on demand and reused for later batches, which
/// receive a zero-copy slice of it.
pub(in crate::aggregates) struct AggregateArgument {
    expr: Arc<dyn PhysicalExpr>,
    /// Set when `expr` is a [`Literal`].
    literal: Option<LiteralArray>,
}

rust
// aggregate_argument.rs:119-131
    fn array_of_size(&mut self, num_rows: usize) -> Result<ArrayRef> {
        if let Some(array) = &self.array
            && array.len() >= num_rows
        {
            return Ok(array.slice(0, num_rows));
        }

        let array = self.value.to_array_of_size(num_rows)?;
        if array.get_array_memory_size() <= MAX_KEPT_LITERAL_ARRAY_BYTES {
            self.array = Some(Arc::clone(&array));
        }
        Ok(array)
    }

So for COUNT(*), the entire “argument evaluation” step is a pointer bump and a slice after the first batch. The 1 MiB cap (MAX_KEPT_LITERAL_ARRAY_BYTES) prevents a pathological literal from being retained forever.

Counting at plan time

Because the argument is a literal, COUNT(*) can be answered from statistics without touching the data at all:

rust
// datafusion/functions-aggregate/src/count.rs:434-443
        } else if let Some(lit_expr) = expr.downcast_ref::<expressions::Literal>() {
            // A non-null literal is counted once per row (this covers the common
            // `count(*)` construct). A non-null literal is never counted.
            let count = if lit_expr.value().is_null() {
                0
            } else {
                i64::try_from(num_rows).ok()?
            };
            return Some(ScalarValue::Int64(Some(count)));
        }

A non-null literal contributes exactly one per row, so the count is just the row count of the input. A null literal would contribute zero — the code handles both.

The optimizer-level count → sum

There is a literal count → sum rewrite in DataFusion, but it lives in a different place: the single_distinct_to_groupby rule, which decomposes count(DISTINCT b) into a two-phase plan. When a non-distinct count sits alongside the distinct aggregate, its outer phase becomes sum over the inner phase’s partial counts:

text
   Before:
   SELECT a, count(*), count(DISTINCT b)
   FROM t GROUP BY a

   After:
   SELECT a,
          CASE WHEN sum(alias2) IS NOT NULL THEN sum(alias2) ELSE 0 END,
          count(alias1)
   FROM (
     SELECT a, b as alias1, count(*) as alias2
     FROM t
     GROUP BY a, b
   )
   GROUP BY a

rust
// datafusion/optimizer/src/single_distinct_to_groupby.rs:343-384
                            } else {
                                index += 1;
                                let alias_str = format!("alias{index}");
                                // `count`'s outer phase is `sum`, over the
                                // partial counts the inner group-by produced.
                                let rollup = count_rollup
                                    .as_ref()
                                    .filter(|rollup| rollup.is_count(&func));
                                let outer_func = match rollup {
                                    Some(rollup) => Arc::clone(&rollup.sum),
                                    None => Arc::clone(&func),
                                };
                                ...
                                // `sum` of no rows is NULL where `count` is 0.
                                // Restoring the 0 also keeps the column
                                // non-nullable, as `count` had it.
                                let proj =
                                    when(outer.clone().is_not_null(), outer.clone())
                                        .otherwise(lit(0_i64))?;

Note the correction: SUM of zero rows is NULL, COUNT of zero rows is 0. The CASE restores the zero — and, as the comment says, keeps the column non-nullable, which COUNT guarantees and SUM does not.

The rule also refuses to do this when it does not pay. The count rewrite moves every other aggregate down into the inner group-by, whose grain is now one row per (group, distinct value) rather than one row per group:

rust
// single_distinct_to_groupby.rs:176-196
/// Whether the rewrite is worth extending to a plan that only qualifies because
/// of the non-distinct `count`.
///
/// The rewrite is not free: every other aggregate moves down to the inner group
/// by, which has a row per `(group, distinct value)` pair rather than per group,
/// and keeps its state at that finer grain. What pays for that is taking the
/// distinct aggregate off `GroupsAccumulatorAdapter` and its one boxed
/// accumulator per group. ...
fn rewrite_pays_for_count(

This is the “is the rewrite worth it” question that DataFusion asks explicitly, and it is a nice illustration of the general principle: a rewrite that eliminates a per-group boxed accumulator can be worth paying a finer grain for, but only if the eliminated thing was actually expensive.

Skipping work: when counting a billion rows is pointless

The partial stage of a multi-phase aggregation exists to reduce cardinality before the shuffle. When the input is high-cardinality, it does the opposite: it builds a hash table with one entry per row, then throws the table away and ships all the rows downstream anyway.

DataFusion detects this at runtime and bails out:

rust
// datafusion/physical-plan/src/aggregates/skip_partial.rs:85-104
    pub(super) fn update_state(&mut self, input_rows: usize, num_groups: usize) {
        if self.is_locked {
            return;
        }
        self.input_rows += input_rows;
        self.num_groups = num_groups;
        if self.input_rows >= self.probe_rows_threshold {
            self.should_skip = self.num_groups as f64 / self.input_rows as f64
                > self.probe_ratio_threshold;
            // Set is_locked to true only if we have decided to skip, otherwise we can try to skip
            // during processing the next record_batch.
            self.is_locked = self.should_skip;
        }
    }

The probe watches the ratio of distinct groups to input rows. Once it exceeds the configured threshold (default 0.8, and only after at least skip_partial_aggregation_probe_rows_threshold rows have been seen), the partial stream switches modes:

rust
// datafusion/physical-plan/src/aggregates/hash_stream.rs:99-108
/// # Optimization: Partial Aggregation Skip
///
/// Partial aggregation can be counterproductive for high-cardinality inputs,
/// where most rows create distinct groups. The stream probes the ratio of
/// accumulated groups to input rows while it is still aggregating. If the ratio
/// crosses the configured threshold and all aggregate accumulators can convert
/// raw inputs directly to partial state, the stream emits any already
/// accumulated groups, then switches to a skip state. In that state, each
/// remaining input batch is converted directly to partial aggregate state rows
/// without inserting the rows into the grouped hash table.

And here is where the “sum of ones” idea comes back full circle. The GroupsAccumulator trait has a convert_to_state method documented as:

rust
// datafusion/expr-common/src/groups_accumulator.rs:335-346
    /// Converts an input batch directly to the intermediate aggregate state.
    ///
    /// This is the equivalent of treating each input row as its own group. It
    /// is invoked when the Partial phase of a multi-phase aggregation is not
    /// reducing the cardinality enough to warrant spending more effort on
    /// pre-aggregation ...
    ///
    /// Examples:
    /// * `COUNT`: an array of 1s for each row in the input batch.
    /// * `SUM/MIN/MAX`: the input values themselves.

“COUNT: an array of 1s for each row in the input batch.” The skip path literally materializes the sum-of-ones. For COUNT(*) that is the sentinel literal column itself; for COUNT(x) it is a null-masked array of ones.

rust
// datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs:128-163
impl AggregateHashTable<PartialSkipMarker> {
    pub(in crate::aggregates) fn convert_batch_to_state(
        &mut self,
        batch: &RecordBatch,
    ) -> Result<RecordBatch> {
        ...
        for (idx, acc) in state.accumulators.iter_mut().enumerate() {
            let values = self.group_by_metrics.time_aggregate_arguments(|| {
                self.aggregate_argument_metrics
                    .time(idx, || acc.evaluate_row_aligned_args(batch))
            })?;

            output.extend(accumulator_metrics.time(
                idx,
                AccumulatorPhase::ConvertToState,
                || acc.convert_to_state(&values),
            )?);
        }

So a COUNT(*) query on a high-cardinality GROUP BY degenerates gracefully: the partial stage stops hashing entirely and just streams (group_key, 1) rows downstream, letting the final stage do the single hash pass.

Which stream does the counting?

COUNT(*) participates in whichever stage the plan needs. The stream types are chosen in AggregateExec::execute:

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)?,
            ),
            ...
            (PartialReduce, GroupClusteringMode::None) => StreamType::PartialReduceHash(
                PartialReduceHashAggregateStream::new(self, context, partition)?,
            ),
            (Final | FinalPartitioned, GroupClusteringMode::None) => {
                StreamType::FinalHash(FinalHashAggregateStream::new(
                    self, context, partition,
                )?)
            }
            (Single | SinglePartitioned, GroupClusteringMode::None) => {
                StreamType::SingleHash(SingleHashAggregateStream::new(
                    self, context, partition,
                )?)
            }

PartialHashAggregateStream — raw rows in, partial states out. This is where the skip-aggregation probe lives.
PartialReduceHashAggregateStream — partial states in, partial states out. Used for tree-reduce plans; it merges states instead of accumulating raw rows, so the count’s state is just an Int64 that gets added.
FinalHashAggregateStream — partial states in, final values out.
SingleHashAggregateStream — the whole logical aggregation in one operator, used when the input is a single partition or already hash-partitioned on the group keys.

The PartialReduce stage is the one that most obviously wants counting to be additive: merging two partial counts is a + b, and the merge implementation is a plain sum over an Int64Array:

rust
// datafusion/functions-aggregate/src/count.rs:627-634
    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
        let counts = downcast_value!(states[0], Int64Array);
        let delta = &compute::sum(counts);
        if let Some(d) = delta {
            self.count += *d;
        }
        Ok(())
    }

Trade-offs

The rewrite changes the plan, not the result. COUNT(*), COUNT(1) and COUNT(NULL) are not the same query — COUNT(NULL) is 0 — so the sentinel expansion is only legal for the wildcard form. DataFusion handles this by expanding only COUNT(*)/COUNT() and leaving COUNT(expr) alone.
The literal column costs memory. COUNT(*) over a billion-row batch still materializes an Int64Array of one billion ones (8 GB) for a single batch, unless the batch is small. The AggregateArgument cache keeps it alive across batches (capped at 1 MiB of array), but the first batch still pays for it. For very large batches this is a real cost that a “count rows directly” implementation would not pay.
The branchless advantage disappears with a filter. COUNT(*) FILTER (WHERE p) reintroduces the opt_filter branch, and accumulate_indices falls to the (None, Some(filter)) arm, which iterates the filter in 64-bit chunks (accumulate.rs:605-638) rather than a bare loop.
COUNT(col) is not a sum of ones. A nullable column needs the null-aware path, and CountAccumulator (the non-groups variant) computes array.len() - null_count instead of incrementing per row.
The skip optimization is a heuristic. The defaults are a ratio threshold of 0.8 and a row threshold of 100_000 (datafusion/common/src/config.rs:1133-1137), so small inputs are always aggregated normally; and the probe is only armed for the Partial mode with a single group-by (hash_stream.rs:242-262), and disabled entirely when the ratio threshold is >= 1.0.
Counting is not free at the plan level either. COUNT(*) with a filter or with DISTINCT is a different animal; COUNT(DISTINCT x) has its own bitmap-based accumulators for small integer domains (count.rs:177-281), which is a completely different performance profile from COUNT(*).

The takeaway

COUNT(*) is the aggregate that DataFusion optimizes hardest, and every step of that optimization comes back to the same idea: represent the count as a sum of ones over a non-null constant. The planner makes the argument a non-null literal. The accumulator drops the null machinery and increments a dense Vec<i64> per group. The argument evaluator caches the literal array so the constant is materialized once. The statistics path folds the whole thing into a row count. And when the partial stage decides aggregation is not worth it, it streams (key, 1) rows downstream instead of hashing them.

None of this is exotic. It is the discipline of noticing that “count the non-null values in this column” and “count the rows” are the same query, and then making the compiler’s job as easy as possible.