DataFusion's Hash Join Keeps One Hash Table Per Partition and One Chain Per Key

TL;DR: DataFusion’s hash join is not one giant hash table. It is target_partitions independent tables, one per partition, because both inputs are hash-repartitioned on the join key first. Each table is two structures: a hashbrown::HashTable<(u64, u32)> mapping a folded 64-bit key hash to a row index, and a flat Vec<u32> “next” array holding the chain of duplicate keys. The payload stays columnar (one concatenated RecordBatch); only the indices are row-oriented. Multi-column keys are not encoded with Arrow’s row format — the per-column hashes are folded together. And when the key is a dense integer range, the whole hash table is replaced by a flat array (ArrayMap), i.e. a perfect hash join.

The shape of the problem

A hash join is two phases: build the table from one input, probe it with the other. The build side is the side you can afford to keep in memory. That is the whole reason DataFusion swaps join inputs around (see the sibling post on side swapping) — the build side is the one that has to fit.

The naive implementation people usually draw is a HashMap<Key, Vec<BuildRow>>. That is fine in a blog post and terrible in practice: a Vec per key means a pointer chase and a heap allocation per distinct key, and the key is stored whole — a Vec<u8> of concatenated bytes, a tuple of ScalarValues, whatever — so every probe pays to compare the full key.

DataFusion’s actual design takes the opposite approach on both counts: the hash table stores hash values and row indices only, never keys, and the payload never leaves Arrow’s columnar arrays.

Partitioning first, tables second

Before any hashing happens, the plan is rewritten so that both sides of the join are partitioned by the join key:

