ClickHouse Splits GROUP BY into 256 Buckets for Parallel Merges

TL;DR: A GROUP BY needs a serial merge at the end — and that’s where the memory high-water mark lives. ClickHouse splits the key space into 256 buckets by the top bits of the hash, so the merge becomes 256 independent, lock-free, parallel jobs.

A GROUP BY in ClickHouse is usually computed by several threads at once, each with its own hash table, and the tables are then merged.

Merging N hash tables of M entries each is not something you want to do on one thread. It’s a serial phase in an otherwise parallel query, and it’s the phase where the memory high-water mark is reached.

The way out is to make the merge divisible. If the keys are partitioned into a fixed number of buckets by a deterministic function of the key, then two tables can be merged bucket by bucket, and different buckets can be merged by different threads with no coordination at all: bucket 17 of table A can only collide with bucket 17 of table B.

That’s what TwoLevelHashTable in src/Common/HashTable/TwoLevelHashTable.h does, and it’s what the group_by_two_level_threshold setting controls.

The structure

The two-level table is literally an array of ordinary hash tables:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:9
/** Two-level hash table.
  * Represents 256 (or 1ULL << BITS_FOR_BUCKET) small hash tables (buckets of the first level).
  * To determine which one to use, one of the bytes of the hash function is taken.
  *
  * Usually works a little slower than a simple hash table.
  * However, it has advantages in some cases:
  * - if you need to merge two hash tables together, then you can easily parallelize it by buckets;
  * - delay during resizes is amortized, since the small hash tables will be resized separately;
  * - in theory, resizes are cache-local in a larger range of sizes.
  *
  * With `BITS_FOR_BUCKET = 0` there is a single bucket: routing folds to a constant.
  * Lookups and inserts compile down to what the single-level table does.
  *
  * Buckets share no state.
  * Threads may fill different buckets at the same time when every bucket is written under its own lock.
  */

The bucket index comes out of the hash:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:58
    static constexpr UInt32 NUM_BUCKETS = 1ULL << BITS_FOR_BUCKET;
    static constexpr UInt32 MAX_BUCKET = NUM_BUCKETS - 1;

    size_t hash(const Key & x) const { return Hash::operator()(x); }

    /// NOTE Bad for hash tables with more than 2^32 cells.
    static size_t getBucketFromHash(size_t hash_value) { return (hash_value >> (32 - BITS_FOR_BUCKET)) & MAX_BUCKET; }

BITS_FOR_BUCKET defaults to 8, so 256 buckets, and the bucket is taken from the high bytes of the 64-bit hash:

text
        64-bit hash
        |<-------------------------->|
        |  bucket   |   sub-table index   |
        | 8 bits    |  (mask & (size-1)) |
        +-----------+---------------------+
        |  0..255   |  depends on the bucket's own size |
        +-----------+---------------------+

   hash >> 24  & 255  ->  which of the 256 sub-tables
   hash & (2^k - 1)    ->  which cell inside that sub-table

The choice of the high bits is not arbitrary, and it’s the subtle part of the whole design. The single-level table uses the low bits of the hash to index its cells:

cpp
// src/Common/HashTable/HashTable.h:238
    /// From the hash value, get the cell number in the hash table.
    size_t place(size_t x) const         { return x & mask(); }

If the bucket were also selected from the low bits, the two uses would collide. Every bucket would receive keys whose low bits are a fixed value, and the sub-table’s own probe would be confined to a single cell. The sub-tables would degenerate into 256 linked lists.

Taking the bucket from the top of the hash and the cell from the bottom keeps the two independent — which is exactly what a good hash function provides.

The default bucket count is 8 bits, and the comment “Bad for hash tables with more than 2^32 cells” is about the other direction: the shift hash_value >> (32 - BITS_FOR_BUCKET) throws away everything above bit 32, so the bucket selection only sees 32 bits of the hash. For a table with more than 2^32 cells, the top bits would carry real information and throwing them away would lose it.

The sub-tables are ordinary tables

