Prepared Hash Join Builds: Sharing One Build Side Across Many Probes

TL;DR: Every hash join rebuilds its hash table. If you run ten queries against the same small dimension table, you build the same hash table ten times. DataFusion’s prepare_build API freezes a build side into an immutable PreparedHashJoinBuild — a hash table, a payload batch, a memory reservation, and the dynamic-filter facts derived from it — that can then be attached to any number of compatible joins. The object deliberately retains no task context and no input stream, so it outlives the task that produced it.

Most query engines are stateless: a query comes in, a plan is built, the plan runs, the results go out. The hash join’s build side is rebuilt every single time.

That is fine for a one-shot analytics query. It is wasteful for the workload that dominates modern data serving: a small dimension table (customers, products, tenants) joined against a large fact table, over and over.

DataFusion’s answer is PreparedHashJoinBuild (datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs, 367 lines), and it is an interesting design because of what it refuses to do.

What a prepared build is

pub struct PreparedHashJoinBuild {
    build: Arc<JoinBuildData>,
    keys: Vec<usize>,
    null_equality: NullEquality,
}

and the build data itself:

struct JoinBuildData {
    /// Hash table with row indices into `batch`, also shared with dynamic filters.
    map: Arc<Map>,
    /// The input rows for the build side.
    batch: RecordBatch,
    /// Evaluated build-side key expressions, with float `-0.0` rewritten to
    /// `+0.0` so probe batches can compare against them without scanning the
    /// build side again. The dynamic filters (`bounds`, `membership`) and
    /// `build_side_has_null` were computed from the original keys in
    /// `collect_left_input`.
    values: Vec<ArrayRef>,
    /// Logical nulls of the scalar key for uncorrelated null-aware `LeftMark`.
    ///
    /// Computed once while collecting the build side because dictionary
    /// logical-null discovery can scan nested dictionary values.
    left_mark_build_key_nulls: Option<NullBuffer>,
    /// Bounds computed from the build side; absent for an empty partition.
    bounds: Option<PartitionBounds>,
    /// IN-list values or a hash-table reference used for filter pushdown.
    membership: PushdownStrategy,
    // Keep the reservation after the allocations it accounts for.
    reservation: MemoryReservation,
}

Everything expensive about a build side is in there: the hash table, the payload, the evaluated keys, the min/max bounds, the membership filter that will be published as a dynamic filter, and the memory reservation that accounts for all of it.

The design constraint: no task, no stream

The doc comment on the type is a list of what it is not:

Created by HashJoinExec::prepare_build. The embedding executor owns cache identity, admission, single-flight coordination, cancellation and eviction. This object retains its input buffers and memory reservation until its last lease is dropped; it never retains an input stream or task context.

That last clause is the whole design. A normal HashJoinExec owns an input stream; the stream is polled by a task; the task lives in a TaskContext that is scoped to the query. If the prepared build held any of those, it could not outlive the query that created it — and sharing across queries would be impossible.

So the prepared build holds only data: buffers, an Arc<Map>, and a MemoryReservation. The ownership of the cache — the identity, the admission control, the eviction, the single-flight coordination so that twenty concurrent requests do not all build the same table — is explicitly left to the caller:

The embedding executor owns cache identity, admission, single-flight coordination, cancellation and eviction.

That is a deliberate division of responsibility. DataFusion is a library; it does not know whether the thing embedding it wants an LRU cache, a TTL cache, a per-tenant cache, or a cache that is invalidated when the dimension table changes. So it provides the expensive, reusable, immutable part and lets the embedder provide the policy.

Per-consumer state is separate

The build data is shared; the per-execution state is not. Each consumer gets its own:

/// Create independent mutable state for one consuming join.
pub(super) fn probe_data(&self, probe_threads: usize) -> JoinLeftData {
    JoinLeftData {
        build: Arc::clone(&self.build),
        null_aware_scope_map: None,
        null_value_build_rows: None,
        visited_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
        null_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
        probe_completion: ProbeCompletion::new(probe_threads),
        build_side_has_null: false,
    }
}

Arc::clone for the shared part; fresh bitmaps and a fresh ProbeCompletion for the mutable part. The visited_indices_bitmap is the “which build rows have already been emitted” tracker used by the null-aware join, and it is per-execution because two concurrent probes must not interfere with each other’s bookkeeping.

The “gather copy” trick

The build side’s payload is a single RecordBatch — columnar. But the hash table stores row indices into it, because a hash table entry must point at a whole row, not a column.

When the probe finds a match, the engine has to produce output rows that combine probe columns and build columns. The naive approach is to build a new RecordBatch by gathering one row at a time — slow, and it defeats Arrow’s columnar layout.

