DataFusion Swaps Join Sides on Statistics and Applies Non-Equi Filters After the Hash Lookup

TL;DR: A hash join’s build side must fit in memory; the probe side is streamed. So which side is the build side? DataFusion’s JoinSelection physical rule compares total_byte_size (falling back to num_rows) of the two inputs and swaps them if the left is bigger. Swapping is not a free rename: the join type flips, the equi-key pairs flip, the residual filter’s column sides flip, and the output projection is restored. Separately, a join predicate that is not an equality (a.x < b.y) cannot be a hash key, so it becomes a JoinFilter that is evaluated after the hash lookup, on the index pairs that survived it.

Two different “swap” mechanisms

People use “swapped join” to mean two quite different things in DataFusion, and they live in the same file, datafusion/physical-optimizer/src/join_selection.rs:

Statistics-driven swap — the better side becomes the build side. This is a pure performance decision.
Unboundedness-driven swap — the finite side becomes the build side so the query can run at all. This is a correctness/runnability decision.

Both end up calling the same primitive, HashJoinExec::swap_inputs, and both have to undo a surprising amount of metadata surgery.

Statistics-driven swapping

The decision itself is tiny:

rust
// datafusion/physical-optimizer/src/join_selection.rs:84
pub(crate) fn should_swap_join_order(
    left: &Arc<dyn ExecutionPlan>,
    right: &Arc<dyn ExecutionPlan>,
    context: &dyn PhysicalOptimizerContext,
) -> Result<bool> {
    if !context.config_options().optimizer.join_reordering {
        return Ok(false);
    }

    let left_stats = get_stats(left, context)?;
    let right_stats = get_stats(right, context)?;

    // First compare total_byte_size, then fall back to num_rows if byte
    // sizes are unavailable.
    match (
        left_stats.total_byte_size.get_value(),
        right_stats.total_byte_size.get_value(),
    ) {
        (Some(l), Some(r)) => Ok(l > r),
        _ => match (
            left_stats.num_rows.get_value(),
            right_stats.num_rows.get_value(),
        ) {
            (Some(l), Some(r)) => Ok(l > r),
            _ => Ok(false),
        },
    }
}

Three things to notice, all of them load-bearing:

Bytes beat rows. A table with 10M narrow rows can be smaller than a table with 1M wide rows, and it is the bytes that have to fit in the memory pool. Comparing row counts would pick the wrong build side.
Unknown statistics mean no swap. Absent statistics yield false. DataFusion never guesses: if it does not know, it leaves the plan as the user wrote it.
It is opt-out-able. datafusion.optimizer.join_reordering (default true) disables the whole thing.

Note the deliberate conservatism: the rule compares sizes, not estimated cardinality after the join. It does not try to predict which side produces fewer output rows, because that estimate is far less reliable than a row count.

CollectLeft vs Partitioned: the other half of the decision

Swapping is only half of it. Even with the small side on the left, DataFusion has to decide whether to collect that side into one partition or partition it:

rust
// datafusion/physical-optimizer/src/join_selection.rs:192
pub(crate) fn try_collect_left(
    hash_join: &HashJoinExec,
    ignore_threshold: bool,
    context: &dyn PhysicalOptimizerContext,
) -> Result<Option<Arc<dyn ExecutionPlan>>> {
    ...
    let left_can_collect = ignore_threshold
        || supports_collect_by_thresholds(
            left,
            optimizer_config.hash_join_single_partition_threshold,
            optimizer_config.hash_join_single_partition_threshold_rows,
            context,
        );
    ...
    match (left_can_collect, right_can_collect) {
        (true, true)  => { /* swap if the right is smaller, then CollectLeft */ }
        (true, false) => { /* CollectLeft, no swap possible */ }
        (false, true) => { /* swap, then CollectLeft */ }
        (false, false) => Ok(None),  // fall through to partitioned
    }
}

The thresholds are datafusion.optimizer.hash_join_single_partition_threshold (4 MiB) and hash_join_single_partition_threshold_rows (128K rows). If the small side fits in one partition, PartitionMode::CollectLeft is used: the build side is gathered into a single partition, and every probe partition shares it. That is strictly better than N independent build sides when the build side is tiny — you hash the build rows once, not N times — but it serialises the build.

If neither side fits, PartitionMode::Partitioned is used and both sides are hash-repartitioned on the join key (see the sibling post on the partitioned hash table).

What `swap_inputs` actually has to fix

Swapping two inputs of a join is not symmetric. This is where most hand-rolled “swap the inputs” code goes wrong, and DataFusion’s implementation is a good catalogue of the cases:

rust
// datafusion/physical-plan/src/joins/hash_join/exec.rs:1521
    pub fn swap_inputs(
        &self,
        partition_mode: PartitionMode,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        assert_or_internal_err!(
            self.dynamic_filter.is_none(),
            "Cannot swap HashJoinExec inputs after dynamic filters have been constructed. \
             Optimizer rules that reorder join inputs must run before optimizer rules `FilterPushdown::new_post_optimization()`"
        );

        let left = self.left();
        let right = self.right();
        let new_join = self
            .builder()
            .with_type(self.join_type.swap())
            .with_new_children(vec![Arc::clone(right), Arc::clone(left)])?
            .with_on(
                self.on()
                    .iter()
                    .map(|(l, r)| (Arc::clone(r), Arc::clone(l)))
                    .collect(),
            )
            .with_filter(self.filter().map(JoinFilter::swap))
            .with_projection(swap_join_projection(
                left.schema().fields().len(),
                right.schema().fields().len(),
                self.projection.as_deref(),
                self.join_type(),
            ))
            .with_partition_mode(partition_mode)
            .build()?;

Five separate fixes:

The join type flips. Left becomes Right, LeftSemi becomes RightSemi, and so on. This is not cosmetic: Left and Right have completely different “emit unmatched rows at the end” behaviour, and the build side of a Right join is the side whose unmatched rows must be produced at the end.
The equi-key pairs flip. (l, r) becomes (r, l). Trivial but mandatory.
The residual filter flips. This is the subtle one:

rust
// datafusion/physical-plan/src/joins/join_filter.rs:91
    /// Rewrites the join filter if the inputs to the join are rewritten
    pub fn swap(&self) -> JoinFilter {
        let column_indices = self
            .column_indices()
            .iter()
            .map(|idx| ColumnIndex {
                index: idx.index,
                side: idx.side.negate(),
            })
            .collect();

        JoinFilter::new(
            Arc::clone(self.expression()),
            column_indices,
            Arc::clone(self.schema()),
        )
    }

The expression is unchanged — a.x < b.y is still a.x < b.y — but every column reference records which side it came from, and that side is negated. The JoinFilter keeps an explicit side per column precisely so this rewrite is mechanical rather than a re-planning exercise.

The projection is remapped, and then
the output order is restored:

rust
// datafusion/physical-plan/src/joins/hash_join/exec.rs:1554
        // Semi / anti / mark joins output columns from one side only, so
        // swapping does not reorder them; an embedded projection already fixes
        // the order too. Otherwise a projection restores the original order.
        if is_existence_join(self.join_type) || self.projection.is_some() {
            Ok(Arc::new(new_join))
        } else {
            reorder_output_after_swap(Arc::new(new_join), &left.schema(), &right.schema())
        }

A naive swap would emit the right side’s columns first. The user asked for left.*, right.*; the swapped plan naturally produces right.*, left.*, so a projection is bolted on to put them back. Semi/anti/mark joins are exempt because their output schema comes from one side only and does not move.

When swapping is forbidden

Not every join can be swapped:

rust
// datafusion/physical-optimizer/src/join_selection.rs:173
/// Determines whether it is possible to swap inputs of a hash join - for null-aware joins, we can only swap an uncorrelated `LeftAnti`
/// (a single join key and no filter), because the swapped `RightAnti` has no per-row NULL handling
fn can_swap_hash_join(hash_join: &HashJoinExec) -> bool {
    hash_join.join_type().supports_swap()
        && (!hash_join.null_aware
            || (*hash_join.join_type() == JoinType::LeftAnti
                && hash_join.on().len() == 1
                && hash_join.filter().is_none()))
}

A null-aware anti join (NOT IN against a nullable subquery) has per-row NULL semantics that only the LeftAnti implementation supports; the swapped RightAnti does not. And once a dynamic filter has been constructed for the join, the build side is already baked in — the assert_or_internal_err! above fires, which is why the JoinSelection rule is documented to run before the filter-pushdown rules.

The unboundedness swap: a different motivation

The other swap is not about performance at all. A hash join is a pipeline breaker: it must consume its build side completely before it can emit anything. If the build side is an unbounded stream, the query never produces a row. The fix is to make sure the finite side is the build side:

rust
// datafusion/physical-optimizer/src/join_selection.rs:524
pub fn hash_join_swap_subrule(
    mut input: Arc<dyn ExecutionPlan>,
    _config_options: &ConfigOptions,
) -> Result<Arc<dyn ExecutionPlan>> {
    if let Some(hash_join) = input.downcast_ref::<HashJoinExec>()
        && hash_join.dynamic_expressions_produced().is_empty()
        && hash_join.left.boundedness().is_unbounded()
        && !hash_join.right.boundedness().is_unbounded()
        && !hash_join.null_aware
        && matches!(
            *hash_join.join_type(),
            JoinType::Inner | JoinType::Left | JoinType::LeftSemi | JoinType::LeftAnti
        )
    {
        input = swap_join_according_to_unboundedness(hash_join)?;
    }
    Ok(input)
}

The join-type restriction is the interesting part: Right, RightSemi, RightAnti and Full cannot be made runnable by swapping, because the output schema of those joins is anchored to the right side. Swapping would change what the query returns. So for those, the plan stays broken (or the user gets a nested loop join). The comment in the source says it plainly:

rust
// datafusion/physical-optimizer/src/join_selection.rs:543
/// This function swaps sides of a hash join to make it runnable even if one of
/// its inputs are infinite. Note that this is not always possible; i.e.
/// [`JoinType::Full`], [`JoinType::Right`], [`JoinType::RightAnti`] and
/// [`JoinType::RightSemi`] can not run with an unbounded left side, even if
/// we swap join sides.

Now the residual filter: the predicate that is not a key

An equi-join predicate a.id = b.id is a hash key: DataFusion hashes the key columns and uses the hash to find candidates. A predicate like a.ts < b.ts or a.x + b.y > 10 cannot be a hash key — there is no single value to hash. Such predicates are separated out at planning time and become the join’s filter.

The physical planner counts the range predicates while it builds the filter:

rust
// datafusion/core/src/physical_planner.rs:1502
                let join_filter = match filter {
                    Some(expr) => {
                        let split_expr = split_conjunction(expr);
                        for expr in split_expr.iter() {
                            match *expr {
                                Expr::BinaryExpr(BinaryExpr { left: _, right: _, op }) => {
                                    if matches!(op,
                                        Operator::Lt | Operator::LtEq
                                        | Operator::Gt | Operator::GtEq)
                                    {
                                        range_filters.push((**expr).clone());
                                        num_range_filters += 1;
                                    }
                                    total_filters += 1;
                                }
                                _ => { total_filters += 1; }
                            }
                        }

and then builds a JoinFilter holding the physical expression, the list of columns it needs (each tagged with the side it came from), and the schema of the intermediate batch those columns form:

rust
// datafusion/core/src/physical_planner.rs:1595
                        let column_indices = join_utils::JoinFilter::build_column_indices(
                            left_field_indices,
                            right_field_indices,
                        );

                        Some(join_utils::JoinFilter::new(
                            filter_expr,
                            column_indices,
                            Arc::new(filter_schema),
                        ))

The trick: evaluate the filter on gathered index pairs

The filter is evaluated after the hash lookup, in the join stream, on the matched pairs:

rust
// datafusion/physical-plan/src/joins/hash_join/stream.rs:959
        // apply join filter if exists
        let (left_indices, right_indices) = if let Some(filter) = &self.filter {
            apply_join_filter_to_indices(
                build_side.left_data.batch(),
                &state.batch,
                left_indices,
                right_indices,
                filter,
                JoinSide::Left,
                None,
                self.join_type,
            )?
        } else {
            (left_indices, right_indices)
        };

and the implementation is the whole point:

rust
// datafusion/physical-plan/src/joins/utils.rs:1225
pub(crate) fn apply_join_filter_to_indices(
    build_input_buffer: &RecordBatch,
    probe_batch: &RecordBatch,
    build_indices: UInt64Array,
    probe_indices: UInt32Array,
    filter: &JoinFilter,
    build_side: JoinSide,
    max_intermediate_size: Option<usize>,
    join_type: JoinType,
) -> Result<(UInt64Array, UInt32Array)> {
    ...
        let intermediate_batch = build_batch_from_indices(
            filter.schema(),
            build_input_buffer,
            probe_batch,
            &build_indices,
            &probe_indices,
            filter.column_indices(),
            build_side,
            join_type,
            None,
        )?;

        filter
            .expression()
            .evaluate(&intermediate_batch)?
            .into_array(intermediate_batch.num_rows())?
    ...
    let mask = as_boolean_array(&filter_result)?;

    let left_filtered = compute::filter(&build_indices, mask)?;
    let right_filtered = compute::filter(&probe_indices, mask)?;

Read that carefully, because it is a genuinely nice piece of engineering:

The hash lookup produced two index arrays: build_indices (which build row) and probe_indices (which probe row), one entry per candidate pair.
A synthetic intermediate batch is materialized by gathering exactly the columns the filter mentions, from the build side and from the probe side, at those indices. This batch has one row per candidate pair, and the filter expression is evaluated on it as a normal Arrow expression — vectorised, no per-row interpreter.
The resulting boolean mask is applied to both index arrays, so the surviving pairs are the join output.

The upshot: the filter is not a per-pair interpreted predicate. It is a batch Arrow computation over a gathered batch. The cost is one gather of the filter’s columns plus one vectorised comparison, per chunk of candidate pairs.

Note the max_intermediate_size parameter: the intermediate batch is built in slices so that a huge fan-out does not materialise a huge intermediate batch:

rust
// datafusion/physical-plan/src/joins/utils.rs:1239
    let filter_result = if let Some(max_size) = max_intermediate_size {
        let mut filter_results =
            Vec::with_capacity(build_indices.len().div_ceil(max_size));

        for i in (0..build_indices.len()).step_by(max_size) {
            let end = min(build_indices.len(), i + max_size);
            let len = end - i;
            let intermediate_batch = build_batch_from_indices(...)?;

How the filter interacts with the probe side

Two details matter for the “does the filter change the probe semantics” question.

Semi/anti joins. The filter can turn a match into a non-match. For a LeftAnti join, “matched” means “matched and passed the filter”. The alignment logic that emits unmatched probe rows therefore has to run after the filter, and the code is explicit about it:

rust
// datafusion/physical-plan/src/joins/hash_join/stream.rs:983
        // The goals of index alignment for different join types are:
        //
        // 1) Right & FullJoin -- to append all missing probe-side indices between
        //    previous (excluding) and current joined indices.
        // 2) SemiJoin -- deduplicate probe indices in range between previous
        //    (excluding) and current joined indices.
        // 3) AntiJoin -- return only missing indices in range between previous
        //    and current joined indices.

and the “last joined probe index” is computed from the filtered indices:

rust
// datafusion/physical-plan/src/joins/hash_join/stream.rs:998
        // if any rows have been joined -- get last joined probe-side (right) row
        // it's important that index counts as "joined" after hash collisions checks
        // and join filters applied.

Visited bitmaps. For outer joins, the build-side “visited” bitmap is updated from the filtered build indices, so a build row that matched the hash but failed the residual filter is still correctly reported as unmatched:

rust
// datafusion/physical-plan/src/joins/hash_join/stream.rs:975
        // mark joined left-side indices as visited, if required by join type
        if need_produce_result_in_final(self.join_type) {
            let mut bitmap = build_side.left_data.visited_indices_bitmap().lock();
            left_indices.iter().flatten().for_each(|x| {
                bitmap.set_bit(x as usize, true);
            });
        }

Note that this code sits after the filter application in the stream, so left_indices here is the filtered array. Getting this ordering wrong is a classic source of “outer join returns the row twice” bugs.

Trade-offs and pathological cases

Statistics can be wrong. The swap is driven by total_byte_size from Statistics, which for a filtered scan can be a gross underestimate. A bad estimate does not merely make the query slow — it makes the build side exceed the memory pool and the query fails, because the hash join has no spill path. Setting datafusion.optimizer.join_reordering = false is the escape hatch.
The filter’s column set drives the intermediate batch size. A filter that references a wide string column from both sides forces a gather of that column for every candidate pair. With a high-fanout key, the gather dominates the join.
A filter that is nearly always false makes the join look like a hash join with extra steps. The hash lookup still produces all the candidates; the filter then discards them. If the filter is selective, a different plan (a nested loop join over a range, or a piecewise merge join) would be better — and DataFusion does have that path: a join with no equi keys and exactly one range predicate is routed to PiecewiseMergeJoinExec when datafusion.optimizer.enable_piecewise_merge_join is on.
Semi/anti joins with a filter are the pathological case for the chained fast path: the filter must be evaluated for every candidate pair, so the per-pair cost does not amortise the way it does for a plain inner join.
Swapping changes the output column order unless the projection is restored. If you write your own optimizer rule that swaps inputs, you inherit all five fixes above, not just the children.

The takeaway

The swap is a memory decision masquerading as a statistics decision: bytes on the build side must fit in the pool, twice, transiently. The residual filter is a post-lookup decision: the hash finds candidates, the filter disqualifies them, and the disqualification happens as a vectorised Arrow computation over a gathered intermediate batch rather than as a per-pair interpreted predicate. Both tricks are about the same thing — keeping the amount of work proportional to the output, rather than to the cross product.