Each bucket is a plain HashTable, so all the machinery from the previous article — zeroed cells, linear probing, Robin Hood ordering — applies inside a bucket. The two-level table only adds the routing:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:316
    template <typename KeyHolder>
    void ALWAYS_INLINE emplace(KeyHolder && key_holder, LookupResult & it, bool & inserted, size_t hash_value)
    {
        size_t buck = getBucketFromHash(hash_value);
        impls[buck].emplace(key_holder, it, inserted, hash_value);
    }

cpp
// src/Common/HashTable/TwoLevelHashTable.h:333
    LookupResult ALWAYS_INLINE find(Key x, size_t hash_value)
    {
        size_t buck = getBucketFromHash(hash_value);
        return impls[buck].find(x, hash_value);
    }

and the storage is a plain array of sub-tables:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:100
    Impl impls[NUM_BUCKETS];

    /// `bucket_cells_prefix[b]` is the number of cells in the buckets before `b`. See `computeBucketPrefix`.
    std::vector<size_t> bucket_cells_prefix;

Impl impls[NUM_BUCKETS] is 256 sub-tables by value, not by pointer. Each sub-table owns its buffer, so the whole structure is 256 small allocations plus the bookkeeping of each.

How a bucket grows

The buckets do grow, but they grow in a different shape from the single-level table. The grower used by the two-level table caps the quadratic-growth phase at 2^15 cells and switches to the Robin Hood mode above that:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:26
/// The buckets grow by 4 times while they are small, and switch to the Robin Hood mode when they become large,
/// see `HashTableGrowerWithPrecalculationAndRobinHood`.
template <size_t initial_size_degree = 8, size_t robin_hood_size_degree = 16>
using TwoLevelHashTableGrower = HashTableGrowerWithPrecalculationAndRobinHood<initial_size_degree, 15, robin_hood_size_degree>;

The template parameters are initial_size_degree = 8, max_size_degree_quadratic = 15, robin_hood_size_degree = 16. So:

up to 2^15 cells per bucket, the sub-table quadruples its size on overflow (a degree increment of two);
from 2^15 to 2^16 it doubles;
above 2^16 cells per bucket it switches to the Robin Hood mode with a 90% load factor, and from then on it keeps doubling like any other table — but at 90% load instead of 50%.

The effect of the switch is memory, not capacity. A bucket that grows large stops paying the factor-of-two penalty of the 0.5 load factor, because the same number of keys now needs half as many cells.

That’s also what makes the external-memory path possible: when the aggregation spills to disk, it spills bucket by bucket, and each bucket’s data is self-contained.

Resizing is amortised

A single-level table resizes by re-inserting every element. At the moment of the resize, the whole table has to be walked and every cell re-placed.

In a two-level table, each bucket resizes on its own, and since the buckets are filled roughly uniformly, the individual resizes happen at different times. The total work is the same, but it’s spread out, and — crucially — the pause is per-bucket rather than table-wide.

The comment in the header says it exactly:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:14
  * - delay during resizes is amortized, since the small hash tables will be resized separately;

When is the two-level table used?

The decision is a setting, and it’s a pair of settings now — one counting keys, one counting bytes:

cpp
// src/Core/Settings.cpp:1500
    DECLARE(UInt64, group_by_two_level_threshold, 100000, R"(
From what number of keys, a two-level aggregation starts. 0 - the threshold is not set.
)", 0) \
    DECLARE(UInt64, group_by_two_level_threshold_bytes, 50000000, R"(
From what size of the aggregation state in bytes, a two-level aggregation begins to be used. 0 - the threshold is not set. Two-level aggregation is used when at least one of the thresholds is triggered.
)", 0) \

So by default a GROUP BY switches to the two-level structure once it holds 100 000 keys, or once its state grows to 50 MB — whichever happens first. Both thresholds can be forced from SQL:

sql
SELECT user_id, count()
FROM events
GROUP BY user_id
SETTINGS group_by_two_level_threshold = 1000000,
         group_by_two_level_threshold_bytes = 50000000;

Setting either threshold to 0 disables that trigger — which is exactly what the pipeline does when a two-level aggregation would be useless.

