Bitwise Semi Joins: Tracking Matches in a Bitset Instead of Materialising Row Pairs

TL;DR: A semi join (WHERE EXISTS (...)) answers a yes/no question per outer row. The naive sort-merge implementation evaluates the filter for every (outer, inner) pair and materialises the matching pairs. If a key group has a million inner rows and a million outer rows, that is 10^12 pairs. DataFusion’s BitwiseSortMergeJoinStream instead keeps a bit-packed bitmap of which outer rows matched, ORs the filter results into it 64 bits at a time, and counts the set bits with popcnt. The inner rows are streamed past, never materialised.

A semi join is WHERE EXISTS (subquery). An anti join is WHERE NOT EXISTS (subquery). A mark join is SELECT expr, EXISTS(...) FROM ....

All three have the same shape: for each outer row, determine whether some inner row matches. The answer is one bit.

The naive sort-merge implementation is a nested loop over key groups: for each outer key group, for each inner row in the matching inner key group, evaluate the filter, and if it passes, emit the (outer, inner) pair. For an inner join, that is exactly right — you need the pairs. For a semi join, it is absurd: you need one bit per outer row, and you are materialising pairs to get it.

The bitwise stream

datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs (1,328 lines) is the answer:

/// Sort-Merge join stream for Semi/Anti/Mark joins.
///
/// Named "bitwise" because it tracks outer-row matches via a per-batch
/// boolean bitset (`BooleanBufferBuilder`) rather than materializing
/// `(outer, inner)` row pairs. Filter results are OR'd into the bitset
/// in `u64` chunks, and emitting applies the bitset directly.
pub(crate) struct BitwiseSortMergeJoinStream {

and the bitset itself, with the reasons spelled out:

    // Per-outer-batch match tracking, reused across batches.
    // Bit-packed (not Vec<bool>) so that:
    //  - emit: finish() yields a BooleanBuffer directly (no packing iteration)
    //  - OR: apply_bitwise_binary_op ORs filter results in u64 chunks
    //  - count: UnalignedBitChunk::count_ones uses popcnt
    matched: BooleanBufferBuilder,

Three reasons, one per use. The bitset is not just a compact Vec<bool> — it is a BooleanBufferBuilder, so the output can be a BooleanBuffer directly, with no unpacking pass.

ORing 64 bits at a time

The core operation is merging the filter’s result into the accumulator:

// OR filter results into the matched bitset. Both sides are
// bit-packed [u8] buffers, so apply_bitwise_binary_op
// processes 64 bits per loop iteration (not 1 bit at a time).
//
// The offsets handle alignment: outer_offset is the bit
// position within matched where this key group starts,
// and filter_buf.offset() is the BooleanBuffer's internal
// bit offset (usually 0, but not guaranteed by Arrow).
let filter_buf = filter_result.values();
apply_bitwise_binary_op(
    matched.as_slice_mut(),
    outer_offset,
    filter_buf.inner().as_slice(),
    filter_buf.offset(),
    outer_group_len,
    |a, b| a | b,
);

apply_bitwise_binary_op is Arrow’s bitwise kernel: it applies a binary operation to two bit-packed buffers, 64 bits (one u64) per iteration. This is the same family of operation as Arrow’s and/or kernels on BooleanArray, and it is vectorised — the CPU processes a whole u64 of bits per instruction.

The offsets are the fiddly part, and the comment explains why they cannot be ignored: outer_offset is the bit position in the accumulator where this key group starts, and filter_buf.offset() is Arrow’s internal bit offset, “usually 0, but not guaranteed by Arrow”. Arrow’s BooleanBuffer can be a view into a larger buffer at a non-zero bit offset, and a naive implementation that assumes byte alignment produces garbage.

Counting with popcnt

How many outer rows have matched so far? You need to know, because the loop can stop early once every row in the group has matched:

// 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();

UnalignedBitChunk reads the bytes in place — zero-copy — and uses popcnt to count the set bits. popcnt is a single instruction on any CPU from the last two decades, so counting a million bits is a few thousand instructions.

The recount after each OR is the same:

// Recount matched bits after the OR. UnalignedBitChunk is
// zero-copy — it reads the bytes in place and uses popcnt.
matched_count =
    UnalignedBitChunk::new(self.matched.as_slice(), outer_offset, outer_group_len)
        .count_ones();

And the early exit:

while matched_count < outer_group_len {
    match spill_stream.next().await {
        ...

If every outer row in the group has already matched, there is no point reading the rest of the inner rows. The inner side is streamed past and discarded.

The nested-loop framing

The doc on the struct explains the terminology, which is helpful because sort-merge join is usually described as a merge rather than a nested loop:

    // Input streams — in the nested-loop model that sort-merge join
    // implements, "outer" is the driving loop and "inner" is probed for
    // matches. The existing MaterializingSortMergeJoinStream calls these "streamed"
    // and "buffered" respectively. For Left* joins, outer=left; for
    // Right* joins, outer=right. Output schema equals the outer side.
    outer: SendableRecordBatchStream,
    inner: SendableRecordBatchStream,

So for a RightSemi join, the right side is the “outer” and the output schema is the right side’s — which is why there is a flag:

    // Decomposed from JoinType: when RightSemi/RightAnti, outer=right,
    // inner=left, so we swap sides when building the filter batch.
    outer_is_left: bool,

The inner key group still has to be buffered

The bitset saves the output memory. It does not save the input memory, because to evaluate the filter for an outer row against an inner row, you need the inner row.

    // Inner key group buffer: all inner rows sharing the current join key.
    // Only populated when a filter is present. Unbounded — a single key
    // with many inner rows will buffer them all. See "Degenerate cases"
    // in exec.rs. On memory pool overflow the buffered slices move to a
    // per-group spill file (see [`Self::buffer_inner_key_group`]).
    inner_key_buffer: Vec<RecordBatch>,

“Unbounded — a single key with many inner rows will buffer them all.” And the escape hatch is a per-group spill file: when the memory pool overflows, the buffered slices are written to a spill file for that key group and read back when the filter is evaluated.

That is a nice design. The spill is scoped to a single key group rather than to the whole join, so at most one group’s worth of inner rows is ever on disk, and the spill file is deleted as soon as the group is done.

And the clock is handled carefully across the spill read:

// Note: the clock keeps running across the spill reads — the
// spill file is the join's own data, so reading it back is
// join work (unlike the child inputs' `next()`).

That is a subtle but correct distinction. The join_time metric is stopped while awaiting the child inputs — that is the operator waiting on its producer, not working — but it keeps running while reading back the join’s own spill file, because that is the join doing work.

The comparators are cached

    // Cached comparators — pre-built to avoid per-row type dispatch.
    /// Comparator for outer vs inner key comparison
    outer_inner_cmp: Option<JoinKeyComparator>,
    /// Comparator for outer self-comparison (find_key_group_end on outer)
    outer_self_cmp: Option<JoinKeyComparator>,
    /// Comparator for inner self-comparison (find_key_group_end on inner)
    inner_self_cmp: Option<JoinKeyComparator>,

Three comparators, cached. Building an Arrow comparator involves a downcast to the concrete array type and a closure; doing that per row would dominate the join. Caching them is the difference between a sort-merge join that is competitive with a hash join and one that is not.

And the group-boundary search is a binary search over the key:

let outer_group_end =
    find_key_group_end(self.get_outer_self_cmp()?, outer_group_start, num_outer);

which is a partition_point-style search — O(log n) instead of a linear scan to find where the key group ends.

The catch

• The bitset is per outer batch. It is sized to the current outer batch and reused across batches, so the memory is O(batch size), not O(rows). But it also means the “already matched” state does not survive a batch boundary — which is fine, because the state that matters is per key group, and a key group is within a batch.

• The inner key group is still unbounded. The bitset does not help with a key group with a million inner rows; that is what the per-group spill file is for.

• It only applies to semi/anti/mark joins. An inner join genuinely needs the pairs. The two implementations coexist:

  rust Copy

  unreachable!("Semi/anti/mark joins are handled by BitwiseSortMergeJoinStream")

  appears in the other stream, which refuses to handle them.

• The alignment handling is subtle. The filter_buf.offset() is “usually 0, but not guaranteed by Arrow”, and getting it wrong produces wrong results rather than a crash.

Why it matters

This is a good example of a general principle: when the answer is much smaller than the work, do not materialise the work.

A semi join’s answer is one bit per outer row. The naive implementation materialises a row pair per match — and in the worst case, a row pair per candidate, matched or not. The bitwise implementation turns the inner side from a source of memory pressure into a source of nothing at all: the inner rows are read, filtered, and discarded, and the only state that survives is a bitmap.

And the bitmap is not a Vec<bool> — it is bit-packed, so the OR is a 64-bit vectorised operation, the count is a popcnt, and the output is the buffer itself. Every part of it was chosen to make the common case free.