The ASOF Join as a Merge Scan: How DataFusion Finds the Closest Preceding Row Without a Hash Table

TL;DR: An ASOF join asks “for each row on the left, find the row on the right that is closest in time”. The obvious implementation is a search per row. DataFusion’s implementation is a merge: sort both inputs by (equality_key, match_key), then walk them forward together with a single monotone scan, keeping one candidate right row per equality group and replacing it whenever a better one comes along. One pass, no random access, no hash table, and the output is produced in batches as the scan advances.

sql
SELECT trades.symbol, trades.price, quotes.price
FROM trades
ASOF JOIN quotes
  ON trades.symbol = quotes.symbol
  AND trades.ts >= quotes.ts;

For each trade, find the most recent quote for that symbol at or before the trade’s timestamp. This is the canonical “time-series join”, and it appears in every market-data, log-correlation and state-machine workload.

The semantics are subtle: “closest preceding”, not “closest”. With >=, the trade matches the latest quote at or before its timestamp. With <=, it matches the earliest quote at or after.

Sort, then merge

DataFusion’s AsOfJoinExec (datafusion/physical-plan/src/joins/asof_join.rs, 2,186 lines) is a broadcast sort-merge join. The header explains the contract:

Both inputs must be ordered by their equality keys followed by the match key. For < and <=, the match ordering is reversed so all directions use the same forward-only state machine.

That last sentence is the first clever trick. There are four match operators (<, <=, >, >=), and naively > and >= would need a backward scan while < and <= need a forward one — two different state machines, two sets of bugs.

Instead, the operator reverses the sort direction for the > and >= cases:

text
ON left.symbol = right.symbol MATCH_CONDITION(left.ts >= right.ts)
  left:  [left.symbol ASC NULLS FIRST, left.ts ASC NULLS FIRST]
  right: [right.symbol ASC NULLS FIRST, right.ts ASC NULLS FIRST]

ON left.symbol = right.symbol MATCH_CONDITION(left.ts <= right.ts)
  left:  [left.symbol ASC NULLS FIRST, left.ts DESC NULLS FIRST]
  right: [right.symbol ASC NULLS FIRST, right.ts DESC NULLS FIRST]

Wait — read that again. For <=, the match key is sorted DESC. So scanning forward visits the largest timestamps first, and the first eligible right row encountered is the smallest one that is still >= the left timestamp… no. Let me re-read the eligibility function:

rust
fn is_eligible(op: Operator, right_vs_left: Ordering) -> bool {
    match op {
        Operator::Gt => right_vs_left == Ordering::Less,
        Operator::GtEq => right_vs_left != Ordering::Greater,
        Operator::Lt => right_vs_left == Ordering::Greater,
        Operator::LtEq => right_vs_left != Ordering::Less,
        _ => unreachable!("ASOF match operator is validated by try_new"),
    }
}

right_vs_left is the comparison of the right row’s match key against the left row’s. For >= (GtEq), the right row is eligible when right_vs_left != Greater, i.e. right <= left. Correct: we want quotes at or before the trade.

For <= (LtEq), the right row is eligible when right_vs_left != Less, i.e. right >= left. Correct: we want quotes at or after the trade.

And the candidate is replaced every time a new eligible row is found:

rust
let (batch, row) = self.right.batch_row()?;
// Replacing the candidate selects the nearest eligible row.
// Equal match values have no secondary ordering, so which tied
// row wins is intentionally nondeterministic.
self.candidate = Some(Candidate { ... });
self.right.advance();

So with an ascending scan and >=, each successive eligible right row is later than the previous one, and the last one seen is the latest eligible — exactly what “closest preceding” means. With a descending scan and <=, each successive eligible right row is earlier, and the last one seen is the earliest eligible — “closest following”. Either way: one state machine, one scan direction, “keep replacing the candidate”.

That is a genuinely nice unification. The direction of the sort is the only thing that differs between the four operators.

The state machine

The main loop is AsOfJoinStream::poll_next_impl:

rust
loop {
    if self.pending_left.len() >= self.batch_size {
        return Poll::Ready(Some(self.flush()));
    }
    if !ready!(self.left.poll_ensure_row(cx, ...)?) {
        if !self.pending_left.is_empty() {
            return Poll::Ready(Some(self.flush()));
        }
        self.metrics.baseline.done();
        return Poll::Ready(None);
    }

    if self.left.match_is_null() || self.left.group_has_null() {
        self.candidate = None;
        self.candidate_group_comparator = None;
        self.push_current_left(None)?;
        self.left.advance();
        continue;
    }
    if self.candidate_is_other_group()? {
        self.candidate = None;
        self.candidate_group_comparator = None;
    }

    loop {
        if !ready!(self.right.poll_ensure_row(cx, ...)?) {
            break;
        }
        if self.right.group_has_null() {
            self.right.advance();
            continue;
        }
        match self.compare_input_groups()? {
            Ordering::Less => {
                // The right row's equality group sorts before the left's:
                // it can never match, skip it.
                self.right.advance();
                continue;
            }
            Ordering::Greater => break,
            Ordering::Equal => {}
        }
        if self.right.match_is_null() {
            self.right.advance();
            continue;
        }
        if !is_eligible(self.op, self.compare_input_matches()?) {
            break;
        }
        ... // replace the candidate
        self.right.advance();
    }

    self.push_current_left(self.candidate.clone())?;
    self.left.advance();
}

Read it as a two-pointer merge with one extra piece of state.

The group comparison is the skip. compare_input_groups() compares the equality keys of the current left row and the current right row. If the right row’s group sorts before the left row’s, it can never match this left row or any later one — so it is skipped. If the right row’s group sorts after, the scan stops: no more right rows can match this left row (or any later one in the same group), so the inner loop breaks and the current candidate is used.

The candidate is reset when the group changes. candidate_is_other_group() compares the candidate’s equality key with the current left row’s, and if they differ, the candidate is dropped. That is what keeps the candidate “per equality group”.

Nulls are handled explicitly. A NULL match key on either side means no match, and a NULL equality key means the row belongs to no group. Both are skipped rather than compared.

The comparator cache

The candidate_group_comparator is a small but careful optimisation. The candidate’s group and the left row’s group have to be compared on every left row, but the arrays they live in change as the cursors advance. The comparator is therefore cached and invalidated by batch id:

rust
if self
    .candidate_group_comparator
    .as_ref()
    .is_none_or(|(candidate, left, _)| {
        *candidate != candidate_batch_id || *left != left_batch_id
    })
{
    let comparator = JoinKeyComparator::new(
        candidate.key_arrays.as_ref(),
        self.left.key_arrays.as_ref(),
        &self.group_sort_options,
        NullEquality::NullEqualsNothing,
    )?;
    self.candidate_group_comparator =
        Some((candidate_batch_id, left_batch_id, comparator));
}

The comparator holds references to the key arrays, so it must be rebuilt whenever either side’s arrays change. The key_batch_id is a monotonic counter that changes whenever a new batch is loaded, and the cache key is the pair of ids. So the comparator is rebuilt exactly when it has to be — no more, no less.

The same pattern appears in the InputCursor struct:

rust
/// Cursor over one ordered input stream.
///
/// Expressions are evaluated once per non-empty batch. `key_batch_id` changes
/// whenever a new batch is loaded so comparators cannot retain stale arrays.
struct InputCursor {

“Expressions are evaluated once per non-empty batch” — the join keys are materialised into arrays per batch, and every row in that batch reuses them. Evaluating trades.ts once per batch instead of once per row is the difference between a columnar engine and a row engine.

The output is built with row references

The output is not built row by row. The join accumulates references — (batch, row index) pairs — and materialises them in batches:

rust
fn push_current_left(&mut self, candidate: Option<Candidate>) -> Result<()> {
    let _timer = self.metrics.baseline.elapsed_compute().timer();
    let (left_batch, left_row) = self.left.batch_row()?;
    self.pending_left.push(left_batch, left_row);
    match candidate {
        Some(candidate) => {
            if self.projects_right {
                self.pending_right.push(candidate.batch, candidate.row);
            }
        }
        None => {
            if self.projects_right {
                self.pending_right.push_null();
            }
        }
    }
    Ok(())
}

and flush() materialises them with Arrow’s interleave:

rust
fn flush(&mut self) -> Result<RecordBatch> {
    let row_count = self.pending_left.len();
    let mut arrays = Vec::with_capacity(self.schema.fields().len());
    let mut interleave_indices = Vec::new();
    for (field, column) in self.schema.fields().iter().zip(&self.column_indices) {
        let pending = match column.side {
            JoinSide::Left => &self.pending_left,
            JoinSide::Right => &self.pending_right,
            ...
        };
        arrays.push(pending.materialize_column(
            column.index,
            field.data_type(),
            &mut interleave_indices,
        )?);
    }
    ...
}

The left and right rows come from different batches — the left row from the current left batch, the candidate from whichever right batch it came from. interleave is the Arrow kernel that gathers from multiple arrays by index, so a single call produces the interleaved output column regardless of which batches the rows came from.

And the “no match” case pushes a null rather than skipping the row:

rust
None => {
    if self.projects_right {
        self.pending_right.push_null();
    }
}

which is the correct ASOF semantics: a left row with no matching quote still appears in the output, with NULLs for the right columns.

The plan shape is not fixed

One more thing worth knowing, because it explains why the ASOF join is sometimes fast and sometimes not:

AsOfJoinExec::input_distribution_requirements requires a single right partition but leaves the left distribution unrestricted. AsOfJoinExec::required_input_ordering requires both inputs to be ordered. The physical optimizer satisfies these contracts by inserting operators such as RepartitionExec, SortExec, CoalescePartitionsExec, or SortPreservingMergeExec, depending on the input properties. The inserted plan shape is therefore not fixed by this operator.

So the ASOF join declares what it needs and lets the physical optimizer figure out how to provide it. If the data is already sorted, the sort disappears. If it is not, a sort is inserted — and a sort of the entire right side is a big cost.

The right side is broadcast to a single partition:

This mode preserves probe-side parallelism when there are no equality keys or when equality keys have low cardinality or skew. It retains the complete right input in the memory pool and may scan it once per left partition. Alternative strategies, including broadcasting the other side or repartitioning both inputs, remain future work for other input-size and key-distribution profiles.

“may scan it once per left partition” — with N left partitions, the right side is scanned N times. That is fine when the right side is small (it lives in the memory pool) and the left side is large. The operator trades a memory-resident broadcast for repeated scans, and the comment is candid that other strategies are future work.

The catch

Both inputs must be sorted. If the data is not, the physical optimizer inserts a sort — potentially a very expensive one. The ASOF join is only fast when the ordering is free or already present.
The right side is fully materialised. “It retains the complete right input in the memory pool.” A large right side is a memory problem, and the operator holds a MemoryReservation for it.
Ties are nondeterministic. “Equal match values have no secondary ordering, so which tied row wins is intentionally nondeterministic.” Two quotes with the identical timestamp — either is a valid answer, and DataFusion does not promise which.
Floating-point equality keys are rejected. The constructor’s contract says so explicitly: “floating-point equality keys are not supported because Arrow sorting distinguishes signed zero while SQL equality does not.” That is a lovely example of a subtle mismatch between two subsystems — Arrow’s sort orders -0.0 before 0.0, SQL’s = says they are equal — and the resolution is to refuse the case rather than produce a wrong answer.
The scan is forward-only per direction. Four match operators are supported by reversing the sort, which works, but it means the operator cannot handle a mixed direction within one query.

Why it matters

The ASOF join is a nice illustration of a general principle: many “search” problems become linear once you are willing to sort first. The ASOF join is a nearest-neighbour search over time, and the classic way to make it linear is to sort by time and merge. DataFusion’s implementation adds two refinements on top of the textbook version:

Unify the directions by flipping the sort, so there is one state machine instead of two.
Keep exactly one candidate, so the memory is O(1) per equality group regardless of how many right rows match.

The result is an operator that runs in one pass over sorted inputs, holds one right row per group, and produces its output in batches as it goes.