The predicate that implements this lives in src/Interpreters/Aggregator.cpp:

cpp
// src/Interpreters/Aggregator.cpp:118
bool worthConvertToTwoLevel(
    size_t group_by_two_level_threshold, size_t result_size, size_t group_by_two_level_threshold_bytes, auto result_size_bytes)
{
    // params.group_by_two_level_threshold will be equal to 0 if we have only one thread to execute aggregation (refer to AggregatingStep::transformPipeline).
    return (group_by_two_level_threshold && result_size >= group_by_two_level_threshold)
        || (group_by_two_level_threshold_bytes && result_size_bytes >= static_cast<Int64>(group_by_two_level_threshold_bytes));
}

Note the && in each disjunct: a threshold of 0 disables that particular trigger, which is why the settings can be turned off individually.

The conversion itself happens during the aggregation, when a block has been added and the table is re-examined:

cpp
// src/Interpreters/Aggregator.cpp:2668
    bool worth_convert_to_two_level = worthConvertToTwoLevel(
        params.group_by_two_level_threshold, result_size, params.group_by_two_level_threshold_bytes, result_size_bytes);

    /** Converting to a two-level data structure.
      * It allows you to make, in the subsequent, an effective merge - either economical from memory or parallel.
      */
    if (result.isConvertibleToTwoLevel() && worth_convert_to_two_level)
        result.convertToTwoLevel();

and the conversion itself rebuilds the variant with the two-level type:

cpp
// src/Interpreters/AggregatedDataVariants.cpp:263
void AggregatedDataVariants::convertToTwoLevel()
{
    ProfileEvents::increment(ProfileEvents::AggregationConvertedToTwoLevel);

    if (aggregator)
        LOG_TRACE(aggregator->log, "Converting aggregation data to two-level.");

    switch (type)
    {
#define M(NAME) \
    case Type::NAME: \
        NAME ## _two_level = std::make_unique<decltype(NAME ## _two_level)::element_type>(*(NAME)); \
        (NAME ## _two_level)->top_k_heap = std::move((NAME)->top_k_heap); \
        (NAME).reset(); \
        type = Type::NAME ## _two_level; \
        break;

    APPLY_FOR_VARIANTS_CONVERTIBLE_TO_TWO_LEVEL(M)

#undef M

    default:
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong data variant passed.");
    }
}

The TwoLevelHashTable has a constructor that takes a single-level table and distributes it, which is what makes the conversion a one-liner:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:116
    template <typename Source>
    requires(!std::is_arithmetic_v<Source>)
    explicit TwoLevelHashTable(const Source & src)
    {
        typename Source::const_iterator it = src.begin();

        /// It is assumed that the zero key (stored separately) is first in iteration order.
        if (it != src.end() && it.getPtr()->isZero(src))
        {
            insert(it->getValue());
            ++it;
        }

        for (; it != src.end(); ++it)
        {
            const Cell * cell = it.getPtr();
            size_t hash_value = cell->getHash(src);
            size_t buck = getBucketFromHash(hash_value);
            impls[buck].insertUniqueNonZero(cell, hash_value);
        }
    }

The zero key is handled explicitly, because it lives outside the buffer and doesn’t follow the bucket rule — it goes to bucket 0.

The merge is the point

Now the reason the whole thing exists. The aggregation pipeline decides whether a two-level aggregation is even allowed:

cpp
// src/Processors/QueryPlan/AggregatingStep.cpp:503
    bool allow_to_use_two_level_group_by = pipeline.getNumStreams() > 1 || params.max_bytes_before_external_group_by != 0;

Two conditions: either there’s more than one stream (so the merge is worth parallelising), or the query is allowed to spill to disk (so the bucket structure is needed to spill incrementally). If neither holds, the thresholds are zeroed out and the aggregation stays single-level:

cpp
// src/Processors/QueryPlan/AggregatingStep.cpp:520
    if (!allow_to_use_two_level_group_by)
    {
        params.group_by_two_level_threshold = 0;
        params.group_by_two_level_threshold_bytes = 0;
    }

When the two-level table is used, the merge stage is a source that hands out one bucket at a time, and the pipeline runs several of them in parallel:

cpp
// src/Processors/Transforms/AggregatingTransform.cpp:208
/// Worker which merges buckets for two-level aggregation.
/// Atomically increments bucket counter and returns merged result.
class ConvertingAggregatedToChunksWithMergingSource final : public ISource
{
public:
    static constexpr UInt32 NUM_BUCKETS = 256;

The worker takes the next bucket with an atomic counter and merges it across all the per-thread tables:

cpp
// src/Processors/Transforms/AggregatingTransform.cpp:272
    Chunk generate() override
    {
        UInt32 bucket_num = shared_data->next_bucket_to_merge.fetch_add(1);

        if (bucket_num >= NUM_BUCKETS)
        {
            data.reset();
            return {};
        }

The atomic counter is the entire coordination mechanism. No lock, no work queue, no barrier. A worker that finishes bucket 3 immediately grabs bucket 4, and the workers end up balanced by the cost of the buckets they happen to pick up. Since the buckets partition the key space, no two workers ever touch the same key.

The actual merge of one bucket, in Aggregator::mergeBucketImpl, is “for each thread’s table, merge its bucket into the first thread’s bucket”:

cpp
// src/Interpreters/Aggregator.cpp:5200
void NO_INLINE Aggregator::mergeBucketImpl(
    ManyAggregatedDataVariants & data, Int32 bucket, Arena * arena, std::atomic<bool> & is_cancelled) const
{
    /// We merge all aggregation results to the first.
    AggregatedDataVariantsPtr & res = data[0];

cpp
// src/Interpreters/Aggregator.cpp:5213
    auto & dst = getDataVariant<Method>(*res).data.impls[bucket];

Note that the bucket is an index into the impls[] array of the two-level table — the same array the routing used. That’s the whole contract: the bucket is a function of the key, therefore the bucket is a partition of the key space.

The same trick appears in the uniqExact aggregate function, which keeps a single-level set and converts it to a two-level one when it grows:

cpp
// src/AggregateFunctions/UniqExactSet.h:386
    static bool worthConvertingToTwoLevel(size_t size) { return size > 100'000; }

and merges it bucket-wise in parallel:

cpp
// src/AggregateFunctions/UniqExactSet.h:225
            auto thread_func = [&two_level_ptrs, &first_two_level, next_bucket_to_merge, &is_cancelled]()
            {
                while (true)
                {
                    if (is_cancelled.load(std::memory_order_seq_cst))
                        return;

                    const auto bucket = next_bucket_to_merge->fetch_add(1);
                    if (bucket >= NUM_BUCKETS)
                        return;

                    for (size_t j = 1; j < two_level_ptrs.size(); ++j)
                    {
                        if (is_cancelled.load(std::memory_order_seq_cst))
                            return;

                        first_two_level.impls[bucket].merge(two_level_ptrs[j]->impls[bucket]);
                    }
                }
            };

with the same bounded-threads trick:

cpp
// src/AggregateFunctions/UniqExactSet.h:246
            const size_t max_threads_to_enqueue = std::min<size_t>(thread_pool.getMaxThreads(), NUM_BUCKETS);

There’s no point in queueing more workers than there are buckets.

The string flavour

Strings get their own two-level table, because the string table isn’t a HashTable but a composite of five sub-tables (one per key size class). The routing therefore has to pick both the bucket and the size class:

cpp
// src/Common/HashTable/TwoLevelStringHashTable.h:77
    /// The bucket-aware size-class dispatch: the shared implementation in `StringHashTable`
    /// classifies and packs the key and computes its hash; the resolver here picks the bucket
    /// from that hash and hands out the bucket's submap of the right class.
    template <typename Self, typename Func, typename KeyHolder>
    static auto ALWAYS_INLINE dispatch(Self & self, KeyHolder && key_holder, Func && func)
    {
        return Impl::dispatchOnKeyClass(
            [&](auto key_class, size_t hash_value) -> auto &
            { return self.impls[getBucketFromHash(hash_value)].template submapForClass<decltype(key_class)::value>(); },
            std::forward<KeyHolder>(key_holder),
            StringHashTableHash{},
            std::forward<Func>(func));
    }

256 buckets times 5 size classes is 1280 sub-tables. That’s a lot of bookkeeping for a table that usually isn’t huge, which is why the string tables are only converted to two-level when the aggregation really needs it.

Offsets across buckets

One thing the two-level table has to solve is numbering the cells globally. The aggregation code refers to cells by an offset into the whole table, and with 256 separate buffers there’s no single array to index into.

The solution is a prefix sum over the buckets:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:435
    /// Prefix sums that `offsetInternal` uses to number cells across all buckets.
    /// Call it after the last insert and before reading offsets, never while another thread reads them.
    void computeBucketPrefix()
    {
        bucket_cells_prefix.assign(NUM_BUCKETS, 0);
        size_t run = 0;
        for (UInt32 i = 0; i < NUM_BUCKETS; ++i)
        {
            bucket_cells_prefix[i] = run;
            run += impls[i].getBufferSizeInCells();
        }
    }

and the offset itself:

cpp
// src/Common/HashTable/TwoLevelHashTable.h:448
    /// Number of the cell over all buckets. The zero cell gets 0. Any other cell gets its position in the
    /// concatenated bucket buffers plus one. So the number fits an array of `getBufferSizeInCells() + 1`.
    size_t offsetInternal(ConstLookupResult ptr) const
    {
        if constexpr (NUM_BUCKETS == 1)
            return impls[0].offsetInternal(ptr);

        const size_t buck = getBucketFromHash(ptr->getHash(*this));
        if (ptr->isZero(impls[buck]))
            return 0;

        chassert(!bucket_cells_prefix.empty(), "computeBucketPrefix must run before an offset is read");
        return bucket_cells_prefix[buck] + (ptr - impls[buck].buf) + 1;
    }

The “+1” is the zero key’s slot, which is numbered 0.

Trade-offs

What the two-level table buys:

A parallelisable merge. The merge of N per-thread tables becomes 256 independent merges with an atomic counter and no locks. The serial tail of a parallel GROUP BY mostly disappears.
Incremental spilling. Because a bucket is self-contained, the aggregation can write one bucket to disk and continue, and the merge can read buckets back one at a time.
Amortised resize pauses. The expensive rehash is per-bucket, so no single pause is as long.
A memory-efficient large bucket. The Robin Hood switch at 2^16 cells means a bucket that grows large halves its memory overhead per key, at the cost of a little CPU for the chain ordering.

What it costs:

One extra indirection on every probe. The single-level table computes hash & mask; the two-level table computes a shift, an array index, and then the mask. The header admits it: “Usually works a little slower than a simple hash table.”
256 allocations instead of one. Each sub-table has its own buffer, its own grower, its own size counter.
A weaker guarantee for large tables. The bucket is selected from 32 bits of the hash, so a table with more than 2^32 cells is explicitly out of scope.
Complexity at the boundaries. The zero key doesn’t belong to a bucket; the prefix sums must be recomputed after growth; the string variant has to route on two axes at once.

This is a good example of a structural optimisation. Rather than making the hash table faster, ClickHouse changes the shape of the data so that the expensive part of the query — the merge — can be decomposed into 256 embarrassingly parallel pieces.

Wrapping up

Here’s the thing: the whole design rests on one sentence. The bucket is a function of the key, so the bucket is a partition of the key space. Every other property — parallel merge, incremental spill, amortised resize — falls out of that.

What to go do about it: if you have a GROUP BY over a high-cardinality column that takes a long time to finish but not a long time to start, you’re probably hitting the serial merge. Check AggregationConvertedToTwoLevel in system.events — if it’s zero and your key count is above 100 000, the conversion isn’t happening, and that’s your smoking gun.

And if you want to tune it, group_by_two_level_threshold is the knob. Lower it and the merge gets parallel sooner; the cost is one extra indirection per probe on the smaller table.