DataFusion's Symmetric Hash Join Builds a Hash Table on Both Sides and Prunes the Rest

TL;DR: The regular hash join is a pipeline breaker: it must consume its build side completely before it emits anything. For two unbounded streams that never happens. DataFusion’s SymmetricHashJoinExec therefore builds a hash table on both sides, and whichever side produces a batch probes the other’s table. Because both sides grow forever, the operator prunes: interval arithmetic on the join filter tells it which buffered rows can no longer match, and those rows are dropped from the buffer and from the hash table. Two tables and two buffers — twice the memory of a classic hash join — but no pipeline break and no unbounded growth.

The problem: a hash join is a pipeline breaker

Every hash join has the same shape: drain the build side, build a table, then stream the probe side through it. That works fine for tables. It does not work for a Kafka topic or a generate_series that never ends, because the build side never finishes and the join never emits a row.

DataFusion’s physical optimizer has a dedicated rule for this situation, and it is worth reading because it shows exactly when the symmetric join is (and is not) allowed:

rust
// datafusion/physical-optimizer/src/join_selection.rs:384
fn hash_join_convert_symmetric_subrule(
    input: Arc<dyn ExecutionPlan>,
    config_options: &ConfigOptions,
) -> Result<Arc<dyn ExecutionPlan>> {
    // Check if the current plan node is a HashJoinExec.
    if let Some(hash_join) = input.downcast_ref::<HashJoinExec>() {
        let left_unbounded = hash_join.left.boundedness().is_unbounded();
        let left_incremental = matches!(
            hash_join.left.pipeline_behavior(),
            EmissionType::Incremental | EmissionType::Both
        );
        ...
        // Process only if both left and right sides are unbounded and incrementally emit.
        if left_unbounded && right_unbounded & left_incremental & right_incremental {
            // Determine the partition mode based on configuration.
            let mode = if config_options.optimizer.repartition_joins {
                StreamJoinPartitionMode::Partitioned
            } else {
                StreamJoinPartitionMode::SinglePartition
            };

Three conditions: both sides unbounded, both sides incremental, and the plan must be a HashJoinExec. Note that it is both sides that must be unbounded — if only one side is infinite, the classic hash join can simply be swapped (see the sibling post on side swapping) and the problem disappears.

The mode is chosen by optimizer.repartition_joins: in Partitioned mode both streams are hash-repartitioned on the join key, so each partition only ever sees the keys it needs:

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

That is where the parallelism comes from: repartitioning both streams by join key means partition i of the left only has to be compared against partition i of the right. The symmetric join’s cost is therefore the same per-partition cost as a classic hash join — it just has two build sides instead of one.

Two hash tables, one per side

The state is symmetric by construction:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1463
pub struct OneSideHashJoiner {
    /// Build side
    build_side: JoinSide,
    /// Input record batch buffer
    pub input_buffer: RecordBatch,
    /// Columns from the side
    pub(crate) on: Vec<PhysicalExprRef>,
    /// Hashmap
    pub(crate) hashmap: PruningJoinHashMap,
    /// Reuse the hashes buffer
    pub(crate) hashes_buffer: Vec<u64>,
    /// Matched rows
    pub(crate) visited_rows: HashSet<usize>,
    /// Offset
    pub(crate) offset: usize,
    /// Deleted offset
    pub(crate) deleted_offset: usize,
}

Each joiner owns: a buffer of everything it has seen, a hash table over that buffer, a scratch buffer for hashes, a set of rows that have been matched (for existence joins), and two cursors — offset (rows appended) and deleted_offset (rows pruned from the front).

Every incoming batch is appended to the buffer and hashed in:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1530
    pub(crate) fn update_internal_state(
        &mut self,
        batch: &RecordBatch,
        random_state: &RandomState,
        null_equality: NullEquality,
    ) -> Result<()> {
        // Merge the incoming batch with the existing input buffer:
        self.input_buffer = concat_batches(&batch.schema(), [&self.input_buffer, batch])?;
        // Resize the hashes buffer to the number of rows in the incoming batch:
        self.hashes_buffer.resize(batch.num_rows(), 0);
        // Get allocation_info before adding the item
        // Update the hashmap with the join key values and hashes of the incoming batch:
        update_hash(
            &self.on,
            batch,
            &mut self.hashmap,
            self.offset,
            random_state,
            &mut self.hashes_buffer,
            self.deleted_offset,
            false,
            null_equality,
        )?;
        Ok(())
    }

Note concat_batches on every batch: the buffer is one contiguous RecordBatch, because the hash table’s chain indices must be valid indices into it. (This is O(total buffered rows) per batch — a real cost, discussed below.)

Both sides probe

The stream is a small state machine over the two inputs, and the interesting part is that the roles are swapped per batch. Whichever side produced the batch is the probe side; the other side is the build side:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:2014
    fn perform_join_for_given_side(
        &mut self,
        probe_batch: &RecordBatch,
        probe_side: JoinSide,
    ) -> Result<Option<RecordBatch>> {
        let (probe_hash_joiner, build_hash_joiner, ...) =
            if probe_side.eq(&JoinSide::Left) {
                (&mut self.left, &mut self.right, ...)
            } else {
                (&mut self.right, &mut self.left, ...)
            };
        ...
        // Update the internal state of the hash joiner for the build side:
        probe_hash_joiner.update_internal_state(
            probe_batch,
            &self.random_state,
            self.null_equality,
        )?;
        // Join the two sides:
        let equal_result = join_with_probe_batch(
            build_hash_joiner,
            probe_hash_joiner,
            ...

So the flow for a batch arriving on the left is: append it to the left buffer, then use the right buffer’s hash table to find the left batch’s matches. Both tables are live at all times, and both grow.

text
   left stream ──► [ left buffer + left hash table ] ─┐
                                                      ├── probe each other
  right stream ──► [right buffer + right hash table ] ─┘

join_with_probe_batch is the classic build/probe lookup, with the symmetric join’s own hash map type:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1279
pub(crate) fn join_with_probe_batch(
    build_hash_joiner: &mut OneSideHashJoiner,
    probe_hash_joiner: &mut OneSideHashJoiner,
    ...
) -> Result<Option<RecordBatch>> {
    if build_hash_joiner.input_buffer.num_rows() == 0 || probe_batch.num_rows() == 0 {
        return Ok(None);
    }
    let (build_indices, probe_indices) = lookup_join_hashmap(
        &build_hash_joiner.hashmap,
        &build_hash_joiner.input_buffer,
        probe_batch,
        ...

For existence joins (semi/anti/mark), no output is produced during the scan; instead the matched rows are recorded in a HashSet<usize>:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1322
    if need_to_produce_result_in_final(build_hash_joiner.build_side, join_type) {
        record_visited_indices(
            &mut build_hash_joiner.visited_rows,
            build_hash_joiner.deleted_offset,
            &build_indices,
        );
    }

and the unmatched rows are produced at the very end, when both streams are exhausted:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1924
    fn process_batches_before_finalization(
        &mut self,
    ) -> Result<StatefulStreamResult<Option<RecordBatch>>> {
        // Get the left side results:
        let left_result = build_side_determined_results(
            &self.left, &self.schema, self.left.input_buffer.num_rows(), ...

That is the price of a symmetric join on an existence join: LeftSemi/LeftAnti results are only available at the end of the stream, never incrementally. The config option says so out loud:

rust
// datafusion/common/src/config.rs:1724
/// Should DataFusion allow symmetric hash joins for unbounded data sources even when
/// its inputs do not have any ordering or filtering If the flag is not enabled,
/// the SymmetricHashJoin operator will be unable to prune its internal buffers,
/// resulting in certain join types - such as Full, Left, LeftAnti, LeftSemi, Right,
/// RightAnti, and RightSemi - being produced only at the end of the execution.
/// This is not typical in stream processing. Additionally, without proper design for
/// long runner execution, all types of joins may encounter out-of-memory errors.
pub allow_symmetric_joins_without_pruning: bool, default = true

The pruning: how a growing buffer stays bounded

Both buffers grow without bound. The only thing that stops them is pruning: rows that provably cannot match any future row are dropped from the front of the buffer and from the hash table.

The pruning signal comes from the join filter’s range component. Consider the classic streaming join:

sql
SELECT * FROM left_stream, right_stream
ON left_key = right_key
   AND left_ts > right_ts - INTERVAL '12 minutes'
   AND left_ts < right_ts + INTERVAL '2 hours'

Because both sides are sorted by ts, once the right stream has advanced past T, any left row with ts < T - 12 minutes can never match again. DataFusion computes this with interval arithmetic over the filter’s sorted expressions:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1567
    pub(crate) fn calculate_prune_length_with_probe_batch(
        &mut self,
        build_side_sorted_filter_expr: &mut SortedFilterExpr,
        probe_side_sorted_filter_expr: &mut SortedFilterExpr,
        graph: &mut ExprIntervalGraph,
    ) -> Result<usize> {
        ...
        // Update the physical expression graph using the join filter intervals:
        graph.update_ranges(&mut filter_intervals, Interval::TRUE)?;
        // Extract the new join filter interval for the build side:
        let calculated_build_side_interval = filter_intervals.remove(0).1;
        // If the intervals have not changed, return early without pruning:
        if calculated_build_side_interval.eq(build_side_sorted_filter_expr.interval()) {
            return Ok(0);
        }
        // Update the build side interval and determine the pruning length:
        build_side_sorted_filter_expr.set_interval(calculated_build_side_interval);

        determine_prune_length(&self.input_buffer, build_side_sorted_filter_expr)

The prune length itself is found by a binary search over the sorted buffer, because the buffer is sorted on the filter’s column:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1042
fn determine_prune_length(
    buffer: &RecordBatch,
    build_side_filter_expr: &SortedFilterExpr,
) -> Result<usize> {
    ...
    // Perform binary search on the array to determine the length of the record batch to be pruned
    bisect::<true>(&[batch_arr], &[target], &[origin_sorted_expr.options])
}

Then the rows are physically removed from both the buffer and the hash table:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1600
    pub(crate) fn prune_internal_state(&mut self, prune_length: usize) -> Result<()> {
        // Prune the hash values:
        self.hashmap.prune_hash_values(
            prune_length,
            self.deleted_offset as u64,
            HASHMAP_SHRINK_SCALE_FACTOR,
        );
        // Remove pruned rows from the visited rows set:
        for row in self.deleted_offset..(self.deleted_offset + prune_length) {
            self.visited_rows.remove(&row);
        }
        // Update the input buffer after pruning:
        self.input_buffer = self
            .input_buffer
            .slice(prune_length, self.input_buffer.num_rows() - prune_length);
        // Increment the deleted offset:
        self.deleted_offset += prune_length;
        Ok(())
    }

The hash table has to be prunable too

This is the part that makes the symmetric join different from the classic one. The classic hash table stores row_index + 1 in the map and the chain in a flat Vec<u32>; pruning the front of the build side would invalidate every index. The symmetric join’s map uses a different structure for exactly this reason:

rust
// datafusion/physical-plan/src/joins/stream_join_utils.rs:157
pub struct PruningJoinHashMap {
    /// Stores hash value to last row index
    pub map: HashTable<(u64, u64)>,
    /// Stores indices in chained list data structure
    pub next: VecDeque<u64>,
}

A VecDeque can be drained from the front in O(1) amortised per element — the chain indices stay valid because they are relative to deleted_offset. Then the map entries whose chain has been entirely pruned are removed:

rust
// datafusion/physical-plan/src/joins/stream_join_utils.rs:224
    pub(crate) fn prune_hash_values(
        &mut self,
        prune_length: usize,
        deleting_offset: u64,
        shrink_factor: usize,
    ) {
        // Remove elements from the list based on the pruning length.
        self.next.drain(0..prune_length);

        // Calculate the keys that should be removed from the map.
        let removable_keys = self
            .map
            .iter()
            .filter_map(|(hash, tail_index)| {
                (*tail_index < prune_length as u64 + deleting_offset).then_some(*hash)
            })
            .collect::<Vec<_>>();

        // Remove the keys from the map.
        for hash_value in removable_keys {
            self.map
                .find_entry(hash_value, |(hash, _)| hash_value == *hash)
                .unwrap()
                .remove();
        }

        // Shrink the map if necessary.
        self.shrink_if_necessary(shrink_factor);
    }

and the map is shrunk so it does not keep the memory of pruned rows:

rust
// datafusion/physical-plan/src/joins/stream_join_utils.rs:193
    pub(crate) fn shrink_if_necessary(&mut self, scale_factor: usize) {
        let capacity = self.map.capacity();

        if capacity > scale_factor * self.map.len() {
            let new_capacity = (capacity * (scale_factor - 1)) / scale_factor;
            // Resize the map with the new capacity.
            self.map.shrink_to(new_capacity, |(hash, _)| *hash)
        }
    }

HASHMAP_SHRINK_SCALE_FACTOR is 4, so the map is shrunk when it is four times larger than its live content — aggressive enough to release memory, lazy enough not to rehash on every prune.

The ordering subtlety

The symmetric join inserts rows into the map in LIFO order (the newest row becomes the chain head), which reverses the order of matches within a chain. The code documents the problem and the fix at length:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1380
    // As SymmetricHashJoin uses LIFO JoinHashMap, the chained list algorithm
    // will return build indices for each probe row in a reverse order as such:
    // Build Indices: [5, 4, 3]
    // Probe Indices: [1, 1, 1]
    //
    // This affects the output sequence. ...
    //
    // Probe rows whose key contains a NULL cannot match any build row and are
    // skipped without a map lookup.
    let valid_keys = matchable_join_keys(&keys_values, null_equality);
    let (mut matched_probe, mut matched_build) = build_hashmap.get_matched_indices(
        Box::new(
            hash_values
                .iter()
                .enumerate()
                .filter(|(i, _)| {
                    valid_keys.as_ref().is_none_or(|valid| valid.is_valid(*i))
                })
                .rev(),
        ),
        deleted_offset,
    );

    matched_probe.reverse();
    matched_build.reverse();

The probe iteration is reversed and the results are reversed back, which restores lexicographic order on both sides. Two reversals instead of one because the chain itself is LIFO — a small thing, but it is the kind of thing that produces subtly wrong output ordering if you “simplify” it.

The “join with a small side” case

There is no special “small side” branch inside SymmetricHashJoinExec — and that is the correct design, because the symmetric join is only ever planned when both sides are unbounded. The small-side case is handled by a different operator entirely:

One side bounded → the classic HashJoinExec, with the bounded (small) side as the build side, PartitionMode::CollectLeft if it fits in one partition. The unbounded side is the probe side and streams through.
One side unbounded, one side unbounded → SymmetricHashJoinExec.
Both sides bounded → the classic hash join, usually Partitioned.

So the “small side” special case is a planner decision, made long before the symmetric join exists. The symmetric join’s own degenerate case is simply “the other side’s buffer is empty”, and it short-circuits:

rust
// datafusion/physical-plan/src/joins/symmetric_hash_join.rs:1290
    if build_hash_joiner.input_buffer.num_rows() == 0 || probe_batch.num_rows() == 0 {
        return Ok(None);
    }

There is one more small-side-shaped behaviour inside the symmetric join, though: the pruning is effectively a small-side optimisation. If the filter’s range window is narrow (say ±1 minute), the pruned prefix of the buffer is everything older than a minute, so the “live” buffer is small even though the stream is infinite. The tighter the range predicate, the smaller the retained buffer. That is the whole reason the operator insists on ordered inputs and a range filter.

Trade-offs and pathological cases

Memory is doubled, and there is no spill. Unlike the sort-merge join, the symmetric hash join has no SpillManager. Its only defence against unbounded growth is pruning. If the join has no range filter — or the range filter does not imply a bound — the buffers grow forever and the query dies with an out-of-memory error. The allow_symmetric_joins_without_pruning option exists precisely because of this, and its documentation says the quiet part out loud: “without proper design for long runner execution, all types of joins may encounter out-of-memory errors.”
The buffer concatenation is quadratic. update_internal_state calls concat_batches on the whole buffer for every incoming batch, so buffering n rows in k batches costs O(n·k) copying. With small batches from a chatty source this dominates. A ring buffer or a segmented buffer (like the LogicalBatch used by the nested-loop join) would fix it; the current code does not.
Existence joins produce nothing until the end. LeftSemi, LeftAnti and LeftMark on symmetric hash join are end-of-stream operators. In a streaming context that means no output until the input ends — which for an infinite input means never. This is why the optimizer’s determine_order closure goes to some trouble to give the symmetric join the sort orderings it needs for pruning (datafusion/physical-optimizer/src/join_selection.rs:417): without ordering there is no pruning, and without pruning there is no early output.
The visited_rows HashSet is a memory tax. Every matched row is recorded in a hash set so the unmatched ones can be emitted later. For a high-cardinality existence join over a long stream, that set is the dominant memory consumer — larger than the hash table itself.
Partitioned mode costs a repartition. StreamJoinPartitionMode::Partitioned requires both inputs to be hash-repartitioned on the join key, which is a full shuffle of two unbounded streams. SinglePartition avoids the shuffle but serialises everything. optimizer.repartition_joins = false picks the latter.
The symmetric join is not a general-purpose join. It exists for the unbounded/unbounded case. Using it for bounded inputs would be strictly worse than the classic hash join: same memory, more work (both sides build tables), worse output latency.

The takeaway

The symmetric hash join is what a hash join looks like when you remove the assumption that one side will end. That assumption is load-bearing in the classic implementation: it is what lets the build side be a single contiguous batch with stable indices. Remove it and you need a hash table that can be pruned from the front, buffers that can be truncated, and a matching set that survives truncation — which is exactly what PruningJoinHashMap, deleted_offset and the visited_rows set provide.

The cost is honest and worth stating plainly: two hash tables, two buffers, a HashSet of visited rows, and a quadratic concatenation per batch. The benefit is that a query over two infinite streams produces results, and keeps producing them, as long as the filter has a range that lets the operator throw away the past.