HashJoin vs SortMergeJoin vs NestedLoopJoin: What DataFusion Actually Uses Statistics For

TL;DR: There is no cardinality-driven algorithm choice in DataFusion’s planner. HashJoin, SortMergeJoin, NestedLoopJoin and CrossJoin are selected by the shape of the join condition: equi-keys present or not, a range filter or not, prefer_hash_join on or off. What the statistics actually decide is (a) which side becomes the hash table, (b) whether that side is collected into one partition or repartitioned, and (c) how many rows the join will emit.

I want to start with a correction, because it is the single most common misconception about DataFusion’s optimizer.

There is no datafusion/optimizer/src/join_reorder.rs in DataFusion. There is no ballistic join enumerator. There is no exhaustive search over join orders. The logical optimizer’s rule list is right there in the source:

rust
// datafusion/optimizer/src/optimizer.rs:289-317
Arc::new(RewriteSetComparison::new()),
Arc::new(OptimizeUnions::new()),
Arc::new(UnionsToFilter::new()),
Arc::new(SimplifyExpressions::new()),
Arc::new(ReplaceDistinctWithAggregate::new()),
Arc::new(EliminateJoin::new()),
Arc::new(DecorrelatePredicateSubquery::new()),
Arc::new(ScalarSubqueryToJoin::new()),
Arc::new(DecorrelateLateralJoin::new()),
Arc::new(ExtractEquijoinPredicate::new()),
Arc::new(EliminateDuplicatedExpr::new()),
Arc::new(EliminateFilter::new()),
Arc::new(EliminateCrossJoin::new()),
Arc::new(EliminateLimit::new()),
Arc::new(PropagateEmptyRelation::new()),
Arc::new(FilterNullJoinKeys::default()),
Arc::new(EliminateOuterJoin::new()),
...

No join reorderer. No cost model over join trees. DataFusion will happily execute A ⋈ B ⋈ C ⋈ D in exactly the order you wrote it, forever, no matter how many statistics it has.

What does exist is the physical optimizer rule JoinSelection, and it is much more modest than the name suggests. But before we get to it, we have to talk about how the algorithm is chosen in the first place — because the answer is surprising.

The algorithm decision is structural, not statistical

Here is the whole decision, from the physical planner:

rust
// datafusion/core/src/physical_planner.rs:1630-1652 (excerpt)
let join: Arc<dyn ExecutionPlan> = if join_on.is_empty() {
    if join_filter.is_none() && *join_type == JoinType::Inner {
        // cross join if there is no join conditions and no join filter set
        Arc::new(CrossJoinExec::new(physical_left, physical_right))
    } else if num_range_filters == 1
        && total_filters == 1
        && session_state
            .config_options()
            .optimizer
            .enable_piecewise_merge_join
    {
        ... PiecewiseMergeJoinExec ...
    } else {
        // there is no equal join condition, use the nested loop join
        Arc::new(NestedLoopJoinExec::try_new(
            physical_left, physical_right, join_filter, join_type, None,
        )?)
    }
} else if session_state.config().target_partitions() > 1
    && session_state.config().repartition_joins()
    && !prefer_hash_join
    && !*null_aware
{
    // Use SortMergeJoin if hash join is not preferred
    let join_on_len = join_on.len();
    Arc::new(SortMergeJoinExec::try_new(
        physical_left, physical_right, join_on, join_filter,
        *join_type, vec![SortOptions::default(); join_on_len], *null_equality,
    )?)
} else {
    let partition_mode = if can_repartition_join {
        PartitionMode::Auto
    } else {
        PartitionMode::CollectLeft
    };
    Arc::new(
        HashJoinExecBuilder::new(physical_left, physical_right, join_on, *join_type)
            .with_filter(join_filter)
            .with_partition_mode(partition_mode)
            .with_null_equality(*null_equality)
            .with_null_aware(*null_aware)
            .with_null_aware_value_keys(*null_aware_value_keys)
            .build()?,
    )
};

Read it as a decision tree:

