How DataFusion's Hash Join Keeps Its Build Side in Order: Reverse the Insertion, Get FIFO Output for Free
DataFusion's hash table is a LIFO chain, so it inserts the build side in reverse order — reversing both the batch sequence and the rows within each batch — so that probe results come out in the original build order.
TL;DR: A hash table with collision chains can be LIFO (prepend) or FIFO (append). DataFusion’s is LIFO, because prepending is cheaper. But LIFO means a probe returns the latest matching row first — which would scramble the join’s output order. The fix is one line: insert the build side in reverse order. Since the chain is LIFO, the last thing inserted is the first thing found, so reversing the insertion reverses the output. The rows come out in the build side’s original order, and the hash table got to keep the cheap prepending.
A hash join’s output order is not just a cosmetic detail. SQL does not guarantee the order of a join’s output, but DataFusion does something specific: it preserves the build side’s row order among the rows that match a given probe row. That is observable, it is tested, and it makes the join’s output stable and reproducible.
Getting it for free out of a hash table is the trick.
The chain is LIFO
DataFusion’s build-side hash table (JoinHashMap) stores, for each hash
bucket, a chain of row indices. The chain is built by prepending — the
newest row goes at the head. That is the cheap operation: you have a
Vec<u32> of next pointers, and prepending is “set the new row’s next
to the old head, and the head to the new row”.
The doc on HashJoinExec states the consequence plainly:
When using the standard
JoinHashMap, hash join uses LIFO data structure as a hash table, and in order to retain original build-side input order while obtaining data during probe phase, hash table is updated by iterating batch sequence in reverse order – it allows to keep rows with smaller indices “on the top” of hash table, and still maintain correct indexing for concatenated build-side data batch.
The diagram
The source includes a worked example that makes the whole thing concrete. Three batches of build-side rows:
Original build-side data Inserting build-side values into hashmap Concatenated build-side batch
┌───────────────────────────┐
hashmap.insert(row-hash, row-idx + offset) │ idx │
┌───────┐ │ ┌───────┐ │
│ Row 1 │ 1) update_hash for batch 3 with offset 0 │ │ Row 6 │ 0 │
Batch 1 │ │ - hashmap.insert(Row 7, idx 1) │ Batch 3 │ │ │
│ Row 2 │ - hashmap.insert(Row 6, idx 0) │ │ Row 7 │ 1 │
└───────┘ │ │ │ │
│ │
┌───────┐ │ ┌───────┐ │
│ Row 3 │ 2) update_hash for batch 2 with offset 2 │ │ Row 3 │ 2 │
Batch 2 │ │ - hashmap.insert(Row 5, idx 4) │ Batch 2 │ Row 4 │ 3 │
│ Row 4 │ - hashmap.insert(Row 4, idx 3) │ │ │ │
│ Row 5 │ │ │ Row 5 │ 4 │
└───────┘ │ └───────┘ │
│ │
┌───────┐ │ ┌───────┐ │
│ Row 6 │ 3) update_hash for batch 1 with offset 5 │ │ Row 1 │ 5 │
Batch 3 │ │ - hashmap.insert(Row 2, idx 6) │ Batch 1 │ │ │
│ Row 7 │ - hashmap.insert(Row 1, idx 5) │ │ Row 2 │ 6 │
└───────┘ │ └───────┘ │Read the three columns carefully.
The concatenated batch is in reverse batch order. Batch 3 comes first in the concatenated batch (rows 6 and 7, at indices 0 and 1), then batch 2 (rows 3, 4, 5 at indices 2, 3, 4), then batch 1 (rows 1, 2 at indices 5, 6). Note that within a batch, the row order in the concatenated batch is the original order: row 6 before row 7, row 3 before row 4.
The insertion order is the exact reverse of the original build order. Batch 3 is processed first, and within it, row 7 is inserted before row 6. Then batch 2: row 5, row 4, row 3. Then batch 1: row 2, row 1. So the insertion sequence is 7, 6, 5, 4, 3, 2, 1.
Because the chain is LIFO, the head of the chain is the last thing inserted — row 1. Row 1 is the first row of the first batch of the original build side. So when a probe row matches, the chain yields row 1 first, then row 2, then row 3, and so on: the original build order.
The index stored in the hash table (5 for row 1) points into the reversed concatenation, so the row data is still fetched correctly. The indices and the order are two different things, and the trick uses that separation.
The code
The batch loop is three lines:
// Updating hashmap starting from the last batch
for batch in batches.iter().rev() {
hashes_buffer.clear();
hashes_buffer.resize(batch.num_rows(), 0);
update_hash(
on_left,
batch,
&mut *hashmap,
offset,
random_state,
&mut hashes_buffer,
0,
true, // <-- fifo_hashmap
null_equality,
)?;
offset += batch.num_rows();
}and the reversal within a batch happens inside update_hash:
// Updating JoinHashMap from hash values iterator
let hash_values_iter = hash_values
.iter()
.enumerate()
.filter(|(i, _)| valid_keys.as_ref().is_none_or(|nulls| nulls.is_valid(*i)))
.map(|(i, val)| (i + offset, val));
if fifo_hashmap {
hash_map.update_from_iter(Box::new(hash_values_iter.rev()), deleted_offset);
} else {
hash_map.update_from_iter(Box::new(hash_values_iter), deleted_offset);
}hash_values_iter.rev() — the rows of the batch are inserted in reverse.
And the fifo_hashmap flag is the switch: true means “insert in reverse
so the output comes out FIFO”.
The flag name is slightly confusing at first — fifo_hashmap does not
mean “the hash map is FIFO”, it means “make the output FIFO”. The
hash map is LIFO; the flag reverses the insertion to compensate.
And the concatenation is reversed to match:
// Merge all batches into a single batch, so we can directly index into the arrays
let batch = if prepared {
concat_batches(schema, batches.iter().rev())?
} else {
concat_build_batches(schema, batches, true, input_bytes, &mut reservation, metrics)?
};batches.iter().rev() — same reversal, so the index arithmetic
(offset += batch.num_rows()) lines up with the concatenated layout.
Why reverse the concatenation at all?
You might think reversing the insertion alone is enough. It is not, quite.
The concatenated batch is the array you index into to fetch a build row.
The hash table stores (hash, index) where index is the row’s position
in the concatenated batch. If the concatenation were in forward order
(batch 1 at indices 0-1, batch 2 at 2-4, batch 3 at 5-6) and the
insertion were reversed (7, 6, 5, …, 1), the chain head would still be
row 1 (index 0) and the output order would still be correct.
So why does the code reverse the concatenation?
Because the offset arithmetic is per-batch and monotonic. The code
processes batches in reverse and increments offset as it goes, so
batch 3 gets offset 0, batch 2 gets offset 2, batch 1 gets offset 5 —
which is exactly the layout of the reversed concatenation. If the
concatenation were forward but the offsets were computed in reverse, the
indices would point at the wrong rows.
So the reversal of the concatenation and the reversal of the offset assignment are two halves of the same decision, and they have to agree. Reversing the concatenation makes “process batches in reverse and assign increasing offsets” work out.
(It also has a second effect: the concatenated batch’s rows are in reverse-batch order, but the within-batch order is preserved. So the concatenated batch is not the original build order — it is the original build order with the batches reversed. The chain is what restores the full original order.)
The cost
Reversing the insertion order costs… nothing, at runtime. The
hash_values_iter.rev() is an iterator adapter, and the hash map’s
update_from_iter does not care about the order it receives rows in
(only about the LIFO/FIFO direction, which is what we want to change).
Reversing the concatenation costs one thing: the output batch’s rows are grouped differently. But the output batch is assembled by gathering rows via indices, and the indices come from the chain, which is in the right order. So the concatenation order is invisible in the output.
The only real cost is the memory reservation dance, which the code handles carefully:
// Concatenating a single batch is zero-copy
let copy_size = if batches.len() > 1 {
...
} else {
0
};
reservation.try_grow(copy_size)?;and after the concatenation:
// The inputs are gone: only hold on to what the concatenated batch retains,
// which includes any buffers it still shares with the inputs.
let held = inputs_reserved + copy_size;
let retained = get_record_batch_memory_size(&batch);
if retained > held {
reservation.try_grow(retained - held)?;
...
} else {
reservation.shrink(held - retained);
...
}The reservation is trimmed to what the concatenated batch actually
retains — including buffers it still shares with the inputs, which is
why the comment says “any buffers it still shares with the inputs”. Arrow’s
concat_batches may share buffers rather than copy when the inputs are
compatible, and the accounting has to know that.
The catch
- This is a
HashJoinExecdetail, not a general one. TheArrayMap(perfect hash join) has its own ordering rules, and the sort-merge join has none at all — it is already ordered. - The
fifo_hashmapflag is a footgun. Passfalseand the output order flips. There is exactly one call site that passestruefor the standard build, and the flag is threaded throughupdate_hashfor the benefit of the other path. - The trick only works because the chain is LIFO. If someone “fixes” the hash map to use FIFO chains (say, to make the insertion order match the output order more directly), this reversal becomes wrong and the output order silently flips. There is no assertion that catches it.
- The reversed concatenation is confusing to read. The concatenated batch — the thing you index into — is not in the original order. A future maintainer who “tidies up” the concatenation to be in forward order will break the index arithmetic.
Why it matters
This is a small thing, and it is the kind of small thing that separates an engine that produces reproducible results from one that produces arbitrary ones. Two runs of the same query over the same data produce the same output, in the same order, even though the join is a hash join with an unspecified output order in the SQL standard.
And the way it is achieved is a nice illustration of a general principle: when a data structure’s natural iteration order is the opposite of what you want, you can either change the data structure (expensive, and it affects everything) or reverse the input order (free, and it affects only this one place).