Hash join

Deep dive

How it works

In: two row sets R (smaller) and S, and a join key · Out: all pairs with R.k = S.k · State: a hash table over R's rows (about 16-32 B per row plus payload); partition files when it exceeds memory · Cost: O(|R| + |S| + output) expected; one cache miss per probe on a large table.

1. Build: for each row r of R, insert (hash(r.k), r) into the table. Duplicates are allowed (chaining or multiple slots).
2. Probe: for each row s of S, hash s.k, walk the matching bucket, compare full keys and emit the matches.
3. Too big for memory (Grace): partition R and S with h₁(k) into P files each, then join each pair (Rᵢ, Sᵢ) in memory using an independent hash h₂. Hybrid: keep partition 0 in memory during partitioning, so its rows never touch disk.
4. Cache-conscious (radix join): partition in memory with 1-2 passes of radix partitioning on hash bits, so each partition's table fits in L2. Then build and probe per partition.
5. Parallel: either partition by thread, or share one build table with atomic inserts and probe with morsels (no partitioning; good when the build side is small).

probe row ──hash──► heads[h >> shift] ──► row 17 ──next──► row 4 ──next──► NIL
                                         compare keys at each step

Code

const NIL: u32 = u32::MAX;

#[inline]
fn hash(k: u64) -> u64 { (k ^ (k >> 32)).wrapping_mul(0x9E37_79B9_7F4A_7C15) } // multiplicative; use the high bits

/// Chained table in two flat arrays: bucket heads + per-row next pointers. No per-entry allocation.
pub struct JoinTable<'a> { keys: &'a [u64], heads: Vec<u32>, next: Vec<u32>, shift: u32 }

impl<'a> JoinTable<'a> {
    pub fn build(keys: &'a [u64]) -> Self {
        let bits = (keys.len().max(1) * 2).next_power_of_two().trailing_zeros(); // load factor <= 0.5
        let mut heads = vec![NIL; 1 << bits];
        let mut next = vec![NIL; keys.len()];
        for (i, &k) in keys.iter().enumerate() {
            let b = (hash(k) >> (64 - bits)) as usize;
            next[i] = heads[b];
            heads[b] = i as u32;
        }
        JoinTable { keys, heads, next, shift: 64 - bits }
    }

    /// Appends (build_row, probe_row) for every key match.
    pub fn probe(&self, probe: &[u64], out: &mut Vec<(u32, u32)>) {
        for (j, &k) in probe.iter().enumerate() {
            let mut i = self.heads[(hash(k) >> self.shift) as usize];
            while i != NIL {
                if self.keys[i as usize] == k { out.push((i, j as u32)); }
                i = self.next[i as usize];
            }
        }
    }
}

/// One radix-partitioning pass over (key, row id) pairs into 2^bits partitions.
/// Partition q is out[start[q]..start[q + 1]]. Uses low hash bits; the table uses high bits.
pub fn radix_partition(rows: &[(u64, u32)], bits: u32) -> (Vec<(u64, u32)>, Vec<usize>) {
    let p = 1usize << bits;
    let part = |k: u64| (hash(k) & (p as u64 - 1)) as usize;
    let mut start = vec![0usize; p + 1];
    for &(k, _) in rows { start[part(k) + 1] += 1; } // histogram
    for q in 0..p { start[q + 1] += start[q]; } // prefix sum → partition starts
    let mut pos = start.clone();
    let mut out = vec![(0u64, 0u32); rows.len()];
    for &(k, r) in rows {
        let q = part(k);
        out[pos[q]] = (k, r); // scatter: real code buffers a cache line per partition first
        pos[q] += 1;
    }
    (out, start)
}

/// Partitioned join: same partitioning on both sides, then a small in-cache table per partition.
pub fn partitioned_join(r: &[(u64, u32)], s: &[(u64, u32)], bits: u32) -> Vec<(u32, u32)> {
    let ((rp, rs), (sp, ss)) = (radix_partition(r, bits), radix_partition(s, bits));
    let mut out = Vec::new();
    let mut tmp = Vec::new();
    for q in 0..(1usize << bits) {
        let (rq, sq) = (&rp[rs[q]..rs[q + 1]], &sp[ss[q]..ss[q + 1]]);
        if rq.is_empty() || sq.is_empty() { continue; }
        let rkeys: Vec<u64> = rq.iter().map(|x| x.0).collect();
        let skeys: Vec<u64> = sq.iter().map(|x| x.0).collect();
        tmp.clear();
        JoinTable::build(&rkeys).probe(&skeys, &mut tmp);
        out.extend(tmp.iter().map(|&(i, j)| (rq[i as usize].1, sq[j as usize].1))); // back to row ids
    }
    out
}