text

                     join_on.is_empty()?
                    /                  \
                 yes                    no
                  |                     |
    join_filter.is_none()          repartition_joins && target_partitions>1
    && join_type == Inner?         && !prefer_hash_join && !null_aware?
         /        \                        /              \
       yes        no                     yes             no
        |          |                      |               |
  CrossJoin   range filter?         SortMergeJoin      HashJoin
                exactly 1?          (sorts both sides)
              /         \
            yes          no
             |            |
        PWMJ (opt-in)  NestedLoopJoin

No cardinality appears anywhere in that tree. The inputs are:

join_on — the extracted equi-join keys.
join_filter — the residual predicate.
num_range_filters / total_filters — how many of the residual conjuncts are range comparisons.
prefer_hash_join — datafusion.optimizer.prefer_hash_join, default true.
repartition_joins — datafusion.optimizer.repartition_joins, default true.
target_partitions > 1.

So: equi-keys present → hash join (by default). No equi-keys → cross join if there’s also no residual predicate, otherwise nested loop. Sort-merge join is only reachable by explicitly setting prefer_hash_join = false.

That’s a defensible design. Hash join is the right default for OLAP: it’s O(n+m) with a single pass and no sorting, whereas sort-merge join pays a full sort of both inputs. But it means “which join algorithm” is not a cost-based question in DataFusion. The cost-based questions are downstream of it.

Why the range filter branch is so specific

The PiecewiseMergeJoin branch requires exactly one residual filter and that filter being a range comparison:

rust
// datafusion/core/src/physical_planner.rs:1497-1531 (excerpt)
let mut num_range_filters = 0;
let mut range_filters: Vec<Expr> = Vec::new();
let mut total_filters = 0;

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;
                }
                // TODO: Want to deal with `Expr::Between` for IEJoins, it counts as two range predicates
                // which is why it is not dealt with in PWMJ
                // Expr::Between(_) => {},
                _ => { total_filters += 1; }
            }
        }
        ...

The TODO about Between is a nice detail: a BETWEEN 1 AND 10 is two range predicates, so it is not eligible for PWMJ. (And of course, by the time it reaches the physical planner, the expression simplifier has already desugared Between into >= and <= — so the Expr::Between arm is mostly dead code, which is exactly the kind of leftover you find in a codebase that has been refactoring this area for years.)

PWMJ also has to pin one operand to each side:

rust
// datafusion/core/src/physical_planner.rs:1668-1720 (excerpt)
// `Neither` covers an operand that references no column from
// either side (e.g. a literal), and `Both` an operand that
// references columns from both. PWMJ needs one operand pinned
// to each side, so both fall back to NestedLoopJoin below rather
// than erroring or (for `Neither`) panicking.
enum Side { Left, Right, Both, Neither }

let side_of = |e: &Expr| -> Side {
    let cols = e.column_refs();
    let any_left = cols.iter().any(|c| left_df_schema.index_of_column(c).is_ok());
    let any_right = cols.iter().any(|c| right_df_schema.index_of_column(c).is_ok());
    match (any_left, any_right) {
        (true, false) => Side::Left,
        (false, true) => Side::Right,
        (true, true) => Side::Both,
        (false, false) => Side::Neither,
    }
};
...
} else if !(left_side == Side::Left && right_side == Side::Right) {
    // Anything other than a clean left/right split -- both
    // operands on one side, one referencing neither side, or
    // either referencing both -- isn't a range predicate PWMJ
    // can plan, so let NestedLoopJoin evaluate it instead.
    return Ok(Arc::new(NestedLoopJoinExec::try_new(...)?));
}

This is the shape of a mature planner: it recognizes “I could use a cleverer algorithm here” and then carefully declines to, falling back to the general one. PWMJ is off by default anyway (enable_piecewise_merge_join = false).

Null-aware joins are pinned

There’s a nice defensive check here. NOT IN over a nullable subquery has null-aware anti-join semantics, and only HashJoinExec implements them:

rust
// datafusion/core/src/physical_planner.rs:1611-1627
// Null-aware joins are pinned to CollectLeft hash joins (see
// `HashJoinExec::null_aware`): never repartition them, and
// never route them to the sort-merge path below.
let can_repartition_join = session_state.config().target_partitions() > 1
    && session_state.config().repartition_joins()
    && !*null_aware;

