Problem 95 · Data Engineering & Streaming

Hash Join and Sort-Merge Join

MediumClassic DSA & systems

01Problem

Implement an inner equi-join of two lists of dict rows on given key columns, once as a hash join and once as a sort-merge join, handling duplicate keys and NULLs.

02Key insight

Hash join builds a hash table on the smaller side and probes with the larger: O(n+m) but needs the build side in memory. Sort-merge join is O(n log n + m log m), streams well when inputs are already sorted, and must emit the cross product of equal-key groups. SQL NULL keys never match.

03Solution

solution.pypython46 lines

from collections import defaultdict
from typing import Any

Row = dict[str, Any]


def _is_null(v: Any) -> bool:
    return v is None or v != v  # None or NaN: never equal to anything, like SQL NULL


def hash_join(left: list[Row], right: list[Row], lkey: str, rkey: str) -> list[Row]:
    build_left = len(left) <= len(right)  # build on the smaller relation
    build, probe = (left, right) if build_left else (right, left)
    bkey, pkey = (lkey, rkey) if build_left else (rkey, lkey)
    table: dict[Any, list[Row]] = defaultdict(list)
    for row in build:
        if not _is_null(row[bkey]):
            table[row[bkey]].append(row)
    out: list[Row] = []
    for p in probe:
        for b in table.get(p[pkey], ()):
            l, r = (b, p) if build_left else (p, b)
            out.append({**l, **r})
    return out


def sort_merge_join(left: list[Row], right: list[Row], lkey: str, rkey: str) -> list[Row]:
    L = sorted((r for r in left if not _is_null(r[lkey])), key=lambda r: r[lkey])
    R = sorted((r for r in right if not _is_null(r[rkey])), key=lambda r: r[rkey])
    i = j = 0
    out: list[Row] = []
    while i < len(L) and j < len(R):
        a, b = L[i][lkey], R[j][rkey]
        if a < b:
            i += 1
        elif a > b:
            j += 1
        else:  # emit cross product of the equal-key groups
            i_end, j_end = i, j
            while i_end < len(L) and L[i_end][lkey] == a:
                i_end += 1
            while j_end < len(R) and R[j_end][rkey] == a:
                j_end += 1
            out.extend({**l, **r} for l in L[i:i_end] for r in R[j:j_end])
            i, j = i_end, j_end
    return out

Try it first.

TimeHash: O(n+m); SMJ: O(n log n + m log m)

SpaceHash: O(min(n,m)); SMJ: O(n+m) for sorting

tests.py · 4 passing

tests.pypython13 lines

orders = [{"oid": 1, "cid": 10}, {"oid": 2, "cid": 20}, {"oid": 3, "cid": 10},
          {"oid": 4, "cid": None}, {"oid": 5, "cid": 99}]
customers = [{"id": 10, "name": "ann"}, {"id": 10, "name": "ann2"}, {"id": 20, "name": "bo"},
             {"id": None, "name": "ghost"}]
canon = lambda rows: sorted(tuple(sorted(r.items())) for r in rows)
hj = hash_join(orders, customers, "cid", "id")
smj = sort_merge_join(orders, customers, "cid", "id")
assert canon(hj) == canon(smj)
assert len(hj) == 5  # 2 orders x 2 dup customers for cid=10, plus cid=20
assert all(r["cid"] == r["id"] for r in hj) and not any(r["name"] == "ghost" for r in hj)
nan = float("nan")  # same object both sides: still must not match, and must not hang
L2, R2 = [{"k": nan}, {"k": 1}, {"k": None}], [{"k": nan}, {"k": 1}, {"k": None}]
assert sort_merge_join(L2, R2, "k", "k") == [{"k": 1}] == hash_join(L2, R2, "k", "k")

04Staff-level follow-up

The build side doesn't fit in memory or is skewed - what now? Grace hash join: partition both sides by hash(key) to disk and join partition pairs; for skewed hot keys, split them across workers and broadcast the matching rows from the other side.