Complexity and constants

• Build and probe: O(1) expected per row. With chaining, expect about 1 + load/2 key comparisons per successful probe.
• Memory here: 8 B key + 4 B next + 2 × 4 B heads ≈ 20 B per build row, plus payload. Open addressing with inline keys (Swiss table) saves the pointer chase.
• Radix partitioning: each pass streams the data once (order of GB/s per core). Fanout per pass is limited by TLB entries and cache lines, typically 2⁸-2¹¹ partitions. Use two passes for more.
• Grace join I/O: about 2 × (|R| + |S|) extra (write partitions, read them back). Hybrid saves the in-memory fraction.

Efficiency

• Cost: per probe about one DRAM miss (80-100 ns) on a table larger than the LLC, so order of 10-50 M probes/s per core; about 20 B per build row plus payload. Past RAM, Grace join adds about 2 × (|R| + |S|) of I/O. Bound: memory latency, then spill I/O.
• Levers:
  • build on the smaller side, project only needed columns — smaller table — always
  • batch and prefetch — hash a batch, prefetch buckets, then probe; often doubles probe throughput (order of) — large tables
  • hash tags in the bucket, or a Bloom filter of build keys pushed to the probe scan — avoids row loads for non-matches — selective joins
  • radix partitioning, 1-2 passes, fanout 2⁸-2¹¹ per pass — each partition's table fits in L2; hardware-dependent (Blanas et al. 2011, Balkesen et al. 2013) — builds far beyond cache
  • broadcast the small side in a cluster — no shuffle of the big side — small dimension tables
• Naive → standard → state of the art:
  • nested loop · O(|R|·|S|)
  • in-memory hash join, Grace or hybrid when spilling · linear plus one miss per probe
  • radix-partitioned, prefetched, Bloom-filtered probe · fits L2, several times faster on large builds

Implementation notes

• Hide the miss: compute hashes for a whole batch, prefetch every bucket, then probe (group or AMAC prefetching). This often doubles probe throughput on large tables.
• Tags in the bucket: store 8-16 bits of the hash next to the head pointer, or as a bucket Bloom tag, so most non-matching probes never load the row.
• Software write-combining for partitioning: buffer one cache line per partition and flush it with streaming stores. This avoids TLB thrash at high fanout.
• Spilling: decide partition counts up front from estimates, and re-partition recursively any partition that still doesn't fit (skew).
• Join types: semi and anti joins stop at the first match. Outer joins keep a "matched" bit per build row and emit the unmatched rows after the probe.
• NULL keys never match in SQL. Filter them before the build.

Pitfalls

• A wrong cardinality estimate makes you build on the big side: memory blow-up or spilling. Re-check actual sizes at runtime when possible.
• Skew: one hot key makes a long chain and an oversized partition that no amount of re-partitioning splits. Handle heavy hitters separately (broadcast them, or nested-loop them).
• A weak hash with power-of-two masking: keys with structure in the low bits collide. Use the high bits of a multiplicative or mixing hash.
• Many-to-many joins can explode the output (n × m). Check EXPLAIN estimates.
• Building on string keys without caching the hash: rehashing during resize or spill costs as much as the join.

Variants and alternatives

In production

• Databases: PostgreSQL (hybrid hash join with batches, sized by work_mem), SQL Server, Oracle, Snowflake, DuckDB, ClickHouse.
• Data processing: Spark (broadcast and shuffled hash joins), Velox.
• Dataframes: Polars and pandas implement merge/join with hash tables.

Implementations

Sources

• D. DeWitt et al., Implementation Techniques for Main Memory Database Systems, SIGMOD, 1984. paper
• M. Kitsuregawa, H. Tanaka, T. Moto-oka, Application of Hash to Data Base Machine and Its Architecture, New Generation Computing, 1983. paper
• S. Manegold, P. Boncz, M. Kersten, Optimizing Main-Memory Join on Modern Hardware, IEEE TKDE, 2002. paper
• S. Blanas, Y. Li, J. Patel, Design and Evaluation of Main Memory Hash Join Algorithms for Multi-core CPUs, SIGMOD, 2011. paper
• C. Balkesen, J. Teubner, G. Alonso, M. T. Özsu, Main-Memory Hash Joins on Multi-Core CPUs: Tuning to the Underlying Hardware, ICDE, 2013. paper