rust
// datafusion/physical-plan/src/joins/hash_join/exec.rs:1693
    fn input_distribution_requirements(&self) -> InputDistributionRequirements {
        match self.mode {
            PartitionMode::Partitioned => {
                let (left_expr, right_expr) = self
                    .on
                    .iter()
                    .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
                    .unzip();
                InputDistributionRequirements::co_partitioned(vec![
                    Distribution::KeyPartitioned(left_expr),
                    Distribution::KeyPartitioned(right_expr),
                ])
            }
            PartitionMode::CollectLeft => InputDistributionRequirements::new(vec![
                Distribution::SinglePartition,
                Distribution::UnspecifiedDistribution,
            ]),
            ...

co_partitioned means: partition i of the left and partition i of the right contain exactly the same set of join keys. The EnforceDistribution rule satisfies that with a RepartitionExec(Partitioning::Hash) on each side, using a fixed seed so both sides route identically:

rust
// datafusion/physical-plan/src/repartition/mod.rs:677
pub const REPARTITION_RANDOM_STATE: SeededRandomState = SeededRandomState::with_seed(0);

and the hash join itself deliberately uses a different seed for its internal table:

rust
// datafusion/physical-plan/src/joins/hash_join/exec.rs:114
/// Hard-coded seed to ensure hash values from the hash join differ from `RepartitionExec`, avoiding collisions.
pub(crate) const HASH_JOIN_SEED: SeededRandomState =
    SeededRandomState::with_seed(12210250226015887276);

That detail matters: if the join’s hash were the same function as the routing hash, an adversarial key distribution could make every key land in one partition — a “hash collision” attack on the routing function. Two independent hash spaces mean the routing decision and the in-partition bucketing cannot be attacked with the same key set.

The table: hash → chain head, chain → row index

The table itself is remarkably small:

rust
// datafusion/physical-plan/src/joins/join_hash_map.rs:170
pub struct JoinHashMapU32 {
    // Stores hash value to last row index
    map: HashTable<(u64, u32)>,
    // Stores indices in chained list data structure
    next: Vec<u32>,
    // Whether any hash value maps to more than one row
    chained: bool,
}

Two things to notice. First, the key of the hash table is the hash value, not the join key. Second, the value stored is row_index + 1, and 0 means “end of chain”. The +1 bias is what lets next use 0 as a sentinel without a separate bitmap.

The chain layout is documented in the source, and the doc comment is worth reading in full because it explains the whole data structure:

text
// datafusion/physical-plan/src/joins/join_hash_map.rs:69 (doc comment)
/// Insert (10,1)            <-- insert hash value 10 with row index 1
/// map:  | 10 | 2 |          next: | 0 | 0 | 0 | 0 | 0 |
/// Insert (10,3)           <-- collision! row index 3 has a hash value of 10 as well
/// map:  | 10 | 4 |          next: | 0 | 0 | 0 | 2 | 0 |
///                                        <--- hash value 10 maps to 4,2

The map holds the head of the chain; next[i] holds the next row with the same hash. Insertion is a compare-and-swap on the head:

rust
// datafusion/physical-plan/src/joins/join_hash_map.rs:366
            Occupied(mut occupied_entry) => {
                // Already exists: add index to next array
                let (_, index) = occupied_entry.get_mut();
                let prev_index = *index;
                // Store new value inside hashmap
                *index = T::try_from(row + 1).unwrap();
                // Update chained Vec at `row` with previous value
                next[row - deleted_offset] = prev_index;
                chained = true;
            }

The chained flag is a genuinely clever micro-optimisation: it records whether any hash value ever had a second row. If not, the probe can skip chain traversal entirely:

rust
// datafusion/physical-plan/src/joins/join_hash_map.rs:456
    // Check if hashmap consists of unique values
    // If so, we can skip the chain traversal
    if unique {

That is the common case for a fact/dimension join on a primary key, and it turns the probe from “hash lookup + chain walk” into “hash lookup”.

u32 or u64: pick the index width from the row count

rust
// datafusion/physical-plan/src/joins/hash_join/exec.rs:3063
fn new_join_hashmap(num_rows: usize, ...) -> Result<Box<dyn JoinHashMapType>> {
    ...
    if num_rows > u32::MAX as usize {
        // (u64, u64) entries plus a u64 chain index per row
        Ok(Box::new(JoinHashMapU64::with_capacity(num_rows)))
    } else {
        // (u32, u64) entries plus a u32 chain index per row
        Ok(Box::new(JoinHashMapU32::with_capacity(num_rows)))
    }
}

u32 halves the chain array and the value half of the table entry — for a 100M-row build side that is 400 MB versus 800 MB of pure index overhead. The reservation is grown before the allocation so the memory pool can reject the join up front rather than after the allocation has already happened.

Multi-column keys: folded hashes, not Arrow row format

This is where most blog posts about DataFusion joins go wrong. There is no RowConverter in the hash join. grep -rn RowConverter datafusion/physical-plan/src/joins/ returns nothing. Arrow’s row format is used elsewhere in DataFusion — notably by the range-partitioning router:

rust
// datafusion/physical-plan/src/repartition/range.rs:46
/// Universal fast path using Arrow's RowConverter for arbitrary types and composite keys.
Row(RowConverterRangeRouter),

and by the sort machinery (datafusion/physical-plan/src/sorts/stream.rs:201). But the hash join does not need it, because the row format exists to make ordered comparison cheap (memcmp of a byte blob). A hash join only needs two things: a hash for bucketing, and an equality test. So DataFusion folds per-column hashes instead:

rust
// datafusion/common/src/hash_utils.rs:1264
pub fn create_hashes<'a, I, T>(
    arrays: I,
    random_state: &impl HashState,
    hashes_buffer: &'a mut [u64],
) -> Result<&'a mut [u64]>
where
    I: IntoIterator<Item = T>,
    T: AsDynArray,
{
    for (i, array) in arrays.into_iter().enumerate() {
        // combine hashes with `combine_hashes` for all columns besides the first
        let rehash = i >= 1;
        hash_single_array(array.as_dyn_array(), random_state, hashes_buffer, rehash)?;
    }
    Ok(hashes_buffer)
}

The folding is type-dependent, which is a nice detail. For primitives, the previous column’s hash is used as the seed of the next hasher, so the columns are hashed together as one stream:

rust
// datafusion/common/src/hash_utils.rs:322
    if array.null_count() == 0 {
        if rehash {
            for (hash, &value) in hashes_buffer.iter_mut().zip(array.values().iter()) {
                let mut hasher = random_state.seeded_state(*hash).build_hasher();
                value.hash_write(&mut hasher);
                *hash = hasher.finish();
            }
        } else {

For everything else (strings, binary, lists…), the fold is an explicit mix:

rust
// datafusion/common/src/hash_utils.rs:100
/// Combines two hashes into one hash
#[inline]
pub fn combine_hashes(l: u64, r: u64) -> u64 {
    let hash = (17 * 37u64).wrapping_add(l);
    hash.wrapping_mul(37).wrapping_add(r)
}

Either way the fold is order dependent and type dependent. Both sides of the join use the same create_hashes call with the same column order, so it does not matter — but it does mean hashes computed by different code paths are not comparable.

Because the key is hashed, not stored, the hash table cannot answer “are these keys equal?” — only “might these keys be equal?”. The exact check happens after the lookup, on the actual key columns, via equal_rows_arr (datafusion/physical-plan/src/joins/utils.rs:2420):

rust
    for (left, right) in indices_left.values().iter().zip(indices_right.values()) {
        let left_idx = usize::try_from(*left)...;
        let right_idx = *right as usize;

        if comparator.is_equal(left_idx, right_idx) {
            left_filtered.push(*left);
            right_filtered.push(*right);
        }
    }

and there is a monomorphised single-column fast path that skips the boxed DynComparator dispatch entirely (utils.rs:2449):

rust
/// Specialized single-column equi-join key filtering.
///
/// Dispatches once on the key column's type and runs a monomorphized equality
/// loop with typed value comparison. This avoids the per-pair boxed
/// `DynComparator` call and the three-way `Ordering` computation ...

So the full probe cost is: hash the probe key (one pass over the key columns), one hashbrown lookup, then one typed equality comparison per candidate. The equality comparison is only paid for candidates that survived the hash — a collision is rare, so it is essentially free.

Where the payload lives

The build side’s payload is columnar. collect_left_input drains the whole build stream of the partition, then concatenates it into a single RecordBatch:

rust
// datafusion/physical-plan/src/joins/hash_join/exec.rs:3125
/// Concatenates the build side `batches` into a single batch, in reverse order
/// if `reverse` is set, keeping `reservation` ahead of the actual memory usage.
fn concat_build_batches(...)

Why concatenate? Because the hash map stores row indices into that batch, and the output is produced by gathering from it. Concatenation makes the index space contiguous; keeping the batches as a list of segments would require a two-level index for every probe. (The nested-loop join does keep them segmented — see LogicalBatch in datafusion/physical-plan/src/joins/logical_batch.rs:62, which exists precisely to avoid the 2x memory blow-up of concatenation for that operator.)

So the answer to “row-oriented vs columnar” is: the indices are row-oriented, the data is columnar, and the join output is built by takeing from both sides via build_batch_from_indices (datafusion/physical-plan/src/joins/utils.rs:1311). That is the entire “row format” of the join: two index arrays, one per side, and the columns gathered by them.

Probing in chunks, because one hot key can explode

A probe row that matches 10M build rows must not produce a 10M-row output batch. DataFusion therefore probes in chunks, with a resumable cursor:

rust
// datafusion/physical-plan/src/joins/join_hash_map.rs:42
pub enum ProbeOffset {
    /// Resume at probe row `row`, from the head of its chain.
    AtRow { row: usize },
    /// Resume in the middle of probe row `row`'s chain, where `next` is the
    /// 1-based index of the next build row in the chain. `next` is never `0`,
    /// the end-of-chain marker.
    MidChain { row: usize, next: u64 },
}

get_matched_indices_with_limit_offset walks the chain until the output limit (batch size) is reached and hands back a ProbeOffset that says exactly where to resume — including inside a chain, which is the subtle part (datafusion/physical-plan/src/joins/chain.rs:48):

rust
    loop {
        match_indices.push(match_row_idx.into());
        input_indices.push(prob_idx as u32);
        *remaining -= 1;

        let next = next_chain[match_row_idx.into() as usize];

        if *remaining == 0 {
            // Limit reached - return offset for next call
            return if is_last_input && next == zero {
                None
            } else if next == zero {
                Some(AtRow { row: prob_idx + 1 })
            } else {
                Some(MidChain { row: prob_idx, next: next.into() })
            };
        }

Every probe chunk is a full pipeline stage: hash → look up → filter → gather → emit. The cursor state lives in the stream, so the join can suspend and resume without redoing work.

The dense special case: ArrayMap

If the key is a single integer and its range is dense enough, the hash table is pointless. DataFusion detects that and replaces it with a flat array:

rust
// datafusion/physical-plan/src/joins/array_map.rs:104
pub struct ArrayMap {
    // data[probSideVal-offset] -> valIdxInBuildSide + 1; 0 for absent
    data: Vec<u32>,
    // min val in buildSide
    offset: u64,
    // next[buildSideIdx] -> next matching valIdxInBuildSide + 1; 0 for end of chain.
    // If next is empty, it means there are no duplicate keys (no conflicts).
    next: Vec<u32>,
    num_of_distinct_key: usize,
}

Probe becomes data[key - min] — one subtraction and one array load, no hashing at all. The decision is made from the build side’s observed min/max and row count:

rust
// datafusion/physical-plan/src/joins/hash_join/exec.rs:191
    let dense_ratio = (num_row as f64) / ((range + 1) as f64);

    if range >= perfect_hash_join_small_build_threshold as u64
        && dense_ratio <= perfect_hash_join_min_key_density
    {
        return Ok(None);
    }

with the two knobs exposed as datafusion.execution.perfect_hash_join_small_build_threshold (default 1024) and datafusion.execution.perfect_hash_join_min_key_density (default 0.15). Dense keys — surrogate IDs, date columns, zip codes — hit this path and the probe becomes nearly free. Sparse keys (UUIDs, hashes) correctly fall back to the hash table. Note the next array is only allocated if there are duplicate keys: the “chain” concept survives into the perfect-hash variant, but it is empty in the common case.

Memory: reserved, double-counted on purpose, never spilled

The build side is charged to the memory pool as it arrives, and the concatenation is reserved before it happens (datafusion/physical-plan/src/joins/hash_join/exec.rs:3134):

rust
    let copy_size = if batches.len() > 1 { ... } else { 0 };
    reservation.try_grow(copy_size)?;
    metrics.build_mem_used.add(copy_size);

    let batch = if reverse {
        concat_batches(schema, batches.iter().rev())?
    } else {
        concat_batches(schema, batches.iter())?
    };

so for a moment the build side occupies twice its final size in the pool: the fragmented batches plus the concatenated copy. Afterwards the reservation is trimmed to what the concatenated batch actually retains (exec.rs:3159):

rust
    let held = inputs_reserved + copy_size;
    let retained = get_record_batch_memory_size(&batch);
    if retained > held {
        reservation.try_grow(retained - held)?;

This is the memory trade-off of a hash join in DataFusion, stated plainly: the build side must fit in RAM, twice over, transiently. Unlike the sort-merge join (which registers a SpillManager and spills its buffered side to disk — datafusion/physical-plan/src/joins/sort_merge_join/exec.rs:966), the hash join has no spill path at all. If the pool refuses the reservation, the query fails. That is why the side-swapping optimizer rule matters so much: choosing the wrong build side is not a slow query, it is an out-of-memory error.

Pathological cases

Duplicate-heavy keys. A build side where every key is the same value produces a chain of length n, and the output is n rows per probe row. The chained flag correctly disables the unique fast path, but nothing stops the quadratic output. This is where the sort-merge join wins: it spills, and it can emit the cross product more cheaply.
Hash collisions. Rare with foldhash, but the equality check after the lookup is what makes them correct rather than merely unlikely. It is on the actual key values, so a collision costs one extra typed comparison.
u32 overflow. A build side above 4.29 billion rows switches the whole table to u64 indices, doubling index memory at exactly the point where memory is tightest.
Skewed routing. PartitionMode::Partitioned assumes the hash distributes keys evenly. It does not when a handful of key values are hot. One partition gets a huge build side, and the join’s memory profile is that of the worst partition.
The perfect-hash trap. ArrayMap allocates 4 * (max - min + 1) bytes. The density heuristic prevents a sparse key set from triggering a giant allocation, but it only applies to integer keys with observed min/max — a key set with one tiny outlier (say i64::MIN among 1..1000) makes the range enormous and the density check correctly rejects the ArrayMap.

What to take away

The hash join’s cleverness is not in the hashing; it is in refusing to store keys. The table maps a folded 64-bit hash to a chain of 32-bit row indices; the data stays in Arrow arrays; the equality check is done on typed values only for hash candidates. Multi-column keys are handled by folding hashes rather than by materialising Arrow row bytes, which is cheaper precisely because the join never needs ordered comparison. And when the key range is dense, the entire hash table is thrown away in favour of a flat array.

The price is memory: the build side must fit in the pool, twice, transiently, with no spill fallback. Every trick in the surrounding optimizer — side swapping, CollectLeft mode, the ArrayMap fast path, the dynamic-filter key-range bitmap that lets the probe side skip containers — exists to keep that build side small enough to fit.