feat: add ASOF join physical operator by Xuanwo · Pull Request #23828 · apache/datafusion

Xuanwo · 2026-07-23T05:06:03Z

Which issue does this PR close?

Part of ASOF join support / Specialize Range Joins #318.
Umbrella PR: feat: support ASOF joins #23738.

Rationale for this change

This is the first layer of the ASOF JOIN stack. It establishes a
broadcast-based physical execution contract independently so later logical-plan,
SQL, DataFrame, and serialization changes can be reviewed as smaller follow-up
PRs.

The initial implementation deliberately favors the simpler broadcast design:
the right input must fit in memory and each left partition scans the shared
right-side batches. A repartitioned implementation can be evaluated separately
without changing the ASOF semantics introduced here.

What changes are included in this PR?

Add AsOfJoinExec for left-preserving, Snowflake-style ASOF semantics.
Coalesce and collect the ordered right input once, then share it across all
left partitions.
Keep the left input partitioned so each partition can scan independently and
preserve the left-side output partitioning.
Preserve merge state across input and output batch boundaries.
Reserve each retained Arrow buffer exactly once, including when right-side
batches are zero-copy slices, and expose build, match, and output metrics.
Define output properties and statistics for the broadcast execution model.
Add physical operator tests covering match directions, equality groups,
batch boundaries, unmatched rows, invalid contracts, shared-buffer memory
accounting, and multi-partition broadcast execution.

Are these changes tested?

Yes:

cargo fmt --all
cargo clippy --all-targets --all-features -- -D warnings
cargo test -p datafusion-physical-plan joins::asof_join --all-features
Extended workspace tests from the contributor guide

Are there any user-facing changes?

This adds a new physical operator API. SQL and DataFrame APIs are intentionally
left to dependent PRs in the ASOF JOIN stack.

codecov-commenter · 2026-07-23T05:34:34Z

Codecov Report

❌ Patch coverage is 83.91167% with 204 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.03%. Comparing base (e8a65f2) to head (a127baa).
⚠️ Report is 121 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/joins/asof_join.rs	83.91%	140 Missing and 64 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #23828      +/-   ##
==========================================
+ Coverage   80.66%   81.03%   +0.36%     
==========================================
  Files        1095     1107      +12     
  Lines      372294   381960    +9666     
  Branches   372294   381960    +9666     
==========================================
+ Hits       300324   309526    +9202     
- Misses      54055    54143      +88     
- Partials    17915    18291     +376

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

2010YOUY01

Thank you! This is a really good start. I have done a quick first pass and left some suggestions.

2010YOUY01 · 2026-07-31T08:55:55Z

