Opensource internals

How DataFusion Turns a Hash Join's Build Side into a Filter the Probe Side Can Execute

9 min read Jay Hubbert

After reading its build side, a hash join publishes the build side's keys as a runtime filter, choosing between an IN list, a hash-table lookup, and min/max bounds.

TL;DR: A hash join knows, the moment it has read its build side, exactly which key values can possibly match. That is a fact, not a guess — and DataFusion publishes it as a runtime filter that the probe-side scan can execute. Small build sides become an IN list, large ones become a membership test against the build side’s own hash table, and every build contributes min/max bounds as a cheap range pre-filter. The hard parts are the coordination: a partitioned join has N independent build sides whose facts must be merged before any filter is published, and a partially-merged filter would silently drop rows.

Consider the canonical star-schema query:

sql 
SELECT o.*, c.name
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.amount > 100;

The customer_id filter is not in the SQL. But after the join has built its hash table over customers.id, it knows that only 4,000 distinct customer ids exist. Every orders row whose customer_id is not in that set can never match, and does not need to be read from disk, decoded, or compared.

If customers.id happens to be a primary key with min/max statistics, the join also learns a range: id BETWEEN 17 AND 91,338. If the build side is small enough, it learns an exact set.

That is a runtime-discovered filter, and it is exactly the kind of fact that static pushdown cannot express.

Two flavours of build side, two flavours of filter

DataFusion’s hash join has two partition modes (datafusion/physical-plan/src/joins/hash_join/):

  • CollectLeft — the build side is read once by partition 0 and shared with every probe partition. One build side, one fact, one filter.
  • Partitioned — both sides are hash-partitioned on the join key and each partition builds its own hash table independently. N build sides, N facts, and they must be merged.

The merge is the interesting part, and it is where a naive implementation loses rows.

The publication strategy

The filter is chosen per partition at collection time:

rust 
// datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
pub(crate) enum PushdownStrategy {
    /// Use InList for small build sides (< 128MB)
    InList(ArrayRef),
    /// Use map lookup for large build sides. If `Some` the second field
    /// represents the build side's keys using a bucket bitmap.
    Map(Arc<Map>, Option<Arc<KeyRangeBitmap>>),
    /// There was no data in this partition, do not build a dynamic filter for it
    Empty,
}

Three strategies, decided by how big the build side is.

InList — the build side’s key values are materialized into an Arrow array and used to build an InListExpr. The source comment notes the try_new_from_array constructor enables the static_filter optimization, which turns the membership test into a hash lookup rather than a linear scan of the list. The list is also used to build a struct expression for multi-column joins:

rust 
// Use InList expression for small build sides
PushdownStrategy::InList(in_list_array) => {
    let expr = if on_right.len() == 1 {
        // Single column: col IN (val1, val2, ...)
        Arc::clone(&on_right[0])
    } else {
        let fields = build_struct_fields(
            on_right.iter()
                .map(|r| r.data_type(schema))
                .collect::<Result<Vec<_>>>()?
                .as_ref(),
        )?;
        // The return field name and the function field name don't really matter here.
        let return_field =
            Arc::new(Field::new("struct", DataType::Struct(fields), true));
        ...
    };
    Ok(Some(Arc::new(InListExpr::try_new_from_array(
        expr, in_list_array, false, schema,
    )?)))
}

Struct-packing the multi-column key is the standard trick: struct(a, b) IN (struct(1, 2), struct(3, 4)) gives you one hashable value instead of a tuple comparison, and Arrow’s struct equality is a single kernel call.

Map — for a large build side, an IN list of a million values is worse than useless: materializing it costs memory, and evaluating it per row is slow. Instead, the filter becomes a membership test against the build side’s own hash table:

rust 
PushdownStrategy::Map(hash_map, pruning_bitmap) => {
    Ok(Some(Arc::new(HashTableLookupExpr::new(
        on_right.to_vec(),
        random_state.clone(),
        hash_map,
        "hash_lookup".to_string(),
        pruning_bitmap,
    )) as Arc<dyn PhysicalExpr>))
}

HashTableLookupExpr (datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs) is a physical expression that hashes its input columns with a preserved seed and probes the map:

rust 
fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
    // Evaluate columns
    let join_keys = evaluate_columns(&self.on_columns, batch)?;

    match self.map.as_ref() {
        Map::HashMap(map) => {
            with_hashes(&join_keys, self.random_state.random_state(), |hashes| {
                let array = map.contain_hashes(hashes);
                Ok(ColumnarValue::Array(Arc::new(array)))
            })
        }
        Map::ArrayMap(map) => {
            let array = map.contain_keys(&join_keys)?;
            Ok(ColumnarValue::Array(Arc::new(array)))
        }
    }
}

This is the elegant part: the build side’s hash table is already there, already built, already warm. Reusing it as a filter costs one hash computation and one probe per row — and the probe side would have done exactly that work anyway inside the join. The filter is not extra work; it is the same work, hoisted earlier so that non-matching rows never get decoded.

The pruning_bitmap is a KeyRangeBitmap — a summary of which buckets of the key space are occupied. It exists so the filter can be used for container pruning (skip whole row groups) without materializing the keys.

Empty — the partition had no build rows at all. Nothing to publish.

Bounds: the cheap pre-filter

Independently of the membership strategy, every partition contributes min/max bounds for each join key column:

rust 
pub(crate) struct ColumnBounds {
    /// The minimum value observed for this column
    pub(crate) min: ScalarValue,
    /// The maximum value observed for this column
    pub(crate) max: ScalarValue,
}

and they are turned into a range predicate:

rust 
fn create_bounds_predicate(
    on_right: &[PhysicalExprRef],
    bounds: &PartitionBounds,
) -> Option<Arc<dyn PhysicalExpr>> {
    let mut column_predicates = Vec::with_capacity(on_right.len());

    for (col_idx, right_expr) in on_right.iter().enumerate() {
        if let Some(column_bounds) = bounds.get_column_bounds(col_idx) {
            // Create predicate: col >= min AND col <= max
            let min_expr = Arc::new(BinaryExpr::new(
                Arc::clone(right_expr), Operator::GtEq, lit(column_bounds.min.clone()),
            )) as Arc<dyn PhysicalExpr>;
            let max_expr = Arc::new(BinaryExpr::new(
                Arc::clone(right_expr), Operator::LtEq, lit(column_bounds.max.clone()),
            )) as Arc<dyn PhysicalExpr>;
            ...

Bounds are combined with membership using AND, and membership alone is enough when bounds are unavailable:

rust 
fn combine_membership_and_bounds(
    membership_expr: Option<Arc<dyn PhysicalExpr>>,
    bounds_expr: Option<Arc<dyn PhysicalExpr>>,
) -> Option<Arc<dyn PhysicalExpr>> {
    match (membership_expr, bounds_expr) {
        (Some(membership), Some(bounds)) =>
            Some(Arc::new(BinaryExpr::new(bounds, Operator::And, membership)) ...),
        (Some(membership), None) => Some(membership),
        (None, Some(bounds)) => Some(bounds),
        (None, None) => None,
    }
}

The bounds are not just an optimisation — they are what makes the filter composable with pruning. A Parquet row group knows its customer_id min/max; a filter of the form customer_id BETWEEN 17 AND 91338 can be evaluated against that statistic and the row group skipped without reading a byte. An IN list of 4,000 values cannot be evaluated against a min/max statistic at all.

The barrier: why a partial filter is a bug

Here is the crux. In Partitioned mode, partition 0 of the build side and partition 1 of the build side are different key ranges. A filter derived from partition 0 alone would reject every probe row that should have matched in partition 1.

The source is blunt about it:

We cannot build a partial filter from some partitions - it would incorrectly eliminate valid join results. We must wait until we have complete information from ALL relevant partitions before updating the dynamic filter.

So DataFusion implements a barrier with an elected finalizer. The doc comment on SharedBuildAccumulator describes the protocol:

  1. Each partition computes information from its build-side data (hash maps and/or bounds)
  2. Information is stored in the shared state, which tracks how many partitions have reported
  3. When the last partition reports, one waiter is elected as the finalizer; it merges the collected information, updates the dynamic filter exactly once, and publishes the terminal result by transitioning CompletionState to Ready
  4. A tokio::sync::Notify wakes any other partitions parked in wait_for_completion, which then observe the Ready state under the mutex and return immediately
rust 
pub(crate) async fn report_build_data(&self, data: PartitionBuildData) -> Result<()> {
    let finalize_input = {
        let mut guard = self.inner.lock();
        self.store_build_data(&mut guard, data)?;
        self.take_finalize_input_if_ready(&mut guard)
    };

    if let Some(finalize_input) = finalize_input {
        self.finish(finalize_input);
    }

    self.wait_for_completion().await
}

Only one thread wins the election — the one that observes the last pending partition turn into a reported one while holding the mutex — and it performs the merge. Every other thread parks on the Notify.

Note that the finalize happens outside the mutex. The finalizer’s work (build the IN list, build the CASE expression, evaluate the bounds) is not free, and holding the lock through it would serialise every partition’s report behind the filter construction.

Routing: one filter, N partitions

The merged partitioned filter has to route each probe row to the right partition’s filter, because partition 3’s build side contains different keys than partition 0’s. DataFusion solves this with a CASE expression keyed on the partition id:

rust 
// Empty partitions reject their routed rows, while canceled partitions stay
// permissive because their build contents are unknown.
fn build_partitioned_filter(&self, partitions: Vec<PartitionStatus>) -> Result<()> {
    ...
    //   WHEN empty_partition_id THEN false  -- only when cancellation exists
    //   WHEN real_partition_id THEN F(real_partition_id)

The generated expression is roughly:

sql 
CASE repartition_partition_id(cols)
  WHEN 0 THEN hash_lookup_0(cols) AND col BETWEEN min0 AND max0
  WHEN 1 THEN hash_lookup_1(cols) AND col BETWEEN min1 AND max1
  ...
  ELSE true
END

Two details here are pure defensive engineering:

  • An empty build partition maps to false — nothing can match it, so rejecting is correct and cheap.
  • A canceled partition (the build task failed or was cancelled) maps to true — permissive, because we do not know what it contained. Guessing “empty” here would silently drop rows.

NULL handling

A join with NullEqualsNull semantics can match a probe-side NULL against a build-side NULL. A pushed filter col IN (1, 2, 3) evaluates to NULL for a NULL col, and a filter that rejects NULLs would drop rows that the join would have matched. The accumulator tracks this:

rust 
/// Null equality of the join. Under `NullEqualsNull` a probe-side NULL can match a
/// build-side NULL, so the pushed filter must keep NULL rows here too.
null_equality: NullEquality,
/// Null-aware anti join (`NOT IN`). A probe-side NULL must reach the join so its
/// three-valued logic can collapse the result, so the pushed filter keeps NULL rows.
null_aware: Option<NullAwareMode>,

and there is a preserve_probe_nulls helper applied to the final filter expression. Getting this wrong is a correctness bug that would show up as “missing rows in a NOT IN join”, which is exactly the kind of bug that is hard to notice and expensive to debug.

The catch

  • The barrier costs latency. The filter cannot be published until all build partitions have reported. For a join whose build side is slow to read, the probe side idles at the filter (or proceeds unfiltered) until the barrier clears. The whole point of the dynamic filter is defeated if the build side takes longer than the probe side.
  • A partitioned merge is not a merge of IN lists. With N partitions you cannot express “row matches if it matches partition k’s filter” as a single IN list without losing the routing. Hence the CASE. For large partitioned builds the Map strategy wins, and each branch is a hash lookup against that partition’s table.
  • The hash seed must match. HashTableLookupExpr hashes with the join’s own HASH_JOIN_SEED ('J', 'O', 'I', 'N' — a four-byte ASCII spelling of “JOIN”), and SeededRandomState exists specifically because RandomState does not expose its seed after construction. Without the preserved seed the filter’s hashes would not match the build table’s hashes, and the filter would reject every row.

That last detail is a nice illustration of a general theme in this code base: performance tricks in DataFusion frequently come with a small, boring piece of bookkeeping (a preserved seed, a generation counter, a monotone cache guard) that is easy to overlook and catastrophic to remove.

DatafusionJoinsDynamic filtersPerformanceInternals