// Only `HashJoinExec` implements null-aware semantics, and it
// needs equi-join keys to do so. Without them the join would be
// planned as a nested loop (or piecewise merge) join, which
// silently ignores the flag and returns wrong results for
// `NOT IN` over a nullable subquery. Fail loudly instead.
if *null_aware && join_on.is_empty() {
    return plan_err!(
        "null_aware {join_type} join requires equi-join keys, but the join has none"
    );
}

“Fail loudly instead” is the right call: the alternative is a silently wrong answer.

Where statistics do matter: `JoinSelection`

Now we get to the physical optimizer rule that people usually mean when they say “DataFusion’s join optimizer”. It runs here:

rust
// datafusion/physical-optimizer/src/optimizer.rs:213
Arc::new(JoinSelection::new()),

and its doc comment is admirably precise about its scope:

rust
// datafusion/physical-optimizer/src/join_selection.rs:70-83
/// Checks whether join inputs should be swapped using available statistics.
///
/// It follows these steps:
/// 1. If a [`datafusion_physical_plan::operator_statistics::StatisticsRegistry`] is
///    provided, use it for cross-operator estimates
///    (e.g., intermediate join outputs that would otherwise have `Absent` statistics).
/// 2. Compare the in-memory sizes of both sides, and place the smaller side on the
///    left (build) side.
/// 3. If in-memory byte sizes are unavailable, fall back to row counts.
/// 4. Do not reorder the join if neither statistic is available, or if
///    `datafusion.optimizer.join_reordering` is disabled.
///
/// Used configurations
/// - `optimizer.join_reordering`: allows or forbids statistics-driven join swapping
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),
        },
    }
}

That’s the whole “reordering” story: compare the two inputs of one join and put the smaller one on the left. Byte size first, row count as fallback, and if neither is known, do nothing. Note the semantics: Ok(l > r) means “the left is bigger, so swap”.

This is not join-order reordering across a join tree. It is build-side selection for a single join. The config name (join_reordering) oversells it.

`CollectLeft` vs `Partitioned`: the real cost decision

This is where the statistics earn their keep. HashJoinExec has three partition modes:

rust
// datafusion/physical-plan/src/joins/mod.rs:119-128
pub enum PartitionMode {
    /// Left/right children are partitioned using the left and right keys
    Partitioned,
    /// Left side will collected into one partition
    CollectLeft,
    /// DataFusion optimizer decides which PartitionMode
    /// mode(Partitioned/CollectLeft) is optimal based on statistics. It will
    /// also consider swapping the left and right inputs for the Join
    Auto,
}

Auto means “the optimizer decides”, and the decision is:

rust
// datafusion/physical-optimizer/src/join_selection.rs:192-247
pub(crate) fn try_collect_left(
    hash_join: &HashJoinExec,
    ignore_threshold: bool,
    context: &dyn PhysicalOptimizerContext,
) -> Result<Option<Arc<dyn ExecutionPlan>>> {
    let left = hash_join.left();
    let right = hash_join.right();
    let optimizer_config = &context.config_options().optimizer;

    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,
        );
    let right_can_collect = ignore_threshold
        || supports_collect_by_thresholds(
            right, ... , context,
        );

    match (left_can_collect, right_can_collect) {
        (true, true) => {
            // For null-aware joins, we only swap `LeftAnti` joins where the left side is > right side
            if can_swap_hash_join(hash_join)
                && should_swap_join_order(left, right, context)?
            {
                Ok(Some(hash_join.swap_inputs(PartitionMode::CollectLeft)?))
            } else {
                Ok(Some(Arc::new(
                    hash_join.builder()
                        .with_partition_mode(PartitionMode::CollectLeft)
                        .build()?,
                )))
            }
        }
        (true, false) => Ok(Some(Arc::new(
            hash_join.builder()
                .with_partition_mode(PartitionMode::CollectLeft)
                .build()?,
        ))),
        (false, true) => {
            if optimizer_config.join_reordering && can_swap_hash_join(hash_join) {
                hash_join.swap_inputs(PartitionMode::CollectLeft).map(Some)
            } else {
                Ok(None)
            }
        }
        (false, false) => Ok(None),
    }
}

