10. Joins and dictionaries

ClickHouse supports JOIN. It is the part of ClickHouse that most often disappoints people coming from PostgreSQL, and the disappointment is avoidable if you know the mechanics.

The single sentence that explains it:

ClickHouse loads the entire right-hand table into a hash table in memory, then streams the left table through it.

Everything below follows from that.

The default algorithm

SELECT p.url, u.name
FROM page_views AS p
INNER JOIN users AS u ON p.user_id = u.user_id;

What happens:

The whole right side (users) is read and built into a hash table in RAM. All of it. On every node involved.
The left side (page_views) is streamed through, probing the hash table per row.

Two consequences follow immediately, and they are the two things people get wrong.

The right table must fit in memory. If users is 200 GB, the query fails with a memory limit error. It does not matter that only a few thousand users appear in the filtered page_views.

Order is not optimised for you. ClickHouse historically had no cost-based join reordering — it joins in the order you wrote. So:

-- ❌ 40 billion rows built into a hash table. Fails.
SELECT … FROM users AS u JOIN page_views AS p ON u.user_id = p.user_id;

-- ✅ 5 million rows built into a hash table. Fine.
SELECT … FROM page_views AS p JOIN users AS u ON p.user_id = u.user_id;

Always put the smaller table on the right. Newer versions have added query-plan-level optimisations and can sometimes reorder, but do not build a system that depends on it — write the order you want.

The join algorithms

join_algorithm selects the strategy, and the default is direct,parallel_hash,hash on modern versions (it has changed over releases — check yours).

Algorithm	How it works	When
`hash`	classic build-and-probe	the default; right side must fit in RAM
`parallel_hash`	builds the hash table with multiple threads	large right side, many cores
`direct`	looks up in a key-value source directly	joining against a `Dictionary` or `Join` engine table
`partial_merge`	sorts both sides, merge-joins with disk spill	right side too large for memory
`full_sorting_merge`	sorts both sides fully, merges	large-to-large joins
`grace_hash`	partitions both sides into buckets, joins bucket by bucket, spills	large-to-large, bounded memory
`auto`	starts with hash, falls back on memory pressure	convenient safety net

The practical guidance:

-- right side too big for RAM
SET join_algorithm = 'grace_hash';         -- or 'partial_merge'

-- both sides huge and already sorted by the key
SET join_algorithm = 'full_sorting_merge';

-- do not think about it
SET join_algorithm = 'auto';

grace_hash is the one worth remembering: it partitions both inputs by hash of the key into buckets that individually fit in memory, and processes bucket by bucket. It makes large-to-large joins possible at a cost in I/O.

Distributed joins — the part that surprises people

On a sharded cluster this gets more subtle, and getting it wrong turns a fast query into a network flood.

By default, when you join in a distributed query, the right-hand table is read and sent to every shard. Each shard then joins locally against its slice of the left table.

   shard 1  ──┐
   shard 2  ──┼── each receives the FULL right table
   shard 3  ──┘    and joins its own left-side slice

For a small dimension table that is fine. For a large one it is a network transfer of the whole table to every shard, per query.

GLOBAL JOIN changes it: the initiator node evaluates the right side once, then broadcasts the result to all shards.

SELECT … FROM distributed_page_views AS p
GLOBAL JOIN (SELECT user_id, name FROM users WHERE active) AS u
USING (user_id);

The rule of thumb: GLOBAL JOIN when the right side is a subquery or a non-distributed table, plain JOIN when both tables are sharded by the same key — because then each shard already holds the matching rows locally and no transfer is needed.

That last case, co-located joins, is the ideal. If both tables are sharded by user_id, a join on user_id is entirely local on every shard. Designing for it is worth real effort when you have a join you cannot avoid.

`USING` versus `ON`, and other differences

JOIN users USING (user_id)          -- the key appears once in the output
JOIN users ON p.user_id = u.user_id -- both columns appear

Some other divergences from standard SQL worth knowing:

ANY and ALL modifiers. ANY LEFT JOIN takes at most one matching row from the right (whichever it finds first) and is faster. ALL — the default — takes every match. SEMI and ANTI joins exist too:

SELECT … FROM a ANY LEFT JOIN b USING (id);      -- at most one match
SELECT … FROM a LEFT SEMI JOIN b USING (id);     -- rows of a that have a match
SELECT … FROM a LEFT ANTI JOIN b USING (id);     -- rows of a with no match

SEMI and ANTI are usually much faster than the IN (SELECT …) or NOT IN you would otherwise write.

ASOF JOIN — join to the nearest preceding row by a time column. This is a genuine specialty and it is excellent for time-series work:

SELECT t.ts, t.symbol, t.price, q.bid
FROM trades AS t
ASOF LEFT JOIN quotes AS q
  ON t.symbol = q.symbol AND t.ts >= q.ts;

For each trade, the most recent quote at or before it. Doing that with a standard join and a correlated subquery is both awkward and slow.

Nulls in LEFT JOIN. By default, unmatched rows get the column’s default value (0, '') rather than NULL. Set join_use_nulls = 1 for standard SQL behaviour. This catches people out silently, because 0 looks like data.

Dictionaries — usually the better answer

For the common case — enriching a large fact table with attributes from a small dimension table — do not join. Use a dictionary.

A dictionary is a key-value structure held in memory (or memory-mapped), refreshed automatically from a source, and queried with a function call.