+        vec![ChildStats::At(partition), ChildStats::Skip]
+    }
+
+    fn statistics_from_inputs(


Just an idea to make this PR smaller, could we use the default implementation here? We could implement it later in a follow-up PR.

I kept this small override because ASOF has two exact facts the default would discard: the output row count equals the left row count, and unmodified left columns retain their statistics. Right-side column statistics remain unknown. I added a comment to make that scope explicit.

I see, this makes sense.

2010YOUY01 · 2026-08-06T08:14:09Z

There is no new commits after the previous review, you may have forgotten to push the local changes 🤔 @Xuanwo

Xuanwo · 2026-08-07T06:25:03Z

There is no new commits after the previous review, you may have forgotten to push the local changes 🤔 @Xuanwo

Oh, sorry, latest commit is a127baa

2010YOUY01

I went over the implementation in detail, and I think it's well organized.

Need to do before merging

Before merging, I think we could add a few basic tests that run the executor and assert the results, just to cover some different cases:

No equality condition in on.
Different comparison operators in the match condition, such as < and >=.
Complex expressions in the on or match conditions, such as MATCH_CONDITION (l_c1 + l_c2) > r_c1 or ON l_c1 = (r_c1 + r_c2). I think these should be supported.

Optional suggestions

The main implementation/design questions I have are:

We need to buffer batch_size output rows before emitting them, now it's implemented through the PendingRows struct, so we need to buffer all source batches to produce the final output. I think this uses more memory and makes the implementation more complex. An alternative is a) keep a in_progress_batch, and emit it after it reaches threshould b) only buffer one (left_batch, right_batch), and materialize its valid indices into the in_progress_batch when the cursor moves across it. This might be fast enough and simpler.
Each loop iteration only advances the right index by one. We can probably explore some fast-forwarding optimization here.

But I suggest we first implement this end to end, including planning, SQL support, more tests, and benchmarks, before exploring these optimizations further. It should be easier to validate the ideas afterward.

For now, I only suggest trying to simplify the existing implementation or adding more documentation to make future iterations easier. I left a few suggestions in the comments.

2010YOUY01 · 2026-08-10T08:12:09Z

+    /// rows in the same group; left EOF flushes the final pending rows. NULL keys
+    /// and group changes clear the candidate, while output flushes only clear
+    /// pending row references.
+    async fn next_batch(&mut self) -> Result<Option<RecordBatch>> {


I recommend to add a pseudocode in the comment for the algorithm implemented in this function.

Besides, since this is the major loop, I think we can also point to this function from execute() and top doc comments, like 'the key state machine entry point is the next_batch() function'

2010YOUY01 · 2026-08-10T08:15:45Z

+    on: JoinOn,
+    match_condition: AsOfMatchExpr,
+    /// Sorted, unique indices of right columns appended after all left columns.
+    right_output_indices: Vec<usize>,


I think this left input is also possible to get projected, we could instead use the existing convention for projection:

datafusion/datafusion/physical-plan/src/joins/nested_loop_join.rs

Line 236 in 308e212

projection: Option<ProjectionRef>,

2010YOUY01 · 2026-08-10T08:24:52Z

+
+    fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+        InputDistributionRequirements::new(vec![
+            Distribution::UnspecifiedDistribution,


Is it the case the optimizer will insert round-robin repartition, if we declare this UnspecifiedDistribution? It should be fine if it's doing so, I was a little bit confused by this name 🤔

2010YOUY01 · 2026-08-10T08:25:30Z

+        vec![ChildStats::At(partition), ChildStats::Skip]
+    }
+
+    fn statistics_from_inputs(


I see, this makes sense.

2010YOUY01 · 2026-08-10T08:33:00Z

+}
+
+#[derive(Clone)]
+struct Candidate {


We could add some comments to this struct.

Optional suggestion: I think we could put it inside the right InputCursor to simplify the implementation.

Logically, this represents the "previous valid row," and it seems more like an internal implementation detail of InputCursor than a standalone module. I imagine the state machine code would be easier to understand if we structured it this way.

2010YOUY01 · 2026-08-10T08:52:54Z

+    }
+
+    #[tokio::test]
+    async fn broadcasts_right_input_to_all_left_partitions() -> Result<()> {


I think this test don't have a meaningful test goal right now, and it adds maintenance overhead.

It would be better to test it by asserting the plan shape, after we have SQL integration. And I think we could remove it now.

2010YOUY01 · 2026-08-10T09:01:22Z

+        Ok(())
+    }
+
+    #[tokio::test]


Suggested change

#[tokio::test]

// Ensure the build-side memory usage equals the sum of all build-side input

// batches, verifying that the build-side buffer is shared.

#[tokio::test]

2010YOUY01 · 2026-08-10T09:04:40Z

+    }
+
+    #[tokio::test]
+    async fn preserves_dictionary_outputs_across_large_flush() -> Result<()> {


I don't know if we should make such guarantee 🤔

2010YOUY01 · 2026-08-10T09:08:44Z

+    }
+
+    #[test]
+    fn properties_and_statistics_follow_left_preserving_contract() -> Result<()> {


Probably it's not necessary to assert the distribution and ordering properties here

2010YOUY01 · 2026-08-10T09:12:20Z

+                    .iter()
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(


Here is a reference to write assertions for similar tests easier:

datafusion/datafusion/physical-plan/src/joins/nested_loop_join.rs

Line 3350 in 308e212

allow_duplicates!(assert_snapshot!(batches_to_sort_string(&batches), @r"

feat: add ASOF join physical operator

e546de5

github-actions Bot added the physical-plan Changes to the physical-plan crate label Jul 23, 2026

Xuanwo mentioned this pull request Jul 23, 2026

feat: add ASOF join logical semantics #23829

Draft

This was referenced Jul 23, 2026

feat: support ASOF JOIN SQL #23830

Draft

feat: add ASOF joins to DataFrame API #23831

Draft

feat: serialize ASOF join plans #23832

Draft

bench: add ASOF join benchmarks #23833

Draft

feat: support ASOF joins #23738

Open

Xuanwo added 5 commits July 23, 2026 14:30

Merge remote-tracking branch 'origin/main' into xuanwo/asof-physical

68ac8ec

Merge remote-tracking branch 'origin/main' into xuanwo/asof-physical

73fd599

feat: broadcast ASOF right input

43110a3

fix: account shared ASOF build buffers once

339d785

perf: compare ASOF keys without scalar materialization

34552d1

2010YOUY01 reviewed Jul 31, 2026

View reviewed changes

refactor: simplify ASOF physical operator

a127baa

Xuanwo marked this pull request as ready for review August 3, 2026 04:13

Xuanwo requested a review from 2010YOUY01 August 3, 2026 04:14

2010YOUY01 reviewed Aug 10, 2026

View reviewed changes

Uh oh!

Conversation

Xuanwo commented Jul 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

codecov-commenter commented Jul 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

2010YOUY01 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

2010YOUY01 commented Aug 6, 2026

Uh oh!

Xuanwo commented Aug 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

2010YOUY01 left a comment

Choose a reason for hiding this comment

Need to do before merging

Optional suggestions

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

2010YOUY01 Aug 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Xuanwo commented Jul 23, 2026 •

edited

Loading

codecov-commenter commented Jul 23, 2026 •

edited

Loading

Xuanwo commented Aug 7, 2026 •

edited

Loading

2010YOUY01 Aug 10, 2026 •

edited

Loading