And the threshold test is a two-tier one:

rust
// datafusion/physical-optimizer/src/join_selection.rs:113-134
fn supports_collect_by_thresholds(
    plan: &Arc<dyn ExecutionPlan>,
    threshold_byte_size: usize,
    threshold_num_rows: usize,
    context: &dyn PhysicalOptimizerContext,
) -> bool {
    let Ok(stats) = get_stats(plan, context) else { return false; };

    // Stats use `Precision<T>` to represent stats, where `Absent` means unknown.
    // `Exact(0)` and `Inexact(0)` are both valid stats, and we should not treat
    // them as unknown, `Absent` will return None (this is in regards to why
    // `!=0` is not checked)
    if let Some(byte_size) = stats.total_byte_size.get_value() {
        *byte_size < threshold_byte_size
    } else if let Some(num_rows) = stats.num_rows.get_value() {
        *num_rows < threshold_num_rows
    } else {
        false
    }
}

The comment about Exact(0) is a nice bit of care: an empty side is a valid input to CollectLeft — you get a hash table with zero entries — so zero must not be confused with “unknown”.

The thresholds default to 4 MiB / 131,072 rows:

rust
// datafusion/common/src/config.rs:1847-1853
/// The maximum estimated size in bytes for one input side of a HashJoin
/// will be collected into a single partition
pub hash_join_single_partition_threshold: usize, default = 4 * 1024 * 1024

/// The maximum estimated size in rows for one input side of a HashJoin
/// will be collected into a single partition
pub hash_join_single_partition_threshold_rows: usize, default = 1024 * 128

Why the threshold exists at all. CollectLeft gathers the entire build side into one partition and builds one hash table there. That is faster than Partitioned (no repartition shuffle, no per-partition hash tables) but it requires the build side to fit in memory and it serializes the build phase. Partitioned repartitions both sides by the join key, so each partition’s build side is a fraction of the total — more memory-efficient, but it pays for a shuffle.

So the rule is: if the build side is small, collect it; if it is large, repartition. That’s a genuine cost model, and it’s the closest thing to a “cost-based join selection” that DataFusion has.

When neither side qualifies, the fallback is:

rust
// datafusion/physical-optimizer/src/join_selection.rs:257-283
pub(crate) fn partitioned_hash_join(
    hash_join: &HashJoinExec,
    context: &dyn PhysicalOptimizerContext,
) -> Result<Arc<dyn ExecutionPlan>> {
    let left = hash_join.left();
    let right = hash_join.right();
    let partition_mode = if hash_join.null_aware {
        PartitionMode::CollectLeft
    } else {
        PartitionMode::Partitioned
    };
    if can_swap_hash_join(hash_join) && should_swap_join_order(left, right, context)? {
        hash_join.swap_inputs(partition_mode)
    } else {
        Ok(Arc::new(
            hash_join.builder()
                .with_partition_mode(partition_mode)
                .build()?,
        ))
    }
}

Even in the repartitioned case, the smaller input still becomes the build side — because even with partitioning, a smaller build side means smaller hash tables in every partition.

The guards that stop statistics from breaking things

There are three guards in statistical_join_selection_subrule, and each one fixes a real bug:

rust
// datafusion/physical-optimizer/src/join_selection.rs:293-334
fn statistical_join_selection_subrule(
    plan: Arc<dyn ExecutionPlan>,
    context: &dyn PhysicalOptimizerContext,
) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
    let transformed = if let Some(hash_join) = plan.downcast_ref::<HashJoinExec>() {
        if !hash_join.dynamic_expressions_produced().is_empty() {
            // Once a HashJoinExec carries a dynamic filter, its build side
            // has already been determined and the dynamic filter has been wired
            // to the probe side. Reordering inputs would invalidate the dynamic
            // filter, so skip this join and leave it unchanged.
            None
        } else {
            match hash_join.partition_mode() { ... }
        }
    } else if let Some(cross_join) = plan.downcast_ref::<CrossJoinExec>() {
        ... cross_join.swap_inputs() ...
    } else if let Some(nl_join) = plan.downcast_ref::<NestedLoopJoinExec>() {
        ... nl_join.swap_inputs() ...
    }