CREATE DICTIONARY user_dict (
    user_id UInt64,
    name    String,
    country String,
    tier    UInt8
)
PRIMARY KEY user_id
SOURCE(POSTGRESQL(
    host 'pg.internal' port 5432 user 'ch' password '…'
    db 'app' table 'users'))
LAYOUT(HASHED())
LIFETIME(MIN 300 MAX 600);

SELECT
    dictGet('user_dict', 'name', user_id)    AS name,
    dictGet('user_dict', 'country', user_id) AS country,
    count()
FROM page_views
WHERE ts >= today()
GROUP BY name, country;

Why this is better than a join:

No hash table build per query. The dictionary is already in memory, permanently.
It refreshes itself from PostgreSQL, MySQL, MongoDB, a ClickHouse table, HTTP, a file, or an executable — on the LIFETIME schedule, without you writing a pipeline.
It works inside WHERE, GROUP BY and ORDER BY, not just in the select list.
It is available on every shard of a distributed cluster, so there is no network transfer.

The layouts, which control how it is stored:

   FLAT              array indexed by key — fastest, for dense small integer keys
   HASHED            hash table — the general default
   SPARSE_HASHED     less memory, slightly slower
   COMPLEX_KEY_HASHED  composite keys
   CACHE             LRU with lookups on miss — for huge sources
   DIRECT            no caching, query the source every time
   RANGE_HASHED      key valid over a date range — for slowly-changing dimensions
   IP_TRIE           IP prefix lookups — for geo-IP

RANGE_HASHED deserves a mention: it handles the “what was this customer’s tier on the date of the order?” problem natively, which is otherwise a fiddly range join.

The functions:

dictGet('dict', 'attr', key)                    -- fetch
dictGetOrDefault('dict', 'attr', key, default)  -- with a fallback
dictHas('dict', key)                            -- existence
dictGetHierarchy('dict', key)                   -- parent chain for tree dictionaries

Monitor them:

SELECT name, status, element_count, formatReadableSize(bytes_allocated),
       last_successful_update_time, last_exception
FROM system.dictionaries;

last_exception non-empty means the dictionary has stopped refreshing and is serving stale data — silently. Alert on it.

Denormalisation — the other answer

The most ClickHouse-native solution is to not need the join at all.

Storage is cheap and compression is excellent, so write the dimension attributes into the fact table at insert time.

CREATE TABLE page_views (
    ts        DateTime,
    user_id   UInt64,
    user_tier LowCardinality(String),    -- denormalised from users
    country   LowCardinality(String),    -- denormalised from users
    url       String
) ENGINE = MergeTree ORDER BY (country, ts);

user_tier repeated across 40 billion rows compresses to almost nothing because it is low-cardinality and sorted-adjacent. You have traded a small amount of disk for eliminating a join from every query.

The objection is the usual one: if a user’s tier changes, historical rows keep the old value. For analytics that is frequently correct — you want to know what tier they were when the event happened, not what they are now. When you genuinely need the current value, that is what a dictionary is for.

In an analytical schema, denormalise by default. This is the opposite instinct to a relational one and it is right here for the same reason it is right in Cassandra: the cost you are avoiding is far larger than the cost you are paying.

The `Join` engine

A table engine that is a persistent hash table, prepared once for repeated use:

CREATE TABLE user_join (user_id UInt64, name String)
ENGINE = Join(ANY, LEFT, user_id);

INSERT INTO user_join SELECT user_id, name FROM users;

SELECT p.url, j.name FROM page_views AS p
ANY LEFT JOIN user_join AS j USING (user_id);

It avoids rebuilding the hash table per query and works with joinGet('user_join', 'name', user_id) as a function. Dictionaries do the same job with better refresh semantics, so this is mostly for cases where a dictionary source does not fit.

What goes wrong

The big table on the right. Memory limit exceeded. Swap the order.

JOIN instead of GLOBAL JOIN in a distributed query — the right table is transferred to every shard, per query.

Assuming LEFT JOIN produces NULL. It produces defaults unless join_use_nulls = 1, and 0 silently looks like a real value.

A three-way join over large tables. Each one builds another hash table. This is where “ClickHouse joins are bad” comes from, and it is usually a schema problem.

A dictionary that stopped refreshing. last_exception in system.dictionaries, serving stale data with no error.

Joining to get a value you could have denormalised. The most common avoidable join.

What to take from this chapter

ClickHouse builds the whole right table into a hash table in memory and streams the left through it. Always put the smaller table on the right.
Join order is largely your responsibility — do not depend on reordering.
grace_hash and partial_merge make large-to-large joins possible with bounded memory, at an I/O cost.
On a cluster, GLOBAL JOIN evaluates the right side once and broadcasts it. Plain JOIN sends the right table to every shard. Co-located joins on a shared shard key are the ideal.
ASOF JOIN (nearest preceding row by time), SEMI/ANTI, and ANY are genuinely useful specialties.
LEFT JOIN gives defaults, not NULL, unless join_use_nulls = 1.
Dictionaries beat joins for small dimension tables: in memory, self-refreshing from an external source, usable anywhere in the query, and present on every shard. Alert on last_exception.
Denormalise by default. In a column store, a repeated low-cardinality column costs almost nothing and removes a join permanently.

Next: how to make a slow query fast by computing it before it is asked.

Written by Sepuri Sai Krishna