ClickHouse's Hash Join Stores the Right Table as 64-bit Row Words
The ClickHouse hash join never copies right-side columns into its hash table: it keeps the right table as columnar blocks and stores one 64-bit word per row, tagged inline or as an arena pointer.
TL;DR: A hash join’s hash table does not hold the joined data. It holds one 64-bit word per key, encoding
(block_no, row_no)of a row in the right table’s stored columnar blocks — with bit 63 tagging whether the word is the row itself or a pointer to an arena node holding the duplicates. The probe loop records those words; a separate gather kernel turns them back into columns withmemcpyand a 32-row software prefetch.
Every join needs the same thing: given a key from the left side, find the right-side rows that carry that key. The textbook hash join puts the right table’s rows into the hash table. That works, but it means the joined columns are copied out of their columnar form, into a hash table’s cell layout, and then copied back out again into the result block. For a wide right table with a narrow key, that is two extra passes over the data.
ClickHouse takes the opposite route. The right table stays where it is — as a
list of columnar blocks — and the hash table stores only an address. The
address is one UInt64. Everything else in the join machinery is built on top
of that one decision.
SELECT l.id, r.payload
FROM left_table AS l
JOIN right_table AS r ON l.id = r.id;The whole join, in one picture:
BUILD PHASE (the right side, streamed once)
right block 0 right block 1 right block 2
+-----------+ +-----------+ +-----------+
| col a | | col a | | col a |
| col b | | col b | | col b |
+-----------+ +-----------+ +-----------+
| | |
+---------+--------+---------+--------+
v v
stored_columns: [blk0][blk1][blk2] arena: dup nodes, string keys
hash table (key -> one 8-byte word)
+------------+------------------------------+
| key 42 | 0x8000_0000_0000_0017 | <- inline RowRef
| key 43 | 0x0000_7fff_aabb_cdde | <- pointer to Batch node
+------------+------------------------------+
PROBE PHASE (the left side, streamed)
left block words recorded, one per output row
+-----------+ +---------------------------------+
| id = 42 | -- lookup ------> | 0x8000_0000_0000_0017 |
| id = 99 | -- lookup ------> | 0x0000_0000_0000_0000 (no hit) |
+-----------+ +---------------------------------+
GATHER PHASE (the words become columns)
words --> for each word: (block_no, row_no) --> memcpy from the stored
block, with a prefetch 32 rows aheadThe word
The address is RowRef, and it is exactly eight bytes, which matters because
it is the hash table’s cell payload:
// src/Interpreters/RowRefs.h:43
struct RowRef
{
static constexpr UInt32 INLINE_FLAG = 0x80000000u;
static constexpr UInt32 BLOCK_NO_MASK = 0x7FFFFFFFu;
static constexpr UInt64 ENCODED_INLINE_FLAG = 1ull << 63;
UInt32 row_no = 0;
UInt32 block_no = 0; /// includes INLINE_FLAG in the MSB
RowRef() = default;
RowRef(size_t block_no_, size_t row_no_)
: row_no(static_cast<UInt32>(row_no_))
, block_no(static_cast<UInt32>(block_no_) | INLINE_FLAG)
{The two halves are a block number — an index into the join’s list of stored right-hand blocks — and a row number within that block. Bit 31 of the block number is not part of the block number: it is the inline flag. When it is set, the encoded word is the row reference. When it is clear, the word is either zero (the “no match” marker) or a pointer to a list of duplicates.
// src/Interpreters/RowRefs.h:96
/// Mapped value of MapsAll join hash maps (ALL JOINs / non-unique keys): a tagged 8-byte word.
/// - bit 63 is 1: the key has exactly one row so far; the word IS the encoded RowRef (inline).
/// - bit 63 is 0 and the word is not 0: a pointer (bits 47..0) to an arena-allocated `Batch` node,
/// with the duplicate count packed into bits 62..48 (saturating; see COUNT_SAT). The count lets
/// the probe loop read `rows` straight from the cell word without dereferencing the node.
The point of the inline flag is that the common case — a key that matches exactly one right row — costs no extra allocation at all. The hash table cell is 8 bytes either way; only keys with duplicates pay for an arena node.
Duplicates live in an arena node
For an ALL join, one key can match many right rows. Those rows are chained
into a Batch node allocated from the join’s arena, and the cell word points
at the first one:
// src/Interpreters/RowRefs.h:113
/// A single 64-byte node. The cell word always points at the FIRST ("cell") node of a key.
/// `head` and the local slots are one contiguous `refs` array (refs[0] is the head) so the
/// iterator can walk them as a single run without out-of-array pointer arithmetic.
///
/// Cell node, unchained (2..7 rows): `refs[0]` holds the first row, `refs[1 .. size-1]` the rest;
/// `size == total_rows`; no pointers.
/// Cell node, chained (>= 8 rows): `refs[0 .. 5]` hold the 6 oldest rows; `refs[SLOTS]` (= refs[6])
/// is a raw pointer to the NEWEST overflow node; `size == 6 != total_rows`.
Seven rows fit inline in the 64-byte node (one head plus six slots); the eighth row turns the node into a chain head. The count of duplicates is packed into bits 48..62 of the cell word, so the probe loop can ask “how many rows does this key have?” without dereferencing anything:
// src/Interpreters/RowRefs.h:181
size_t rows() const
{
if (isInline())
return 1;
const UInt32 count = static_cast<UInt32>((word >> COUNT_SHIFT) & COUNT_SAT);
if (count != COUNT_SAT)
return count;
return asBatch()->total_rows;
}The pointer field is 48 bits, and the code explains at length why that is safe on current Linux and what would happen if it were not:
// src/Interpreters/RowRefs.h:81
/// Thrown when an arena pointer does not fit in the low 48 bits of a RowRefList word, which would
/// make the pointer+count packing ambiguous. Cannot happen on Linux x86-64/aarch64 today: even with
/// 5-level paging (`CONFIG_X86_5LEVEL`, 57-bit VA) or arm64 52-bit LVA, the kernel hands out
/// mappings above the 47-bit boundary only when the mmap address hint explicitly requests them,
So a hash table cell is: a key (whatever width the key needs) plus 8 bytes.
For an INNER JOIN on a UInt64 key that is 16 bytes of payload per
distinct key, and the joined columns are not in the table at all.
Choosing the map
Which map flavour runs is decided once, from the types of the key columns,
before a single row is read. chooseMethod is a small decision tree over
the key columns:
// src/Interpreters/HashJoin/HashJoin.cpp:427
/// If there is one numeric key that fits in 64 bits
if (keys_size == 1 && key_columns[0]->isNumeric())
{
size_t size_of_field = key_columns[0]->sizeOfValueIfFixed();
/// The loop above bails out before assigning `key_sizes` for a `LowCardinality` column.
key_sizes[0] = size_of_field;
if (size_of_field == 1)
return Type::key8;
if (size_of_field == 2)
return Type::key16;
if (size_of_field == 4)
return Type::key32;
if (size_of_field == 8)
return Type::key64;
if (size_of_field == 16)
return Type::keys128;
if (size_of_field == 32)
return Type::keys256;
throw Exception(ErrorCodes::LOGICAL_ERROR, "Numeric column has sizeOfField not in 1, 2, 4, 8, 16, 32.");
}Multiple key columns are interesting: if all of them are fixed-width and together fit in 4, 8, 16 or 32 bytes, the columns are packed into one integer and the hash table has a single integer key:
// src/Interpreters/HashJoin/HashJoin.cpp:448
/// If the keys fit in N bits, we will use a hash table for N-bit-packed keys
if (all_fixed && keys_bytes <= 4)
return Type::keys32;
if (all_fixed && keys_bytes <= 8)
return Type::keys64;
if (all_fixed && keys_bytes <= 16)
return Type::keys128;
if (all_fixed && keys_bytes <= HashJoin::max_packed_keys_bytes)
return Type::keys256;The packing itself is a hashing method, not a join-specific hack:
// src/Interpreters/HashJoin/KeyGetter.h:221
template <typename Value, typename Mapped> struct KeyGetterForTypeImpl<HashJoin::Type::keys32, Value, Mapped>
{
using Type = ColumnsHashing::HashMethodKeysFixed<Value, UInt32, Mapped, false, false, false, use_offset>;
};
template <typename Value, typename Mapped> struct KeyGetterForTypeImpl<HashJoin::Type::keys64, Value, Mapped>
{
using Type = ColumnsHashing::HashMethodKeysFixed<Value, UInt64, Mapped, false, false, false, use_offset>;
};The last resort is Type::hashed: a cryptographic hash of an unambiguously
serialised tuple of values. It is correct for any key type, but it is a
hash — two different keys could in principle collide, so it is only used
when nothing better fits.
There are 34 map types in total, including “two level” variants (a fixed
number of buckets, each with its own sub-table, to avoid a long chain of
cache misses on huge tables), “range” variants (a FixedHashMap for a dense
integer range), and low_cardinality_key_* variants that key on a
dictionary index. All of them are listed in one macro, APPLY_FOR_JOIN_VARIANTS
in src/Interpreters/HashJoin/HashJoin.h:247.
The right table itself
The right-hand blocks are kept in a std::list of stored blocks, alongside
the arena that holds the duplicate nodes and the string keys:
// src/Interpreters/HashJoin/HashJoin.h:526
struct RightTableData
{
Type type = Type::hashed;
/// tab1 join tab2 on t1.x = t2.x or t1.y = t2.y
/// =>
/// tab1 join tab2 on t1.x = t2.x
/// join tab2 on [not_joined(t1.x = t2.x)] and t1.y = t2.y
std::vector<MapsVariant> maps;
Block sample_block; /// Block as it would appear in the BlockList
StoredBlocksList columns; /// Columns of "right" table.
NullmapList nullmaps; /// Nullmaps for blocks of "right" table (if needed)
Note std::vector<MapsVariant> maps — one map per ON clause. A join with
two disjunctive conditions gets two independent hash tables, one per clause,
and the probe loop tries each in turn. This is how ClickHouse supports
ON a.x = b.x OR a.y = b.y without a general-purpose join.
The build phase appends each incoming right block to that list, and this is also where the memory accounting happens:
// src/Interpreters/HashJoin/HashJoin.cpp:1046
doDebugAsserts();
/// Register the block and account for it while a local list still owns it: `splice` cannot throw,
/// so `data->columns`, `data->allocated_size` and `data->rows_to_join` always describe the same set
/// of stored blocks. Same ordering as `tryRerangeRightTableDataImpl`; a list node keeps its address.
StoredBlocksList new_block;
new_block.emplace_back(std::move(columns), std::move(selector), std::move(row_store));
auto * stored_columns = &new_block.back();
size_t data_allocated_bytes = stored_columns->allocatedBytes();
stored_columns->block_no = data->stored_columns_index->add(stored_columns);
data->allocated_size += data_allocated_bytes;
data->rows_to_join += rows;
data->columns.splice(data->columns.end(), new_block);splice is used rather than push_back so that the list node’s address
never changes: the block number is the block’s index in the list, and the
index must stay valid for the lifetime of the join.
The “used” flags
For a RIGHT or FULL join, the right table must be reported at the end:
rows that never matched have to be emitted with default left columns. That
requires knowing, per right row, whether it was used. The flags live in a
separate structure, not in the hash table, because the hash table is
immutable once the build phase ends:
// src/Interpreters/HashJoin/HashJoin.h:665
/// Right table data. StorageJoin shares it between many Join objects.
/// Flags that indicate that particular row already used in join.
/// Flag is stored for every record in hash map.
/// Number of this flags equals to hashtable buffer size (plus one for zero value).
/// Changes in hash table broke correspondence,
/// so we must guarantee constantness of hash table during HashJoin lifetime (using method setLock)
mutable std::shared_ptr<JoinStuff::JoinUsedFlags> used_flags;There are three granularities: no flags at all, one flag per hash table
offset (the whole key), or one flag per right row (a
std::vector<std::atomic_bool> per stored block). The per-row variant is
needed when a right row can be reachable through more than one key, and it
is the reason the flags are atomic — the probe phase can run on several
threads at once (ConcurrentHashJoin shards the probe). The relaxed
ordering is enough: the flag says “some thread saw this row”, and nothing
depends on which thread.
The probe does not materialise columns
Here is the part that makes the whole design pay off. The probe loop does not copy a single right-side value into the result. It records words:
// src/Interpreters/HashJoin/AddedColumns.h:63
struct LazyOutput
{
/// Entries share the encoding of the join hash map cell values (see RowRefs.h):
/// 0 - default row; bit 63 set - inline encoded RowRef; otherwise - a RowRefList
/// list word (pointer to a Batch node in the low 48 bits + row count).
/// ASOF matches are inline encoded RowRef words too (the leaf of the sorted lookup vector).
PaddedPODArray<UInt64> row_refs;
size_t row_count = 0; /// Total number of rows in all refs and ref lists
size_t hash_table_matches = 0; /// Total number of hash table matches
row_refs is a PaddedPODArray<UInt64> — one word per output row. The
probe loop is a pure hash-table walk that appends words:
// src/Interpreters/HashJoin/AddedColumns.h:231
/// Encoded RowRef word (inline single ref, including an ASOF match) or a RowRefList
/// list word (pointer + count).
ALWAYS_INLINE void appendFromBlock(UInt64 ref_word)
{
#ifndef NDEBUG
/// `ref_word` may be an inline single ref or a list word (pointer + count); firstWord yields
/// the head ref of either, whose block is valid for the column-structure assertion.
checkColumns(*lazy_output.stored_columns[refWordBlockNo(RowRefList::fromWord(ref_word).firstWord())]);
#endif
if (record_row_refs)
lazy_output.addRef(ref_word);
}Because the words are recorded and not values, the same recording works for
every strictness: INNER, LEFT, ANY, SEMI, ANTI, ASOF. What differs
between them is only which word (or no word) is appended for a given left row.
That is what the processMatch template does:
// src/Interpreters/HashJoin/HashJoinMethodsImpl.h:619
else /// ANY LEFT, SEMI LEFT, old ANY (RightAny)
{
setUsed<need_filter>(added_columns.filter, i, added_columns.matched_rows);
used_flags.template setUsed<join_features.need_flags, flag_per_row>(find_result);
added_columns.appendFromBlock(firstRefWord(mapped));
}An ANY join appends the first word of the key’s cell; an ALL join
appends the whole cell word and lets the emit kernel expand the list.
The gather kernel
The words are turned into columns by a separate pass, and this is where the
memory bandwidth of the join actually goes. The kernel for fixed-width
columns is a memcpy loop with a software prefetch 32 rows ahead:
// src/Interpreters/HashJoin/gatherJoinOutputColumns.cpp:106
/// 32 rows of lead cover one source row's memory latency.
constexpr size_t look_ahead = 32;// src/Interpreters/HashJoin/gatherJoinOutputColumns.cpp:151
const size_t num_refs = row_refs_end - row_refs_begin;
for (size_t i = 0; i < num_refs; ++i)
{
if (i + look_ahead < num_refs)
{
/// Only an inline word carries a (block_no, row_number) address.
const UInt64 ahead = row_refs_begin[i + look_ahead];
if (refWordIsInline(ahead))
__builtin_prefetch(source_at(ahead));
}
const UInt64 word = row_refs_begin[i];
if (!word)
{
memcpy(out, default_pattern, stride);
out += stride;
continue;
}A zero word means “no match”, and it writes the column’s default pattern — which is not necessarily all-zero bytes, hence the deliberate indirection through the data type:
// src/Interpreters/HashJoin/gatherJoinOutputColumns.cpp:919
/// The `stride` bytes an unmatched row of a fixed-width column writes. Held as data rather than
/// derived, because a fixed-width default is not always bitwise zero - an `Enum`'s is its first value.
/// `type` is null below a `Nullable`, where `insertDefault` fills the nested planes from the nested
/// *column*'s default instead; the two differ for an `Enum`, and `assumeNotNull` sees it.
std::vector<char> fixedDefaultPattern(const IColumn & column, const IDataType * type)
{
MutableColumnPtr default_row = column.cloneEmpty();
if (type)
type->insertDefaultInto(*default_row);
else
default_row->insertDefault();
const std::string_view default_bytes = default_row->getRawData();
return {default_bytes.begin(), default_bytes.end()};
}The stride is a compile-time constant whenever it can be, which turns the copy into a single load and a single store:
// src/Interpreters/HashJoin/gatherJoinOutputColumns.cpp:187
template <typename F>
ALWAYS_INLINE void withStride(size_t stride, F && f)
{
switch (stride)
{
case 1: f.template operator()<1>(); break;
case 2: f.template operator()<2>(); break;
case 4: f.template operator()<4>(); break;
case 8: f.template operator()<8>(); break;
case 16: f.template operator()<16>(); break;
case 32: f.template operator()<32>(); break;
default: f.template operator()<0>(); break;
}
}Some joins need no right rows at all
LEFT SEMI and LEFT ANTI never take a value from the right side:
SEMI emits the left row alone if the key is present, ANTI emits it alone
if the key is missing. For those, the hash table does not need a mapped
value at all — it becomes a set.
That is expressed at the type level with a VoidMapped placeholder and a
conditional type alias:
// src/Interpreters/HashJoin/HashJoin.h:351
/** Different data structures, that are used to perform JOIN.
*
* A join whose result never contains a value taken from a right row - see `MapGetter` - does not
* need the mapped part of a cell at all: the table only has to answer whether a key is present.
* Such a join instantiates the maps with `VoidMapped`, which selects the set counterpart of every
* table below, so a cell holds the key alone.
*/
template <typename Key, typename Mapped, typename Hash>
using JoinHashMap = std::conditional_t<std::is_same_v<Mapped, VoidMapped>, HashSet<Key, Hash>, HashMap<Key, Mapped, Hash>>;and the choice of set maps is a policy decision with a very precise predicate:
// src/Interpreters/HashJoin/HashJoin.cpp:578
bool HashJoin::canUseSetMaps() const
{
if (!table_join->enableJoinKeyOnlyHashTables())
return false;
/// A mixed join expression is evaluated against the right rows themselves.
if (preferUseMapsAll() || table_join->getMixedJoinExpression())
return false;
/// `StorageJoin` reads its rows back out of the maps, both to `SELECT` from the table and for
/// `joinGet`.
if (table_join->isSpecialStorage())
return false;
if (kind != JoinKind::Left)
return false;
/// LEFT ANTI emits a left row only when its key is missing, and fills the right columns with
/// defaults, so no right row is ever read whatever is selected from the right side.
if (strictness == JoinStrictness::Anti)
return true;
/// LEFT SEMI emits the matched left row alone, so it qualifies when nothing of the right side
/// besides the join keys - which are taken from the left row - is selected.
return strictness == JoinStrictness::Semi && sample_block_with_columns_to_add.columns() == 0;
}A set map holds no reference into a right block, so the right blocks can be dropped as soon as the build phase ends:
// src/Interpreters/HashJoin/HashJoin.cpp:618
void HashJoin::dropRightBlocksKeptForAnotherAlgorithm()
{
if (!data || getMapsKind() != JoinMapsKind::Set)
return;and, in the build phase itself, a block that nothing references is dropped before the next one arrives:
// src/Interpreters/HashJoin/HashJoin.cpp:1175
if (!block_is_referenced && last_clause)
{
doDebugAsserts();
LOG_TRACE(log, "Skipping inserting block with {} rows", rows);
data->allocated_size -= data_allocated_bytes;
/// A set map references no block at all, so every block is dropped here even though its
/// rows did take part in the join and must stay counted.
if (maps_kind != JoinMapsKind::Set)
data->rows_to_join -= rows;For a LEFT SEMI join on a large right table, this is the difference between
a join that holds the entire right side in memory and one that holds only
the distinct keys.
Twenty-four variants, each its own binary
The join’s behaviour depends on two independent axes: the kind
(INNER/LEFT/RIGHT/FULL) and the strictness (ALL/ANY/SEMI/
ANTI/ASOF/RIGHT ANY). That is 4 × 6 = 24 combinations, and each one
generates different code: ALL needs replication, INNER needs a filter,
RIGHT needs flags, ASOF needs an inequality on top of the equality.
ClickHouse compiles all of them, and does it with an explicit instantiation per variant so the compiler can optimise each separately:
// src/Interpreters/HashJoin/LeftHashJoinAny.cpp:1
#include <Interpreters/HashJoin/HashJoinMethodsImpl.h>
namespace DB
{
template class HashJoinMethods<JoinKind::Left, JoinStrictness::Any, HashJoin::MapsOne>;
}Six lines per variant, twenty-four files. The dispatch at run time is a
static_for over the cross product, which unrolls to a switch:
// src/Interpreters/joinDispatch.h:142
/// Call function on specified join map
template <typename MapsVariant, typename Func>
inline bool joinDispatch(JoinKind kind, JoinStrictness strictness, MapsVariant & maps, JoinMapsKind maps_kind, Func && func)
{
return static_for<0, KINDS.size() * STRICTNESSES.size()>([&](auto ij)
{
// NOTE: Avoid using nested static loop as GCC and CLANG have bugs in different ways
// See https://stackoverflow.com/questions/44386415/gcc-and-clang-disagree-about-c17-constexpr-lambda-captures
constexpr auto i = ij / STRICTNESSES.size();
constexpr auto j = ij % STRICTNESSES.size();
if (kind == KINDS[i] && strictness == STRICTNESSES[j])
{The static_for is a compile-time loop, so the “if” is resolved at compile
time for each of the 24 pairs and the runtime branch disappears entirely.
What is left is a call to a template instantiation that the compiler can
specialise all the way down.
The set-map dispatch is separate, because a MapsSet has a different cell
layout and the code that handles a match is genuinely different — a match
only marks the left row:
// src/Interpreters/HashJoin/HashJoinMethodsImpl.h:503
/// A set map records that a key is present and nothing else: LEFT ANTI emits a row only when the key
/// is missing, and LEFT SEMI has no right-side value to emit, so a match only marks the left row.
template <
JoinKind KIND,
JoinStrictness STRICTNESS,
bool need_filter,
bool flag_per_row,
typename MapsTemplate,
typename Map,
typename KeyGetter,
typename AddedColumns>
requires SetJoinMaps<MapsTemplate>
void processMatch(Trade-offs
What the design buys:
- The joined columns are never copied into the hash table. A wide right table costs a key plus 8 bytes per distinct key in the map, not the full row.
- The probe loop is a hash lookup and an append. No column copying, no type switches on the joined columns, no virtual calls per value.
- The emit is a
memcpyloop with a prefetch, which is close to the memory bandwidth limit and can be vectorised by the compiler. - Set maps make
SEMI/ANTIjoins cheap — the right table can be dropped after the build.
What it costs:
- The right table stays in memory in full. There is no “the map holds the
data” option, so a join whose right side does not fit in memory has to use
a spilling algorithm (
grace_hash) or a merge join. - The ref word is 8 bytes and the block number is 31 bits. The join is limited to 2^31 stored blocks and 2^32 rows per block; both limits are checked explicitly rather than assumed.
- The 48-bit arena pointer is an assumption about the address space. It holds on Linux x86-64 and arm64 today; the code carries a diagnostic for the case where it might not.
- The laziness is a two-phase affair. The probe produces words and the
gather produces columns, so the result of a probe is not usable until the
gather runs. That is invisible in the API but it is why the result object
is a separate type (
HashJoinResult) rather than aBlock. - Twenty-four code paths. Every change to join semantics has to be checked against each kind × strictness combination, and the flags logic alone has three different granularities (no flags, per offset, per row).
Wrapping up
The one thing to remember: the hash table is a directory, not a store. It maps a key to an 8-byte address, and the address is resolved into columns later. If you are reasoning about the memory of a ClickHouse join, you should reason about the right table plus key + 8 bytes per distinct key, not about the joined rows.
The part people get wrong: the join does not drop the right table. A
LEFT SEMI join does (it uses a set map), but an ordinary INNER ALL join
holds every right-side column in memory until the query ends. When people
expect a hash join to be “one copy of the right table”, they are right, and
that is exactly the limit that pushes large joins to grace_hash.
What to go do about it: if a join is spilling or OOMing, look at the right side’s size, not at the number of result rows. And if the right side is small but the join is still slow, remember that the gather is where the time goes: it is a random-access gather from the stored blocks, in the order the probe decided, not in the order the right table is stored. The 32-row prefetch is there to hide exactly that.