The dynamic-filter guard is the interesting one. A hash join can push a runtime filter (a set of build-side keys, or a min/max range) into the probe-side scan. Once that filter is wired up, the build side is fixed — swapping the inputs would invalidate the filter’s wiring. This is a case where a purely “cost-based” decision would silently break a correctness- adjacent mechanism, so the optimizer refuses to make it.

Note also that the same swapping logic applies to CrossJoinExec and NestedLoopJoinExec — because for those, “left” and “right” are symmetric and putting the smaller input first matters for memory.

Unbounded inputs: a separate, non-statistical rule

The rule also has a “pipeline fixer” phase that runs before the statistical one:

rust
// datafusion/physical-optimizer/src/join_selection.rs:524-541
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 // Don't swap null-aware anti joins
        && matches!(
            *hash_join.join_type(),
            JoinType::Inner | JoinType::Left | JoinType::LeftSemi | JoinType::LeftAnti
        )
    {
        input = swap_join_according_to_unboundedness(hash_join)?;
    }
    Ok(input)
}

If one side is an infinite stream and the other is finite, the finite side must be the build side — otherwise the hash table can never be finished. This is not a heuristic; it’s a runnability requirement. The rule refuses to swap Full, Right, RightSemi and RightAnti joins because their semantics don’t survive the swap:

rust
// datafusion/physical-optimizer/src/join_selection.rs:550-563
pub(crate) fn swap_join_according_to_unboundedness(
    hash_join: &HashJoinExec,
) -> Result<Arc<dyn ExecutionPlan>> {
    ...
    match (*partition_mode, *join_type) {
        (
            _,
            JoinType::Right | JoinType::RightSemi | JoinType::RightAnti
            | JoinType::RightMark | JoinType::Full,
        ) => internal_err!("{join_type} join cannot be swapped for unbounded input."),
        ...

And when both sides are unbounded, a HashJoinExec becomes a SymmetricHashJoinExec, which is a genuinely different algorithm designed for streaming:

rust
// datafusion/physical-optimizer/src/join_selection.rs:389-405
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
    );
    let right_unbounded = hash_join.right.boundedness().is_unbounded();
    let right_incremental = matches!(
        hash_join.right.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 {

Note the & instead of && — a bitwise AND on two bools. It works (both operands are bool, no short-circuit) but it’s the kind of thing a reviewer should flag. Whether it’s intentional (forcing both sides to be evaluated) or a typo, the semantics happen to be right.

Estimating the join’s output cardinality

The other half of the answer: how big will this join be? That number feeds the aggregate above it, the limit above that, and the choice of build side for the next join.

The estimator is the classic NDV formula:

rust
// datafusion/physical-plan/src/operator_statistics/mod.rs:936-968
/// Estimate equi-join output using NDV of join key columns:
///   left_rows * right_rows / product(max(left_ndv_i, right_ndv_i))
/// Falls back to Cartesian product if any key lacks NDV on both sides.
fn equi_join_estimate(
    on: JoinOnRef,
    left: &Statistics,
    right: &Statistics,
    left_rows: usize,
    right_rows: usize,
) -> usize {
    if on.is_empty() {
        return left_rows.saturating_mul(right_rows);
    }
    let mut ndv_divisor: usize = 1;
    for (left_key, right_key) in on {
        let left_ndv = left_key
            .downcast_ref::<Column>()
            .and_then(|c| left.column_statistics.get(c.index()))
            .and_then(|s| s.distinct_count.get_value().copied());
        let right_ndv = right_key
            .downcast_ref::<Column>()
            .and_then(|c| right.column_statistics.get(c.index()))
            .and_then(|s| s.distinct_count.get_value().copied());
        match (left_ndv, right_ndv) {
            (Some(l), Some(r)) if l > 0 && r > 0 => {
                ndv_divisor = ndv_divisor.saturating_mul(l.max(r));
            }
            _ => return left_rows.saturating_mul(right_rows),
        }
    }
    let max_rows = left_rows.saturating_mul(right_rows);
    max_rows.checked_div(ndv_divisor).unwrap_or(max_rows)
}

left_rows × right_rows ÷ Π max(left_ndv_i, right_ndv_i), with the Cartesian product as the fallback whenever any key lacks a usable NDV on both sides.

Then the join type adjusts the base estimate:

rust
// datafusion/physical-plan/src/operator_statistics/mod.rs:991-1019
// Apply join-type-aware cardinality bounds
let estimated = match join_type {
    JoinType::Inner => inner_estimate,
    JoinType::Left => inner_estimate.max(left_rows),
    JoinType::Right => inner_estimate.max(right_rows),
    JoinType::Full => {
        // At least left + right - matched, but never less than inner
        let outer_bound = left_rows
            .saturating_add(right_rows)
            .saturating_sub(inner_estimate);
        inner_estimate.max(outer_bound)
    }
    JoinType::LeftSemi => inner_estimate.min(left_rows),
    JoinType::RightSemi => inner_estimate.min(right_rows),
    JoinType::LeftAnti => left_rows.saturating_sub(inner_estimate.min(left_rows)),
    JoinType::RightAnti => {
        right_rows.saturating_sub(inner_estimate.min(right_rows))
    }
    JoinType::LeftMark => left_rows,
    JoinType::RightMark => right_rows,
};

// NL join inner with exact inputs is an exact Cartesian product;
// NDV-based estimates are inherently inexact.
let num_rows = if is_exact_cartesian && join_type == JoinType::Inner {
    Precision::Exact(estimated)
} else {
    Precision::Inexact(estimated)
};

Note the Precision::Exact only for the cross-join-with-exact-inputs case. Everything else is Inexact. That’s the honest labeling: an NDV-based join estimate is never exact, no matter how good the NDVs are.

The same file has the Calcite-derived NDV formulas that feed this:

rust
// datafusion/physical-plan/src/operator_statistics/mod.rs:510-577 (doc excerpts)
/// Estimate the number of distinct values when sampling from a population.
///
/// Given a domain with `domain_size` distinct values and `num_selected` rows
/// sampled/filtered from it, estimates how many distinct values will appear
/// in the sample.
///
/// Uses the formula: `Expected distinct = N * [1 - (1 - 1/N)^n]`
///
/// Based on Calcite's `RelMdUtil.numDistinctVals()`:
pub fn num_distinct_vals(domain_size: usize, num_selected: usize) -> usize { ... }

/// Estimate NDV after applying a selectivity factor (filtering).
///
/// When filtering rows, each distinct value has multiple rows. If a value
/// appears `k` times, the probability it survives the filter is `1 - (1-s)^k`
/// where `s` is the selectivity.
///
/// Assuming uniform distribution (each value appears `rows/ndv` times):
/// ```text
/// NDV_after ~ NDV_before * [1 - (1 - selectivity)^(rows/NDV)]
/// ```
pub fn ndv_after_selectivity(...) { ... }

The extension point: `StatisticsRegistry`

Since core DataFusion can’t know your catalog’s statistics, there’s a pluggable provider chain:

rust
// datafusion/physical-plan/src/operator_statistics/mod.rs:394-399
/// Registry that chains [`StatisticsProvider`] implementations.
///
/// The registry is a stateless provider chain: it holds no mutable state
/// and is cheaply `Clone`able / `Send` / `Sync`.
pub struct StatisticsRegistry { ... }

and the shipped example datafusion-examples/examples/statistics/join_reorder.rs demonstrates the whole point — a catalog that knows user_id has 100 distinct values and amount lives in [0, 1000]:

rust
// datafusion-examples/examples/statistics/join_reorder.rs:60-74
fn catalog_stats(
    plan: &dyn ExecutionPlan,
    _child_stats: &[ExtendedStatistics],
) -> Result<StatisticsResult> {
    let schema = plan.schema();
    let user_id = schema.index_of("user_id")?;
    let amount = schema.index_of("amount")?;
    let mut stats = (*plan.statistics_from_inputs(&[], &StatisticsArgs::new())?).clone();
    stats.column_statistics[amount].min_value =
        Precision::Inexact(ScalarValue::Int32(Some(0)));
    stats.column_statistics[amount].max_value =
        Precision::Inexact(ScalarValue::Int32(Some(999)));
    stats.column_statistics[user_id].distinct_count = Precision::Inexact(100);
    Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats)))
}

