Hash Partition Join: Joining Tables Bigger Than RAM

Concept. Hash partition join joins two tables that do not fit in RAM. It hashes both tables on the join key into matching partitions, then joins each partition pair independently.

Intuition. When both Users and Listens exceed RAM, hash user_id % 16 to split each table into 16 partitions. Mickey's Users row and Mickey's Listens rows map to the same partition, so a join can run on partition 1 with partition 1 in RAM, then 2 with 2, and so on. Cost: 3(N+M) IOs. Spark and Hadoop use the same partition-by-key shuffle to join across a cluster.

When Both Tables Don't Fit in Memory

The query we want to run:

SELECT u.name, l.song_id, l.rating
FROM Users u
JOIN Listens l ON u.user_id = l.user_id;

If Users is 100 GB and Listens is 800 GB but RAM is only 64 GB, neither side fits in memory. BNLJ would re-scan the larger side once per outer block, which is too slow. Sort-merge requires sorting both, which is also expensive. Hash Partition Join is the answer.

Hash Partition Join: The Solution

Core Insight

Partition both tables using the same hash function on the join key. Rows from each side with the same key land in the same partition number, so matching can happen one partition at a time, in RAM.

Hash Partition Join: in Phase 1 both R and S are hashed on the join key into matching partitions (same key always lands in the same partition). In Phase 2 each partition pair is joined independently in RAM.

Figure 1. Hash Partition Join, R ⋈ S on user_id. Phase 1 hashes both tables with the same function so matching user_ids always land in the same partition number. Phase 2 joins each (Rᵢ, Sᵢ) pair independently in RAM, building a hash table on R and probing with S, three parallelizable joins that yield 4 + 2 + 2 = 8 rows. Best-case cost is 3(N+M) + OUT (read each table, write partitions, read them back, write the output). The worst case is hash skew, where a hot key overflows a partition past RAM and recursive re-partitioning or a sort-merge fallback adds another pass.

Algorithm · Hash Partition Join

HashPartitionJoin(T1, T2, key):       // assume |T1| < |T2|
    B = √(|T2|/RAM)                   // partition count

    Phase 1: Partition both tables on the same hash
    for row in T1: write to T1_p[h(row.key) % B]
    for row in T2: write to T2_p[h(row.key) % B]    // same hash!

    Phase 2: Join matching partitions in RAM
    for i in 0..B-1:
        HT = hash_table(T1_p[i])      // build on the smaller side
        for row in T2_p[i]:           // probe with the larger
            if HT[row.key]: output matches

Multi-Table Joins

Joining Three Tables: Users, Listens, Songs

Algorithm · Three-Way Join

ThreeWayJoin(U, L, S):              // assume |U| < |L| < |S|
    temp   = HPJ(U, L, "user_id")   // U ⋈ L
    result = HPJ(temp, S, "song_id")   // temp ⋈ S
    return result

Handling Skew in HPJ

The figure assumes partitions are roughly balanced. In practice, hot keys break that. If user_id = 1 (a system account) has 10M listens, partition 1 won't fit in RAM even though the others do. The fix is to detect oversized partitions during Phase 1 and recursively re-partition them with a different hash, until each piece fits.