ClickHouse Aggregates In Order and Stops Reading Early

TL;DR: If your table is sorted by the GROUP BY key, ClickHouse can skip the hash table entirely: it counts each contiguous run of equal keys and emits it. With optimize_aggregation_in_order_limit, it also pushes LIMIT into the aggregation so the read itself stops early.

SELECT count() FROM hits GROUP BY UserID is, conceptually, a hash table build: read every row, hash every key, update a counter. The hash table’s size is the number of distinct keys — a million entries for a column with a million distinct values, a billion entries for a billion distinct values, and then the query dies.

But if the table is sorted by UserID, that is a lot of wasted work. The rows arrive grouped: all the rows with UserID = 1 are contiguous, then all the rows with UserID = 2, and so on.

You don’t need a hash table at all. You need to count the run, emit it, count the next run, emit it.

That is the optimize_aggregation_in_order optimisation. It turns an unbounded-memory operation into a streaming one whose memory footprint is the size of a single group. And in recent versions it also enables an even more aggressive trick: pushing the LIMIT down into the aggregation so the read stops early.

Let’s walk the machinery: the plan-level decision, the run-by-run aggregation loop, and the limit pushdown with its subtle “prefix boundary” rule.

When the optimisation applies

The decision is made in AggregatingStep::transformPipeline, and it is based on a single field: sort_description_for_merging. If that is non-empty, the input is already sorted on the group-by key and the in-order path is taken.

cpp
// src/Processors/QueryPlan/AggregatingStep.cpp:505
    /// optimize_aggregation_in_order
    if (!sort_description_for_merging.empty())
    {
        /// two-level aggregation is not supported anyway for in order aggregation.
        allow_to_use_two_level_group_by = false;

        /// It is incorrect for in order aggregation.
        params.stats_collecting_params.disable();

        /// Aggregation in order rebuilds the aggregation-method state for every run of equal
        /// order-key values, so the whole-block `prealloc_serialized` method would make it
        /// quadratic. Fall back to the plain `serialized` method (see `Params::aggregation_in_order`).
        params.aggregation_in_order = true;
    }

Three consequences fall out of this one branch, and each is a real limitation.

Two-level aggregation is disabled. The classic ClickHouse optimisation for GROUP BY with many distinct keys is two-level aggregation: hash the key, bucket by hash, aggregate each bucket in a separate hash table, and merge the buckets in parallel. That is exactly what you do not want when the data is already grouped — the whole point is to avoid a hash table entirely.

Statistics collection is disabled. The uniqCombined / uniqExact family of aggregate functions maintains auxiliary state to compute statistics that are only meaningful for the whole dataset. In-order aggregation finalises each group as it goes, so those statistics would be wrong.

The prealloc_serialized aggregation method is downgraded. This one deserves an explanation, because it is a performance pathology that the comment records:

cpp
// src/Interpreters/Aggregator.cpp:849
    /// See `Params::aggregation_in_order` and `method_chosen_for_in_order`: the `prealloc_serialized`
    /// method serializes the whole block's keys on state construction, which is pathological for the
    /// per-run in-order path, where a fresh state is constructed for every run of equal order-key
    /// values. There it uses the plain `serialized` method, whose construction is O(1) and which
    /// serializes keys lazily. The whole-block paths (including `mergeBlocks`) keep `method_chosen`.
    method_chosen_for_in_order = method_chosen;
    if (params.aggregation_in_order)
    {
        if (method_chosen_for_in_order == AggregatedDataVariants::Type::prealloc_serialized)
            method_chosen_for_in_order = AggregatedDataVariants::Type::serialized;

prealloc_serialized is an optimisation for the hash table case: when you create a state for a key, you serialise the key into a contiguous buffer right away, so the key data is contiguous and the hash table lookup is cheap. In the in-order case a fresh state is created for every run of equal keys — potentially millions of times for a million groups — and serialising the whole key each time would be quadratic in the key width. The lazy serialized variant amortises it.

The pipeline shape

When the in-order path is taken and there is more than one stream, the pipeline looks like this:

text
 --> AggregatingInOrder --> FinishAggregatingInOrder --> Resize --> MergingAggregatedBucket
 --> AggregatingInOrder --> FinishAggregatingInOrder --> Resize --> MergingAggregatedBucket
 --> AggregatingInOrder --> FinishAggregatingInOrder --> Resize --> MergingAggregatedBucket

as the comment in AggregatingStep says:

cpp
// src/Processors/QueryPlan/AggregatingStep.cpp:678
            /** The pipeline is the following:
             *
             * --> AggregatingInOrder                                                  --> MergingAggregatedBucket
             * --> AggregatingInOrder --> FinishAggregatingInOrder --> ResizeProcessor --> MergingAggregatedBucket
             * --> AggregatingInOrder                                                  --> MergingAggregatedBucket
             */

Each stream runs its own AggregatingInOrderTransform, each with its own ManyAggregatedData variant, and the partial results are merged later. The transform is constructed per stream:

cpp
// src/Processors/QueryPlan/AggregatingStep.cpp:692
                return std::make_shared<AggregatingInOrderTransform>(
                    header,
                    transform_params,
                    sort_description_for_merging,
                    group_by_sort_description,
                    max_block_size,
                    aggregation_in_order_max_block_bytes / new_merge_threads,
                    many_data,
                    counter++,
                    limit_hint,
                    limit_hint_prefix_columns,

The counter++ is the stream index — each transform gets its own slot in the shared ManyAggregatedData, so the later merge knows which partial state belongs to which stream.

The core loop: aggregating one run at a time

AggregatingInOrderTransform::consume() is the heart of it. The key observation is that the input is sorted, so the rows belonging to one group form a contiguous interval, and the interval can be found with a binary search.

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:183
    size_t interval_index = 0;

    /// Will split block into segments with the same key
    while (key_end != rows)
    {
        /// Cancellation is only checked between work() calls, but one consume() over a chunk with many
        /// keys can run for a long time; check per key interval so a cancelled query stops promptly.
        if (isCancelled())
        {
            LOG_TEST(log, "Cancelled between key intervals");
            return;
        }

and the search itself:

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:208
        /// Find the first position of new (not current) key in current chunk
        auto indices = collections::range(key_begin, rows);
        auto it = std::upper_bound(indices.begin(), indices.end(), cur_block_size - 1,
            [&](size_t lhs_row, size_t rhs_row)
            {
                return less(res_key_columns, key_columns, lhs_row, rhs_row, group_by_description);
            });

        key_end = (it == indices.end() ? rows : *it);

        /// Add data to aggr. state if interval is not empty. Empty when haven't found current key in new block.
        if (key_begin != key_end)
        {
            if (params->params.only_merge)
            {
                if (group_by_key)
                    params->aggregator.mergeOnBlockSmall(variants, key_begin, key_end, aggregate_columns_data, key_columns_raw);
                else
                    params->aggregator.mergeOnIntervalWithoutKey(variants, key_begin, key_end, aggregate_columns_data, is_cancelled);
            }
            else
            {
                if (group_by_key)
                    params->aggregator.executeOnBlockSmall(variants, key_begin, key_end, key_columns_raw, aggregate_function_instructions.data());
                else
                    params->aggregator.executeOnIntervalWithoutKey(variants, key_begin, key_end, aggregate_function_instructions.data());
            }
        }

res_key_columns holds the current key (one row), key_columns holds the whole chunk’s key column. upper_bound with the less comparator finds the first row whose key is strictly greater than the current key. Everything between key_begin and key_end belongs to the current group.

less() is the generic column-wise comparison:

cpp
// src/Core/SortCursor.h:969
template <typename TLeftColumns, typename TRightColumns>
bool less(const TLeftColumns & lhs, const TRightColumns & rhs, size_t i, size_t j, const SortDescriptionWithPositions & descr)
{
    for (const auto & elem : descr)
    {
        size_t ind = elem.column_number;
        int res = elem.base.direction * lhs[ind]->compareAt(i, j, *rhs[ind], elem.base.nulls_direction);
        if (res < 0)
            return true;
        if (res > 0)
            return false;
    }

    return false;
}

So the loop is: binary-search the end of the current run, aggregate the run into the current state, finalise the state into the output columns, record the next key, repeat.

Two modes: with a key and without

group_by_key selects between two quite different implementations. The header comment on the aggregator methods explains the split:

cpp
// src/Interpreters/Aggregator.h:553
    /// Used for optimize_aggregation_in_order:
    /// - No two-level aggregation
    /// - No external aggregation
    /// - No without_key support (it is implemented using executeOnIntervalWithoutKey())
    void executeOnBlockSmall(
        AggregatedDataVariants & result,
        size_t row_begin,
        size_t row_end,
        ColumnRawPtrs & key_columns,
        AggregateFunctionInstruction * aggregate_instructions) const;

    void executeOnIntervalWithoutKey(
        AggregatedDataVariants & data_variants,
        size_t row_begin,
        size_t row_end,
        AggregateFunctionInstruction * aggregate_instructions) const;

group_by_key is true when the GROUP BY key is a prefix of the sort description — i.e. the input is sorted by exactly the grouping key. In that case there is only ever one group at a time, so no hash table is needed at all: the aggregate state is a single object that is created, filled, and finalised per run.

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:147
    /// If we don't have a block we create it and fill with first key
    if (!cur_block_size)
    {
        res_key_columns.resize(params->params.keys_size);
        for (size_t i = 0; i < params->params.keys_size; ++i)
            res_key_columns[i] = res_header.safeGetByPosition(i).type->createColumn();

        if (group_by_key)
            params->aggregator.fillKeyColumnsWithSingleKey(key_columns, key_begin, res_key_columns);
        else
            params->aggregator.createStatesAndFillKeyColumnsWithSingleKey(variants, key_columns, key_begin, res_key_columns);

fillKeyColumnsWithSingleKey — the state is not created; the key is just copied into the output column. createStatesAndFillKeyColumnsWithSingleKey — a state is created in the hash table, because the grouping key is a prefix of the sort key but not the whole of it, so several distinct groups may share the same sort-key prefix and must be kept apart.

When the group is finished, the state is written out and the key advances:

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:239
        /// We finalize last key aggregation state if a new key found.
        if (key_end != rows)
        {
            if (!group_by_key)
                params->aggregator.addSingleKeyToAggregateColumns(variants, res_aggregate_columns);

            /// Enough groups have been emitted and the new key starts a new value of the
            /// `ORDER BY` prefix, so the rest of the input cannot be needed (see `limit_prefix_columns`).
            /// With `group_by_key`, `cur_block_size` counts runs of the sorted key columns rather than
            /// groups; the groups accumulated for the current block are the entries of the hash table.
            size_t cur_block_groups = group_by_key ? variants.size() : cur_block_size;
            if (limit_hint && cur_block_groups + res_rows >= limit_hint && isLimitPrefixBoundary(key_columns, key_end))
                limit_reached = true;

And the “the rest of the chunk is not needed” case:

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:253
            /// If max_block_size or limit_hint is reached we have to stop consuming and generate the block. Save the extra rows into new chunk.
            if (cur_block_size >= max_block_size || cur_block_bytes + current_memory_usage >= max_block_bytes || limit_reached)
            {
                if (group_by_key)
                    group_by_chunk
                        = params->aggregator.prepareChunkAndFillSingleLevel</* return_single_block */ true>(variants, /* final= */ false).chunk;
                cur_block_bytes += current_memory_usage;

                /// When limit is reached, don't save leftover rows — we're done.
                /// Set block_end_reached + need_generate to trigger generate(),
                /// which will produce the output and then set is_consume_finished.
                if (limit_reached)
                {
                    block_end_reached = true;
                    need_generate = true;
                    variants.invalidate();
                }
                else
                {
                    finalizeCurrentChunk(std::move(chunk), key_end);
                }

Note the difference between the two branches. Normally the leftover rows of the chunk (those after key_end) are saved — finalizeCurrentChunk cuts them off and keeps them for the next consume():

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:298
void AggregatingInOrderTransform::finalizeCurrentChunk(Chunk chunk, size_t key_end)
{
    size_t rows = chunk.getNumRows();
    Columns source_columns = chunk.detachColumns();

    for (auto & source_column : source_columns)
        source_column = source_column->cut(key_end, rows - key_end);

    current_chunk = Chunk(source_columns, rows - key_end);
    src_rows -= current_chunk.getNumRows();

    block_end_reached = true;
    need_generate = true;
    variants.invalidate();
}

But when the limit is reached, the leftovers are dropped: the query is over, and reading the rest of the chunk would be wasted work.

Pushing `LIMIT` into the aggregation

The limit pushdown is a plan-level optimisation, in src/Processors/QueryPlan/Optimizations/optimizeLimitForAggregationInOrder.cpp, and it is gated on the optimize_aggregation_in_order_limit setting, which defaults to true:

text

// src/Core/Settings.cpp:5163
    DECLARE(Bool, optimize_aggregation_in_order_limit, true, R"(
When enabled and aggregation in order is active, pushes LIMIT into the aggregation step to enable early termination after producing enough groups. This reduces the amount of data read when ORDER BY matches the GROUP BY key prefix. May reduce the value reported by `rows_before_limit_at_least`; use `exact_rows_before_limit` if exact counts are needed.
)", 0) \

The optimisation looks for the shape LIMIT → SORT → AGGREGATE:

cpp
// src/Processors/QueryPlan/Optimizations/optimizeLimitForAggregationInOrder.cpp:77
void optimizeLimitForAggregationInOrder(QueryPlan::Node & root)
{
    Stack stack;
    stack.push_back({.node = &root});

    while (!stack.empty())
    {
        auto & frame = stack.back();

        if (frame.next_child < frame.node->children.size())
        {
            stack.push_back({.node = frame.node->children[frame.next_child++]});
            continue;
        }

        auto * node = frame.node;
        stack.pop_back();

        auto * limit_step = typeid_cast<LimitStep *>(node->step.get());
        if (!limit_step || limit_step->withTies() || limit_step->alwaysReadTillEnd())
            continue;

        size_t limit = limit_step->getLimitForSorting();
        if (!limit)
            continue;

WITH TIES and always_read_till_end are excluded immediately — both change the meaning of the limit in ways the pushdown cannot honour.

Then it walks down through “transparent” steps (steps that neither change the row count nor the sort order) looking for the aggregation:

cpp
// src/Processors/QueryPlan/Optimizations/optimizeLimitForAggregationInOrder.cpp:117
        /// Find AggregatingStep below SortingStep, skipping transparent steps.
        /// Stop at TotalsHavingStep — HAVING may filter groups.
        while (current)
        {
            auto * aggregating_step = typeid_cast<AggregatingStep *>(current->step.get());
            if (aggregating_step)
            {
                /// overflow_row is set by WITH TOTALS.
                if (!aggregating_step->inOrder()
                    || !aggregating_step->getFinal()
                    || aggregating_step->getParams().overflow_row
                    || aggregating_step->isGroupingSets())
                    break;

and then the crucial check:

cpp
// src/Processors/QueryPlan/Optimizations/optimizeLimitForAggregationInOrder.cpp:131
                /// The sort description may be a strict prefix of the group-by sort description.
                /// Each in-order stream then stops only at a boundary of that prefix (see
                /// `AggregatingInOrderTransform`), so the groups it drops sort strictly after
                /// at least `limit` complete groups and the final sort can never pick a group
                /// with a partial aggregate value (issue #116849).
                const auto & sort_desc = sorting_step->getSortDescription();
                const auto & agg_sort_desc = aggregating_step->getGroupBySortDescription();
                if (sort_desc.empty() || !agg_sort_desc.hasPrefix(sort_desc))
                    break;

                /// Use the smallest limit if multiple LimitSteps point to the same AggregatingStep.
                size_t current_hint = aggregating_step->getLimitHint();
                if (!current_hint || limit < current_hint)
                    aggregating_step->setLimitHint(limit, sort_desc.size());
                break;

agg_sort_desc.hasPrefix(sort_desc) — the ORDER BY columns must be a prefix of the GROUP BY sort description. This is the condition that makes the pushdown safe, and the reasoning in the comment is worth unpacking because it is subtle.

The prefix boundary rule

The ORDER BY of the query may be a prefix of the GROUP BY key. Say:

sql
SELECT a, b, count()
FROM t
GROUP BY a, b
ORDER BY a
LIMIT 10

Here the table is sorted by (a, b). The GROUP BY produces groups in (a, b) order. The ORDER BY a then sorts them by a — and since the groups already come out in (a, b) order, sorting by a is a stable reordering: groups with the same a keep their relative (a, b) order.

Now: the aggregation stream can stop early once it has produced 10 groups. But it must not stop in the middle of a run of groups that share the same a — because another stream might produce a group with the same a but a smaller b, and the final sort (by a only) could then pick that group over the partial one.

The rule is: stop only at a boundary of the ORDER BY prefix. That is what limit_prefix_columns counts and what isLimitPrefixBoundary() checks:

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:85
bool AggregatingInOrderTransform::isLimitPrefixBoundary(const Columns & key_columns, size_t row) const
{
    for (size_t i = 0; i < limit_prefix_columns; ++i)
    {
        const auto & elem = group_by_description[i];
        size_t ind = elem.column_number;
        if (res_key_columns[ind]->compareAt(cur_block_size - 1, row, *key_columns[ind], elem.base.nulls_direction) != 0)
            return true;
    }
    return false;
}

It compares the last emitted group’s value of the prefix column with the next group’s value. If they differ, the run of equal prefix values has ended, and it is safe to stop: every group with that prefix value has been emitted.

Note the comment on the member:

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.h:97
    /// The stream may stop early once it has emitted `limit_hint` groups, but only at a
    /// boundary of the first `limit_prefix_columns` columns of `group_by_description`, which
    /// are the columns the query is ordered by. Otherwise a group tied on those columns with
    /// the emitted ones could be cut off here while another stream emits its partial state,
    /// and the final sort could pick that partial group.
    size_t limit_hint = 0;
    size_t limit_prefix_columns = 0;
    bool limit_reached = false;

And the prefix count is clamped to the group-by description:

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:80
    /// The `ORDER BY` may extend past the columns the input is sorted by; a boundary of the
    /// sorted columns alone is then a boundary of the `ORDER BY` prefix as well.
    limit_prefix_columns = std::min(limit_prefix_columns_, group_by_description.size());

This is the bug that issue #116849 was about, and the comment in the optimisation pass records it.

What it buys, concretely

sql
SELECT user_id, count()
FROM hits
GROUP BY user_id
ORDER BY user_id
LIMIT 10

on a table sorted by user_id. Without the pushdown, the whole table is read, aggregated, sorted, and then 10 rows are kept. With it, the aggregation stops after the first 10 groups, and — crucially — the source stops too, because the transform closes its input:

cpp
// src/Processors/Transforms/AggregatingInOrderTransform.cpp:354
        /// When limit is reached, we set is_consume_finished in generate()
        /// and don't save leftover rows. Finish immediately after pushing.
        if (is_consume_finished)
        {
            output.finish();
            input.close();
            LOG_DEBUG(log, "Aggregated. {} to {} rows (from {})", src_rows, res_rows, formatReadableSizeWithBinarySuffix(src_bytes));
            return Status::Finished;
        }

input.close() propagates the stop all the way up to the storage layer. On a MergeTree, that means the remaining granules are never read.

The cost is stated in the setting’s description: rows_before_limit_at_least becomes an underestimate, because the rows that were never read are not counted. exact_rows_before_limit restores the exact count at the price of reading everything.

Trade-offs

What this buys you.

The in-order aggregation replaces a hash table whose size is the number of distinct keys with a state whose size is one group. For a table sorted by the grouping key, that is the difference between a query that fits in memory and one that does not.

It also composes with the parallel execution model: each stream aggregates its own runs independently, and the partial states are merged at the end by FinishAggregatingInOrderTransform and MergingAggregatedBucketTransform. The parallelism is preserved; only the memory behaviour changes.

The limit pushdown extends the idea from “less memory” to “less work”: the read itself stops. On a large MergeTree with a selective LIMIT, this is the difference between reading 10 granules and reading a thousand.

What it costs.

The optimisation is conditional on the data layout. If the table is not sorted by the grouping key, the plan does not use it — and if the table is sorted but the GROUP BY key is not a prefix of the sort key, it does not apply either. ClickHouse has to prove the sortedness, and it does so from the table’s primary key and the optimize_read_in_order machinery, so the whole optimisation is contingent on the optimiser’s knowledge of the physical layout.

The in-order path gives up several general-purpose optimisations: two-level aggregation, external aggregation, and statistics collection. For a query with a small number of distinct keys on a large table, the ordinary hash-table path with two-level aggregation might actually be faster, because it parallelises the hash table itself rather than relying on the input being pre-grouped. The optimize_aggregation_in_order setting defaults to false for exactly this reason.

The prealloc_serialized → serialized downgrade is a real regression in the hash-table case that the in-order case forced, paid for by every query that takes the in-order path.

The limit pushdown has a correctness-adjacent cost that the code is careful about but that is easy to get wrong: the prefix-boundary rule. Stop at the wrong place and the final sort may pick a group whose aggregate state was truncated — a wrong answer, not a slow one. That is why the rule is conservative (stop only at a prefix boundary, never in the middle of a run of equal prefix values) and why the plan-level check requires the ORDER BY to be a prefix of the group-by description.

And the pushdown changes the meaning of a system counter: rows_before_limit_at_least becomes “rows that were actually read”, not “rows that would have been read without the limit”. Any monitoring built on that counter has to know about the optimisation.

Wrapping up

The one thing to remember: sorting your table by the GROUP BY key is not just about skipping data — it changes the shape of the aggregation. A hash table the size of your cardinality becomes a state the size of one group.

The part people get wrong: the LIMIT pushdown only fires when ORDER BY is a prefix of the GROUP BY key. If it isn’t, you get the memory benefit but not the early-termination benefit — and the read still scans the whole table.

What to go do about it: check whether your table’s ORDER BY matches your most common GROUP BY. If it doesn’t, you are leaving one of the largest optimisations in the engine on the table.