The example’s own description is the cleanest summary of what statistics buy you:

the in-memory events table has no column statistics, so the grouped side is estimated at 1000 rows and dims (60 rows) becomes the build side. A provider supplies the amount range and the user_id distinct count a catalog knows; with them the operators estimate the grouped side at 50 rows, below dims, which flips the join build side.

Trade-offs and failure modes

A wrong NDV picks the wrong build side, and it is expensive. The build side is hashed into memory. If you build on the 10-million-row side because you thought it was the 10-thousand-row side, you OOM or thrash. The Precision::Inexact labeling protects you from correctness issues but not from this.

CollectLeft is a memory cliff. The threshold is 4 MiB by estimated size. If the estimate is 10× low, CollectLeft gathers 40 MiB into one partition. And CollectLeft is a pipeline break: nothing downstream of it can start until the whole build side is collected.

Partitioned is a shuffle cliff. Repartitioning both sides of a join by the join key is expensive — a full exchange over the network in a distributed deployment. prefer_hash_join = false (to force sort-merge) trades the hash join’s memory for two sorts, which is usually worse and sometimes better.

No join reordering means a bad plan stays bad. If your SQL has the tables in the wrong order for the data, no amount of statistics will fix it. You have to write the query in the right order, or use a system with a real join optimizer. This is the single biggest limitation in this area.

OR in a join predicate kills the hash join. If the join condition is a.x = b.x OR a.y = b.y, ExtractEquijoinPredicate can’t extract equi-keys, and you fall to NestedLoopJoinExec — O(n×m). DataFusion does not decompose disjunctive equi-joins into a union of hash joins.

NOT BETWEEN and other non-equi predicates go to nested loop. Same story.

Swapping changes semantics for asymmetric join types. can_swap_hash_join is careful:

rust
// datafusion/physical-optimizer/src/join_selection.rs:171-179
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 LeftAnti join can only be swapped if it has exactly one join key and no residual filter — because the swapped RightAnti has no per-row NULL handling. This is the kind of detail that costs a day to discover.

Configuration

Option	Default	Effect
`datafusion.optimizer.prefer_hash_join`	`true`	`false` routes equi-joins to `SortMergeJoinExec` instead of `HashJoinExec`.
`datafusion.optimizer.join_reordering`	`true`	Allow statistics-driven swapping of a join’s inputs. Does not reorder joins in a join tree.
`datafusion.optimizer.hash_join_single_partition_threshold`	`4 MiB`	Byte threshold under which a side qualifies for `CollectLeft`.
`datafusion.optimizer.hash_join_single_partition_threshold_rows`	`131072`	Row threshold under which a side qualifies for `CollectLeft`.
`datafusion.optimizer.enable_piecewise_merge_join`	`false`	Allow `PiecewiseMergeJoinExec` for a single range filter.
`datafusion.optimizer.repartition_joins`	`true`	Whether joins may be repartitioned at all.
`datafusion.optimizer.top_down_join_key_reordering`	`true`	Reorders the keys within a join, not the joins.
`datafusion.execution.collect_statistics`	`true`	Whether statistics exist at all. Off means every estimate is `Absent` and no swap ever happens.

Bottom line

The honest description of DataFusion’s join planning is: the algorithm is chosen structurally, and statistics tune the parameters.

Structural choice: equi-keys → hash join. No equi-keys → cross or nested loop. Sort-merge only if you ask for it.

Statistical tuning: which side builds the hash table, whether to collect or repartition, and how many rows come out.

That is a real, working cost model — the CollectLeft threshold in particular is a good piece of engineering — but it is not a join-order optimizer, and if you are comparing DataFusion to a system that enumerates join trees, that is the gap you will feel first. The workaround is the classic one: write the joins in the order that makes the small side the build side, and use the StatisticsRegistry to make sure the optimizer sees enough to agree with you.