Pruning Joins in DataFusion: Monotone Merge Cursors, a Key-Range Bitmap, and Watermarks
How DataFusion's sort-merge and piecewise merge joins skip work they cannot match: monotone key-group cursors, a one-bit-per-outer-row match bitset, a shared min_marked watermark, and a bucket bitmap over the build side's key range.
TL;DR: A hash join probes everything; a merge join does not. DataFusion’s sort-merge join keeps two cursors that only ever move forward, skipping whole key groups when the keys differ, and it never materialises
(left, right)pairs for semi/anti/mark joins — it keeps one bit per outer row instead. The piecewise merge join goes further: because a range predicate matches a suffix of the sorted build side, one extreme key and one atomic watermark (min_marked) replace a bitmap entirely. And on the hash-join side,KeyRangeBitmapsummarises the build side’s key range as one bit per bucket so the probe-side scan can skip whole containers.
First, a correction to the usual framing. There is no “chain of merge joins with
range pruning” in DataFusion, and key_range_bitmap.rs does not belong to the
sort-merge join — it belongs to the hash join’s dynamic-filter pushdown. What
exists is three separate pruning mechanisms, in three separate operators. They are
worth understanding together because they answer the same question in three
different ways: how do I avoid touching rows that cannot possibly match?
The question
A hash join’s cost is O(probe_rows × hash_lookup + matches). The hash lookup is
cheap, but it is not free, and — more importantly — the output is what kills you.
A hash join cannot spill: the build side must fit in the memory pool, twice,
transiently (see the sibling post on the partitioned hash table). If your build side
is 500 GB, the hash join is not an option at all.
The sort-merge join is the answer to that: it needs no hash table, it needs sorted inputs, and it can spill its buffered side to disk.
// datafusion/physical-plan/src/joins/sort_merge_join/exec.rs:966
let spill_manager = SpillManager::new(
context.runtime_env(),
SpillMetrics::new(metrics, partition),
buffered.schema(),
)
.with_compression_type(context.session_config().spill_compression());But a merge join has its own waste: if the two sides’ key ranges barely overlap, a naive implementation still walks every key group on both sides. The pruning mechanisms are about avoiding that.
Mechanism 1: the monotone merge cursor
The core of a sort-merge join is two cursors over sorted data, and the invariant that makes it fast: neither cursor ever moves backwards.
outer cursor ──► [1, 2, 2, 3, 5, 5, 7]
inner cursor ──► [2, 2, 4, 5, 6, 7, 7]
▲
compare keys at cursorsThe algorithm, as documented in the source:
- outer < inner: skip the outer key group (no match exists).
- outer > inner: skip the inner key group.
- outer == inner: process the match.
A key group is a contiguous run of equal keys on one side. Finding its end is optimised for the common case (groups of size 1):
// datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:170
fn find_key_group_end(cmp: &JoinKeyComparator, from: usize, len: usize) -> usize {
let next = from + 1;
if next >= len {
return len;
}
// Fast path: single-row group (common with unique keys).
if cmp.compare(from, next) != Ordering::Equal {
return next;
}
// Check if the entire remaining batch shares this key.
let last = len - 1;
if cmp.compare(from, last) == Ordering::Equal {
return len;
}
// Binary search the interior: key at `next` matches, key at `last` doesn't.
...That is a small but representative piece of engineering: check the two most likely
answers (group of 1, group of everything) before paying for a binary search. The
comparison itself is a JoinKeyComparator — a pre-built arrow_ord comparator that
avoids per-row type dispatch.
The pruning property falls out of monotonicity. When the outer key is less than the inner key, the entire outer key group can be discarded, because the inner cursor will only move forward and the inner side has no key equal to the outer group’s key anymore. No hash lookup, no candidate pairs, no output.
Key groups that span batches
Sorted data arrives in batches, and a key group can straddle a batch boundary. The stream handles this by remembering the last key of the current batch and checking whether the next batch starts with the same key:
// datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:618
async fn advance_inner_past_key_group(&mut self) -> Result<()> {
loop {
...
let group_end =
find_key_group_end(self.get_inner_self_cmp()?, from, num_inner);
if group_end < num_inner {
self.inner_offset = group_end;
return Ok(());
}
// Key group extends to the end of the batch — it may continue
// into the next one; save the last key so we can check.
let saved_inner_keys = slice_keys(&self.inner_key_arrays, num_inner - 1);
if !self.next_inner_batch().await? {
self.inner_batch = None;
return Ok(());
}
if !keys_match(
&saved_inner_keys,
&self.inner_key_arrays,
&self.sort_options,
self.null_equality,
)? {
return Ok(());
}
}
}Mechanism 2: the bitwise bitset (semi/anti/mark joins)
The general merge-join stream materialises (outer, inner) row pairs, applies the
filter, and then deduplicates with a “corrected filter mask”. For a semi/anti/mark
join that is pure waste: the question is not “which pairs match” but “does this outer
row have any match”.
So DataFusion has a second, specialised stream:
// datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:17
//! Sort-merge join stream specialized for semi/anti/mark joins.
//!
//! # Motivation
//!
//! The general-purpose `MaterializingSortMergeJoinStream`
//! handles semi/anti joins by materializing `(outer, inner)` row pairs,
//! applying a filter, then using a "corrected filter mask" to deduplicate.
//! Semi/anti joins only need a boolean per outer row (does a match exist?),
//! not pairs. The pair-based approach incurs unnecessary memory allocation
//! and intermediate batches.
//!
//! This stream instead tracks matches with a per-outer-batch bitset,
//! avoiding all pair materialization.
The bitset is one bit per outer row, and the filter results are OR’d into it with u64-chunked bitwise operations:
// datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:764
// Count already-matched bits using popcnt on u64 chunks (zero-copy).
let mut matched_count = UnalignedBitChunk::new(
self.matched.as_slice(),
outer_group_start,
outer_group_len,
)
.count_ones();and the loop short-circuits as soon as every outer row in the group has a match:
// datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:826
if matched_count == outer_group_len {
break 'outer;
}That short-circuit is the real win: for an existence join, once the answer is “yes” for every outer row in the group, the remaining inner rows in the group are irrelevant and are never touched. The join’s cost becomes proportional to the number of distinguishing inner rows, not to the size of the cross product.
Mechanism 3: the piecewise merge join and its watermark
PiecewiseMergeJoinExec handles the case where there is no equi-join key, only a
single range predicate (a.x < b.y). This is a different problem, and the
observation that makes it fast is beautiful:
Every match set is a suffix of the sorted buffered side’s non-null keys, so a streamed row matches anything at all exactly when it matches the last of them.
From the module documentation:
// datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs:109
/// extreme = buffer_batch[last non-null]
/// candidates = [row for row in stream_batch if compare(row, extreme)]
/// emit the other rows as unmatched (Right/Full) or drop them (Inner/Left)
///
/// start = 0
/// for stream_row in sort(candidates):
/// start = binary_search(buffer_batch[start:], first row matching stream_row)
/// output stream_row X buffer_batch[start:]So the algorithm is: compute the extreme (last non-null) key of the sorted buffered side once; test every streamed row against it with one vectorised comparison to decide which rows can match at all; sort only the survivors; then binary-search the first match for each, starting from where the previous row left off (the cursor is monotone again).
// datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs:627
/// Which rows of `stream_values` can match at least one buffered row.
///
/// Every match set is a suffix `[k, non_null_end)` of the sorted buffered side's non-null keys,
/// so a streamed row matches anything at all iff it matches the last of them, `buffered_extreme`
/// (None when no buffered key is non-null, so nothing can match). A NULL on either side
/// compares to NULL, which is no match.
fn matchable_rows(...)The watermark: a matched set as a single index
The existence variant (LeftSemi/LeftAnti/LeftMark) goes one step further. If
the matched set is always a suffix of the buffered side, you do not need a bitmap
to record it — you need exactly one integer:
// datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs:553
// Mark the matched buffered rows. `buffered_range` is always the suffix
// `[start, buffered_len)` -- a match emits every buffered row from the first match on --
// so the union of everything marked is `[min over matches, buffered_len)` and lowering a
// single watermark records it exactly. That replaces a mutex plus one `set_bit` per
// matched row, which was `O(buffered_len)` work for *every* matched streamed row.
//
// `buffer_idx` is monotone non-decreasing across a stream batch, so only the batch's
// first match can lower the watermark; `batch_min_marked` keeps the atomic off the hot
// path for all the others.
if need_produce_result_in_final(join_type) && buffered_range.0 < *batch_min_marked {
*batch_min_marked = buffered_range.0;
buffered_side
.buffered_data
.min_marked
.fetch_min(buffered_range.0, AtomicOrdering::SeqCst);
}fetch_min on an AtomicUsize is a lock-free, wait-free “lower the watermark”
operation. The watermark lives in the shared buffered data, so partitions benefit
from each other’s marking:
// datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs:59
/// `min_marked` lives in the shared buffered data, so partitions benefit from each other's
/// marking rather than each rediscovering it. It only ever decreases, so a partition that
/// reads a stale value scans a wider range than it had to -- never a narrower one.
and because the watermark only decreases, the search can stop at it:
// datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs:54
/// The search stops at `min_marked`: rows from there on were already marked, by this
/// partition or another, so a match found among them would record nothing. Marking is then a
/// single `fetch_min`, and finding the first match is a binary search, so a batch costs
/// `O(log buffered)` rather than `O(buffered)`.
That is a genuinely nice piece of engineering: the matched set is a single atomic integer, updated in O(1) per match, and the per-batch cost drops from O(buffered) to O(log buffered). A bitmap would have been a less compact encoding of one integer.
Mechanism 4: the key-range bitmap (on the hash-join side)
Back to the hash join. The build side’s keys are summarised by a bitmap of buckets over the key range:
// datafusion/physical-plan/src/joins/key_range_bitmap.rs:52
pub struct KeyRangeBitmap {
/// Smallest key, in the `u64` ordering the join's array map uses.
offset: u64,
/// `max - min`: a key is in range iff `key.wrapping_sub(offset) <= span`.
span: u64,
/// Each bucket covers `1 << shift` keys.
shift: u32,
/// One bit per bucket.
bits: BooleanBuffer,
}with the layout explained in the doc comment:
// datafusion/physical-plan/src/joins/key_range_bitmap.rs:42
/// keys 3 12 42 97
/// v v v v
/// bucket | 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 |
/// covers 3-18 19-34 35-50 51-66 67-82 83-98 unused unused
/// bit 1 0 1 0 0 1 0 0The probe then asks, “could this container hold a build key?”:
// datafusion/physical-plan/src/joins/key_range_bitmap.rs:127
pub fn may_contain_range(&self, lo: u64, hi: u64) -> bool {
// Offsets from the minimum; anything outside `[min, max]` wraps to
// something huge, which the clamping below turns into "past the end".
let lo_off = lo.wrapping_sub(self.offset);
let hi_off = hi.wrapping_sub(self.offset);
if lo_off > self.span && hi_off > self.span && lo_off <= hi_off {
return false;
}
...
self.bits
.slice(lo_b as usize, (hi_b - lo_b + 1) as usize)
.has_true()
}false is a proof that no build key lies in the container’s range; true only
means “maybe”. This is the same semantics as a Bloom filter, but exact per bucket.
The bitmap is not free, and the code bails out aggressively:
// datafusion/physical-plan/src/joins/key_range_bitmap.rs:28
/// Largest bitmap, in bits (128 KiB), which bounds what one build side can hold.
const MAX_BUCKETS: u64 = 1 << 20;// datafusion/physical-plan/src/joins/key_range_bitmap.rs:80
// One bucket per key value where the range allows it, so the bitmap is
// exact; otherwise as many as the cap permits.
let buckets = span.saturating_add(1).min(MAX_BUCKETS).next_power_of_two();
// Use a heuristic to bail-out for the common shape which renders each bucket
// set (contiguous surrogate keys, a date range) before actually computing it.
if (distinct_keys as u64).saturating_mul(2) > span.min(buckets) {
return None;
}Two bail-outs: (a) if the build side has more than half as many distinct keys as there are buckets, the bitmap would be saturated and could not exclude anything, so it is not built; (b) if every reachable bucket ends up set, the bitmap is useless and is discarded:
// datafusion/physical-plan/src/joins/key_range_bitmap.rs:94
// Every bucket set means nothing can be excluded.
let reachable = ((span >> shift) + 1) as usize;
bits.slice(0, reachable).has_false().then_some(Self { ... })Where is it used? Not for probing the hash table — for pruning the probe side’s scan. The build side reports the bitmap to the scan as a “may match” predicate:
// datafusion/pruning/src/pruning_predicate.rs:1553
/// Builds a "may match" expression for a [`HashTableLookupExpr`] (a large join
/// build side pushed down as an opaque hash-table lookup): tests container
/// min/max stats against the build side's key-range bitmap, which is fixed size
/// and built once with the join's hash table.
fn build_hash_lookup_pruning_expr(and the expression is evaluated per container:
// datafusion/pruning/src/key_range_bitmap_expr.rs:76
fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
let rows = batch.num_rows();
let min = self.min.evaluate(batch)?.into_array(rows)?;
let max = self.max.evaluate(batch)?.into_array(rows)?;
let matches = self.bitmap.may_contain_ranges(&min, &max);
Ok(ColumnarValue::Array(Arc::new(matches)))
}So: the hash join’s build side is pushed to the probe side as a dynamic filter, and
when the build side is too big to push as an IN list, it is pushed as an opaque
hash-table lookup plus this bitmap. The scan then skips row groups whose min/max
cannot contain any build key. This is the “prune probes that cannot match” idea in
its most literal form — and it lives in the hash join, not the sort-merge join.
Trade-offs and pathological cases
Sorted input is not free. The sort-merge join requires both inputs sorted on the join key. For a small table that is a
SortExecover the whole thing; for a large one it is a full external sort. This is why DataFusion defaults to the hash join:rust// datafusion/common/src/config.rs:1840 /// When set to true, the physical optimizer will prefer HashJoin over SortMergeJoin. /// HashJoin can work more efficiently than SortMergeJoin but consumes more memory pub prefer_hash_join: bool, default = trueSkewed keys destroy the merge join’s advantage. One hot key means one key group containing most of the data. The
find_key_group_endbinary search still finds it, but the output is a cross product. The bitwise stream’s filtered path has to buffer the whole inner key group in memory (or spill it), which is explicitly documented as a degenerate case:rust// datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:105 /// **Highly skewed key (filtered joins only):** When a filter is present, /// the inner key group is buffered so each inner row can be evaluated /// against the outer group. If one join key has N inner rows, all N rows /// are held in memory simultaneously (or spilled to disk if the memory /// pool limit is reached).The piece-wise merge join is experimental. It is off by default (
datafusion.optimizer.enable_piecewise_merge_join = false) and it only handles one range predicate with no equi-keys — the planner rejects anything else:rust// datafusion/core/src/physical_planner.rs:1639 } else if num_range_filters == 1 && total_filters == 1 && session_state .config_options() .optimizer .enable_piecewise_merge_joinThe key-range bitmap only works for integers.
ArrayMap::is_supported_typegates it toInt8..UInt64. For string keys there is no ordering to bucket by, so the bitmap does not exist and the probe side gets no container pruning from it.The bitmap’s usefulness depends on the key distribution. A contiguous surrogate-key range sets every bucket and the bitmap is discarded (the
has_falsebail-out). A sparse key set with holes is the ideal case — the test in the source builds exactly that shape:rust// datafusion/physical-plan/src/joins/key_range_bitmap.rs:198 fn prunes_gaps_and_never_drops_a_match() { let keys: Vec<i64> = (0..2_000_000).filter(|k| k % 10000 < 200).collect();False negatives are impossible, false positives are free. The bitmap is a summary, not a filter: it can only ever say “maybe”. The correctness of the join never depends on it.
The takeaway
All three mechanisms share a philosophy: do not touch what cannot match. The
merge join prunes with monotone cursors over key groups; the bitwise stream prunes
with a bitset and a popcnt short-circuit; the piecewise merge join prunes with an
extreme key and an atomic watermark; the hash join prunes the scan with a bucket
bitmap over the build side’s key range.
The unifying trick is representing a matched set as the smallest possible object: a suffix of a sorted array is one integer, not a bitmap. When you can prove your matched set has structure, do not pay for a general-purpose data structure.