DataFusion gathers whole columns instead. The comment on the type says:

Hash-join gathers copy supported build columns into output buffers, including contiguous selections. Output batches can therefore outlive this object without retaining unaccounted cached payload. View, dictionary and nested build columns remain unsupported.

“Output batches can therefore outlive this object” — that is the important part. If the output batch referenced the prepared build’s buffers, then every output batch would keep the entire build side alive, and the memory accounting would be a lie. By copying the selected build columns into the output batch, the output is self-contained and the prepared build can be released as soon as the last consumer is done.

The cost of that copy is paid per matched row, and the alternative — a reference — would have made the memory model unworkable. The comment lists exactly which column types are supported: no views, no dictionaries, no nested types. Those would need the copy to be more complicated, and the restriction is stated up front rather than discovered at runtime.

Validation: the scary part

Attaching the wrong build to a join would silently produce wrong results. So prepare_build and validate check a lot:

pub(super) fn validate(&self, join: &HashJoinExec) -> Result<()> {
    let keys = prepared_key_indices(join)?;
    if join.left.schema() != self.build.batch.schema()
        || keys != self.keys
        || join.null_equality != self.null_equality
    {
        return plan_err!(
            "Prepared hash-join build does not match schema, keys or null equality"
        );
    }
    if let Some(filter) = &join.dynamic_filter {
        let probe_keys = join.on.iter().map(|(_, right)| right);
        if !filter.filter.children().into_iter().eq(probe_keys) {
            return plan_err!(
                "Prepared hash-join dynamic filter keys do not match probe keys"
            );
        }
    }
    Ok(())
}

Schema, key indices, null equality, and the dynamic filter’s probe keys. All must match.

But — and the doc is refreshingly honest about this — the checks are structural, not semantic:

Schema and key compatibility cannot establish that two inputs contain the same data. Attaching the wrong snapshot can silently change query results.

That is the contract. The caller owns cache identity. If you cache the build of customers and the table changes underneath you, the join will use stale data. DataFusion checks everything it can check; the rest is the caller’s responsibility.

The byte-key special case

One more piece of engineering. The prepare_build path has a special case for byte keys:

let byte_keys = keys.iter().any(|&key| {
    matches!(
        schema.field(key).data_type(),
        DataType::Utf8 | DataType::FixedSizeBinary(_)
    )
});
// Range accumulation and IN-list publication materialize ScalarValue
// copies of byte keys. Hash membership borrows the admitted table instead.
let config = if byte_keys {
    let mut config = config.as_ref().clone();
    config.optimizer.hash_join_inlist_pushdown_max_size = 0;
    Arc::new(config)
} else {
    config
};

The dynamic filter has two strategies: publish the build side’s keys as an IN list, or publish a reference to the hash table and let the filter do a membership probe. For byte keys, the IN list strategy would copy every key into a ScalarValue — for a build side of a million UTF-8 strings, that is a million string copies, and it defeats the entire point of sharing a prepared build (whose value is that the data is copied once).

So for byte keys, the IN list strategy is switched off (the max size is set to zero) and the hash-membership strategy is used instead, which borrows the table rather than copying from it.

The same reasoning shows up in the type’s own doc:

UTF-8 and fixed-size binary keys use hash-table membership filters instead of copying range or IN-list values.

The catch

• The caller owns cache invalidation. The prepared build has no idea when the underlying table changed. A service embedding DataFusion has to invalidate the cache when customers is updated, and DataFusion provides no help with that.
• Memory is retained until the last lease drops. The reservation is held for the lifetime of the object, so a cached build side that nobody is using still occupies its memory. The embedder’s eviction policy is what actually frees it.
• The supported column types are restricted. Views, dictionaries and nested build columns are not supported, precisely because the gather copy is not implemented for them.
• Single-flight coordination is not provided. If twenty requests arrive at once and the build is not cached yet, they will all build it. The doc says the embedding executor owns “single-flight coordination” — meaning the embedder has to implement it. This is the single most important piece of the contract, and it is the piece most likely to be got wrong.
• Null-aware joins are limited. “Residual conditions belong to each consuming join; null-aware joins remain unsupported.” The null-aware join’s semantics involve per-execution state that does not fit the immutable-share model.

Why it matters

The idea here is small but it is the direction the whole industry is moving: the plan is cheap to throw away, the data is expensive to recompute, and the two should be separated.

DataFusion’s normal execution model is per-query and per-plan. The prepared build is the first API that lets an embedder say “this specific piece of the plan is expensive, immutable, and worth caching” — and builds that API by refusing to hold on to anything that would tie it to the query that created it.