Skip to content

feat: support ASOF joins#23738

Open
Xuanwo wants to merge 30 commits into
apache:mainfrom
Xuanwo:xuanwo/asof-join
Open

feat: support ASOF joins#23738
Xuanwo wants to merge 30 commits into
apache:mainfrom
Xuanwo:xuanwo/asof-join

Conversation

@Xuanwo

@Xuanwo Xuanwo commented Jul 21, 2026

Copy link
Copy Markdown
Member

I understand there are many overlapping discussions and refactoring efforts
regarding JOIN and ASOF JOIN. This PR aims to build the first workable
foundation for subsequent work, and I am open to adjusting the scope or stack.

Parts of this PR were drafted with assistance from Codex (with
gpt-5.6-sol-max) and fully reviewed and edited by me. I take full
responsibility for all changes.


Stacked PRs

This PR remains the umbrella and end-to-end reference for ASOF JOIN. The
implementation is split into reviewable PRs refreshed onto
origin/main at e8a65f2d6.
Each child is intended to be independently mergeable in dependency order. The
physical operator can merge first; the logical layer then adds planning while
failing closed at unsupported frontends; SQL, DataFrame, and protobuf support
can merge independently after that.

Because GitHub cannot select a fork branch as the base of an upstream PR,
dependent PR pages show cumulative diffs against main. Each PR body links its
isolated fork-to-fork diff.

#23828 physical operator
└── #23829 logical semantics and physical planning
    ├── #23830 SQL frontend and unparser
    │   └── #23833 benchmarks
    ├── #23831 DataFrame API
    └── #23832 protobuf serialization
Layer PR Depends on Scope
Physical operator #23828 Broadcast AsOfJoinExec, properties, statistics, metrics, and unit tests
Logical semantics #23829 #23828 Logical node/builder, optimizer integration, physical planning, and fail-closed boundaries
SQL frontend #23830 #23829 SQL planning, unparser, docs, integration tests, and SLT
DataFrame API #23831 #23829 join_asof, join_asof_using, prelude exports, and API test
Serialization #23832 #23829 Logical/physical protobuf messages and round trips
Benchmarks #23833 #23830 End-to-end dfbench workloads and physical Criterion benchmark

Which issue does this PR close?

Rationale for this change

Time-series and ordered-event workloads often need to match each left row with
the nearest eligible right row, optionally within equality-key groups.
Expressing this through correlated subqueries is awkward and does not give the
physical planner a dedicated ordered execution contract.

This PR adds a first-class, left-preserving ASOF join for bounded inputs.

Initial execution design

This version follows the broadcast-based design suggested in the review:

  • The right input is required to be single-partitioned and ordered. It is
    collected once, retained under a memory reservation, and shared immutably by
    all left partitions.
  • The left input keeps its existing partitioning. Each left partition maintains
    an independent merge cursor and scans the shared right-side batches, so output
    partitioning is inherited from the left.
  • Merge state, equality-group state, and the current candidate survive input
    batch boundaries and output flushes. Group changes and EOF cannot reuse a
    candidate from another equality group or drop the final eligible candidate.
  • The operator requires bounded inputs and ordered match keys, supports <,
    <=, >, and >=, and NULL-pads unmatched right columns.

The tradeoff is explicit: the complete right input must fit in memory, and its
rows may be scanned once per left partition. This keeps the first operator and
its correctness contract small. A repartitioned ASOF implementation can be
added later as a separate physical strategy without changing logical or SQL
semantics.

What changes are included in this PR?

  • Add the broadcast-based AsOfJoinExec and its state, properties, statistics,
    memory accounting, and metrics.
  • Add logical plan, optimizer, physical planner, SQL, unparser, and DataFrame
    support.
  • Add logical and physical protobuf round trips. Substrait serialization fails
    closed until an ASOF extension is defined.
  • Add end-to-end and physical-plan benchmarks covering ordered inputs,
    optimizer-inserted sort/repartition, wide and dictionary payloads, descending
    successor matching, and one versus four left partitions.

Are these changes tested?

Yes. On the current umbrella head:

  • cargo fmt --all
  • ./ci/scripts/doc_prettier_check.sh --write --allow-dirty
  • ./datafusion/proto-models/regen.sh
  • cargo clippy --all-targets --all-features -- -D warnings
  • Focused physical, SQL/DataFrame, protobuf, Substrait, and ASOF SLT tests
  • The extended workspace test command from the contributor guide, including all
    500 sqllogictest files
  • cargo bench -p datafusion-physical-plan --bench asof_join --features test_utils -- --test

The current benchmark branch also completed all four dfbench asof workloads
with the expected row counts and all six physical Criterion smoke cases. These
runs validate the benchmark harness; they are not presented as comparative
performance claims because main cannot plan the ASOF workloads.

Are there any user-facing changes?

Yes. Users can construct ASOF joins through SQL, LogicalPlanBuilder, and
DataFrame. The initial contract is left-preserving, supports optional equality
keys plus one ordered match condition, and rejects unbounded inputs. Existing
join behavior is unchanged.

@github-actions github-actions Bot added sql SQL Planner logical-expr Logical plan and expressions optimizer Optimizer rules core Core DataFusion crate substrait Changes to the substrait crate proto Related to proto crate physical-plan Changes to the physical-plan crate labels Jul 21, 2026
# Conflicts:
#	datafusion/proto/src/physical_plan/mod.rs
@Xuanwo
Xuanwo marked this pull request as ready for review July 21, 2026 07:26
@github-actions

github-actions Bot commented Jul 21, 2026

Copy link
Copy Markdown

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details
     Cloning apache/main
    Building datafusion v54.1.0 (current)
       Built [  86.394s] (current)
     Parsing datafusion v54.1.0 (current)
      Parsed [   0.029s] (current)
    Building datafusion v54.1.0 (baseline)
       Built [  84.283s] (baseline)
     Parsing datafusion v54.1.0 (baseline)
      Parsed [   0.030s] (baseline)
    Checking datafusion v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.689s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [ 174.364s] datafusion
    Building datafusion-expr v54.1.0 (current)
       Built [  21.230s] (current)
     Parsing datafusion-expr v54.1.0 (current)
      Parsed [   0.064s] (current)
    Building datafusion-expr v54.1.0 (baseline)
       Built [  21.382s] (baseline)
     Parsing datafusion-expr v54.1.0 (baseline)
      Parsed [   0.064s] (baseline)
    Checking datafusion-expr v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   1.566s] 223 checks: 221 pass, 1 fail, 1 warn, 30 skip

--- failure enum_variant_added: enum variant added on exhaustive enum ---

Description:
A publicly-visible enum without #[non_exhaustive] has a new variant.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.49.0/src/lints/enum_variant_added.ron

Failed in:
  variant LogicalPlan:AsOfJoin in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:245
  variant LogicalPlan:AsOfJoin in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:245

--- warning partial_ord_enum_variants_reordered: enum variants reordered in #[derive(PartialOrd)] enum ---

Description:
A public enum that derives PartialOrd had its variants reordered. #[derive(PartialOrd)] uses the enum variant order to set the enum's ordering behavior, so this change may break downstream code that relies on the previous order.
        ref: https://doc.rust-lang.org/std/cmp/trait.PartialOrd.html#derivable
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.49.0/src/lints/partial_ord_enum_variants_reordered.ron

Failed in:
  LogicalPlan::Repartition moved from position 7 to 8, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:249
  LogicalPlan::Union moved from position 8 to 9, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:253
  LogicalPlan::TableScan moved from position 9 to 10, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:256
  LogicalPlan::EmptyRelation moved from position 10 to 11, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:260
  LogicalPlan::Subquery moved from position 11 to 12, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:263
  LogicalPlan::SubqueryAlias moved from position 12 to 13, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:265
  LogicalPlan::Limit moved from position 13 to 14, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:267
  LogicalPlan::Statement moved from position 14 to 15, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:269
  LogicalPlan::Values moved from position 15 to 16, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:274
  LogicalPlan::Explain moved from position 16 to 17, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:277
  LogicalPlan::Analyze moved from position 17 to 18, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:281
  LogicalPlan::Extension moved from position 18 to 19, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:284
  LogicalPlan::Distinct moved from position 19 to 20, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:287
  LogicalPlan::Dml moved from position 20 to 21, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:289
  LogicalPlan::Ddl moved from position 21 to 22, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:291
  LogicalPlan::Copy moved from position 22 to 23, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:293
  LogicalPlan::DescribeTable moved from position 23 to 24, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:296
  LogicalPlan::Unnest moved from position 24 to 25, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:299
  LogicalPlan::RecursiveQuery moved from position 25 to 26, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:301
  LogicalPlan::Repartition moved from position 7 to 8, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:249
  LogicalPlan::Union moved from position 8 to 9, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:253
  LogicalPlan::TableScan moved from position 9 to 10, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:256
  LogicalPlan::EmptyRelation moved from position 10 to 11, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:260
  LogicalPlan::Subquery moved from position 11 to 12, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:263
  LogicalPlan::SubqueryAlias moved from position 12 to 13, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:265
  LogicalPlan::Limit moved from position 13 to 14, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:267
  LogicalPlan::Statement moved from position 14 to 15, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:269
  LogicalPlan::Values moved from position 15 to 16, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:274
  LogicalPlan::Explain moved from position 16 to 17, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:277
  LogicalPlan::Analyze moved from position 17 to 18, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:281
  LogicalPlan::Extension moved from position 18 to 19, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:284
  LogicalPlan::Distinct moved from position 19 to 20, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:287
  LogicalPlan::Dml moved from position 20 to 21, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:289
  LogicalPlan::Ddl moved from position 21 to 22, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:291
  LogicalPlan::Copy moved from position 22 to 23, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:293
  LogicalPlan::DescribeTable moved from position 23 to 24, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:296
  LogicalPlan::Unnest moved from position 24 to 25, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:299
  LogicalPlan::RecursiveQuery moved from position 25 to 26, in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/plan.rs:301

     Summary semver requires new major version: 1 major and 0 minor checks failed
     Warning produced 1 major and 0 minor level warnings
    Finished [  45.426s] datafusion-expr
    Building datafusion-optimizer v54.1.0 (current)
       Built [  21.870s] (current)
     Parsing datafusion-optimizer v54.1.0 (current)
      Parsed [   0.024s] (current)
    Building datafusion-optimizer v54.1.0 (baseline)
       Built [  21.562s] (baseline)
     Parsing datafusion-optimizer v54.1.0 (baseline)
      Parsed [   0.026s] (baseline)
    Checking datafusion-optimizer v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.180s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  44.632s] datafusion-optimizer
    Building datafusion-physical-plan v54.1.0 (current)
       Built [  29.212s] (current)
     Parsing datafusion-physical-plan v54.1.0 (current)
      Parsed [   0.120s] (current)
    Building datafusion-physical-plan v54.1.0 (baseline)
       Built [  29.185s] (baseline)
     Parsing datafusion-physical-plan v54.1.0 (baseline)
      Parsed [   0.119s] (baseline)
    Checking datafusion-physical-plan v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.741s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  60.882s] datafusion-physical-plan
    Building datafusion-proto v54.1.0 (current)
       Built [  48.660s] (current)
     Parsing datafusion-proto v54.1.0 (current)
      Parsed [   0.016s] (current)
    Building datafusion-proto v54.1.0 (baseline)
       Built [  49.204s] (baseline)
     Parsing datafusion-proto v54.1.0 (baseline)
      Parsed [   0.017s] (baseline)
    Checking datafusion-proto v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.250s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  99.626s] datafusion-proto
    Building datafusion-proto-models v54.1.0 (current)
       Built [  19.219s] (current)
     Parsing datafusion-proto-models v54.1.0 (current)
      Parsed [   0.105s] (current)
    Building datafusion-proto-models v54.1.0 (baseline)
       Built [  19.330s] (baseline)
     Parsing datafusion-proto-models v54.1.0 (baseline)
      Parsed [   0.108s] (baseline)
    Checking datafusion-proto-models v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   1.830s] 223 checks: 222 pass, 1 fail, 0 warn, 30 skip

--- failure enum_variant_added: enum variant added on exhaustive enum ---

Description:
A publicly-visible enum without #[non_exhaustive] has a new variant.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.49.0/src/lints/enum_variant_added.ron

Failed in:
  variant LogicalPlanType:AsOfJoin in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:83
  variant LogicalPlanType:AsOfJoin in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:83
  variant PhysicalPlanType:AsOfJoin in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:1407
  variant PhysicalPlanType:AsOfJoin in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:1407

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  41.879s] datafusion-proto-models
    Building datafusion-sql v54.1.0 (current)
       Built [  32.759s] (current)
     Parsing datafusion-sql v54.1.0 (current)
      Parsed [   0.024s] (current)
    Building datafusion-sql v54.1.0 (baseline)
       Built [  32.491s] (baseline)
     Parsing datafusion-sql v54.1.0 (baseline)
      Parsed [   0.026s] (baseline)
    Checking datafusion-sql v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.267s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  67.123s] datafusion-sql
    Building datafusion-sqllogictest v54.1.0 (current)
       Built [ 149.908s] (current)
     Parsing datafusion-sqllogictest v54.1.0 (current)
      Parsed [   0.023s] (current)
    Building datafusion-sqllogictest v54.1.0 (baseline)
       Built [ 146.677s] (baseline)
     Parsing datafusion-sqllogictest v54.1.0 (baseline)
      Parsed [   0.018s] (baseline)
    Checking datafusion-sqllogictest v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.099s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [ 301.780s] datafusion-sqllogictest
    Building datafusion-substrait v54.1.0 (current)
       Built [ 273.383s] (current)
     Parsing datafusion-substrait v54.1.0 (current)
      Parsed [   0.013s] (current)
    Building datafusion-substrait v54.1.0 (baseline)
       Built [ 280.021s] (baseline)
     Parsing datafusion-substrait v54.1.0 (baseline)
      Parsed [   0.015s] (baseline)
    Checking datafusion-substrait v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.233s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [ 556.075s] datafusion-substrait

@github-actions github-actions Bot added the auto detected api change Auto detected API change label Jul 21, 2026

@xudong963 xudong963 left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the PR, the implementation is a complete vertical slice, looks promising.

Some todos in my mind:

  1. Add an ASOF section documenting syntax, left preservation, optional equality keys, operator directions, null behavior, bounded-only execution, single-partition behavior without equality keys, and nondeterministic selection among tied right rows.
  2. Add an asof_join.slt covering at least all four operators, grouped and ungrouped matching, nulls, USING, unmatched rows, invalid conditions, and EXPLAIN

Comment thread datafusion/sql/src/unparser/plan.rs
Comment thread datafusion/optimizer/src/push_down_filter.rs Outdated
@github-actions github-actions Bot added documentation Improvements or additions to documentation sqllogictest SQL Logic Tests (.slt) labels Jul 21, 2026
@2010YOUY01

Copy link
Copy Markdown
Contributor

I think we should first agree on the high-level direction and then start polishing the implementation. We could ignore the implementation details and test coverage for now.

I think the two most important questions are:

  • What are the precise semantics of an ASOF join? I suspect that different systems, such as DuckDB and Snowflake, may differ slightly, so we should first define the semantics rigorously.
  • What is the high-level algorithm used by the ASOF join operator?

@codecov-commenter

codecov-commenter commented Jul 21, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 82.62811% with 817 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.56%. Comparing base (ed22a4a) to head (2a49feb).
⚠️ Report is 86 commits behind head on main.

Files with missing lines Patch % Lines
datafusion/common/src/hash_utils/build_hasher.rs 63.01% 93 Missing and 15 partials ⚠️
...ysical-plan/src/aggregates/ordered_final_stream.rs 75.93% 82 Missing and 8 partials ⚠️
datafusion/expr/src/logical_plan/plan.rs 55.62% 63 Missing and 12 partials ⚠️
datafusion/physical-plan/src/aggregates/mod.rs 76.97% 37 Missing and 33 partials ⚠️
benchmarks/src/asof.rs 0.00% 62 Missing ⚠️
...l-expr/src/expressions/in_list/primitive_filter.rs 83.67% 9 Missing and 31 partials ⚠️
datafusion/optimizer/src/push_down_filter.rs 44.06% 25 Missing and 8 partials ⚠️
datafusion/common/src/hash_utils.rs 92.28% 7 Missing and 16 partials ⚠️
datafusion/functions-aggregate/src/array_agg.rs 77.00% 5 Missing and 18 partials ⚠️
datafusion/common/src/scalar/mod.rs 80.80% 19 Missing ⚠️
... and 45 more
Additional details and impacted files
@@            Coverage Diff             @@
##             main   #23738      +/-   ##
==========================================
- Coverage   80.69%   80.56%   -0.14%     
==========================================
  Files        1089     1097       +8     
  Lines      368447   374790    +6343     
  Branches   368447   374790    +6343     
==========================================
+ Hits       297332   301957    +4625     
- Misses      53405    54773    +1368     
- Partials    17710    18060     +350     

☔ 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.

@Xuanwo

Xuanwo commented Jul 21, 2026

Copy link
Copy Markdown
Member Author

I think we should first agree on the high-level direction and then start polishing the implementation. We could ignore the implementation details and test coverage for now.

Thank you @2010YOUY01 joining the review. I agree we should settle the semantics and execution model first.

  • What are the precise semantics of an ASOF join? I suspect that different systems, such as DuckDB and Snowflake, may differ slightly, so we should first define the semantics rigorously.

I propose Snowflake-style semantics: each left row selects the closest eligible right row within an optional equality-key group. >/>= selects the greatest right key; </<= selects the smallest. Left rows are always preserved, unmatched rows are NULL-padded, NULL keys do not match, and ties are nondeterministic.

  • What is the high-level algorithm used by the ASOF join operator?

The initial operator uses an ordered merge: co-partition and order both inputs, then advance the right cursor monotonically while retaining the latest eligible candidate. This is O(L + R) after ordering with bounded join state. Other implementations could later share the same logical semantics.


Does this direction make sense?

@alamb alamb left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks neat -- I skimmed it quickly

| LogicalPlan::Aggregate(_)
| LogicalPlan::Sort(_)
| LogicalPlan::Join(_)
| LogicalPlan::AsOfJoin(_)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why does AsOfJoin get its own logical plan? Isn't the choice of join algorithm typically done via a physical plan?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think ASOF joins and regular joins are different relational operations. If many planning or logical optimization tasks apply only to ASOF joins, splitting them would probably simplify the implementation; otherwise, they should remain combined.

(I’m not sure about the current state of the implementation, and which approach should we take.)

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, the execution algorithm should remain a physical decision.

The separate logical node represents different relational semantics: for left ts = 10 and right ts = {1, 5}, a regular >= join returns both rows, while ASOF returns only 5. Replacing a regular Join’s physical algorithm with AsOfJoinExec would therefore change results.

The current implementation also needs ASOF-specific cardinality, type coercion, projection/filter pushdown, and must not participate in ordinary join reordering or input swapping. The logical node can still map to different physical implementations—ordered merge now, indexed probe later.

We could instead add an explicit match mode to Join, but then every generic join rule would need to account for it. I currently prefer a separate node for isolation, but I’m open to a generalized representation if that is preferred.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it -- sorry -- I am still coming up to speed with AS OF (and how the relate to range joins).

Is there a standard syntax for AS OF joins? For example it seems like DuckDB has a different syntax https://duckdb.org/docs/current/guides/sql_features/asof_join

(no MATCH_CONDIITION) and we could maybe model it as a diferent join type 🤔

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think @Xuanwo proposed to implement Snowflake syntax (with match_condition keyword)
https://docs.snowflake.com/en/sql-reference/constructs/asof-join

Personally I also think Snowflake syntax is better than the DuckDB syntax

-- DuckDB synatx
SELECT t.*, p.price
FROM trades t
ASOF JOIN prices p
ON t.symbol = p.symbol AND t.when >= p.when;

-- Snowflake syntax
SELECT t.*, p.price
FROM trades t
ASOF JOIN prices p
MATCH_CONDITION t.when >= p.when
ON t.symbol = p.symbol;

The reason is the in-equality predicate has different semantical meaning:

  • equal condition: pre-filter all pairs before the ASOF join step
  • in-equality condition: for all satisfied prices table rows, only return the closest match

Separating them is more intuitive given the special semantics of ASOF joins.

The downside is that we informally use DuckDB as our primary reference system, so following Snowflake here might depart from that convention.

@alamb

alamb commented Jul 22, 2026

Copy link
Copy Markdown
Contributor

Thank you @Xuanwo -- I think it is not likely I will be able to find time to carefully review a 4500 line PR

It is really helpful to see the design running end to end

Is there any chance you can break this one up into smaller PRs for easier review:

  1. benchmarks
  2. SQL support / LogicalPlan
  3. Physical operator (maybe we can break it down more)
  4. The Dataframe API

@xudong963

Copy link
Copy Markdown
Member

Thank you @Xuanwo -- I think it is not likely I will be able to find time to carefully review a 4500 line PR

It is really helpful to see the design running end to end

Is there any chance you can break this one up into smaller PRs for easier review:

  1. benchmarks
  2. SQL support / LogicalPlan
  3. Physical operator (maybe we can break it down more)
  4. The Dataframe API

Yeah, agree.

If we agree with the PR direction, then it's better to split into a couple of small PRs to make it easier to move forward.

@2010YOUY01

Copy link
Copy Markdown
Contributor

I think we should first agree on the high-level direction and then start polishing the implementation. We could ignore the implementation details and test coverage for now.

Thank you @2010YOUY01 joining the review. I agree we should settle the semantics and execution model first.

  • What are the precise semantics of an ASOF join? I suspect that different systems, such as DuckDB and Snowflake, may differ slightly, so we should first define the semantics rigorously.

I propose Snowflake-style semantics: each left row selects the closest eligible right row within an optional equality-key group. >/>= selects the greatest right key; </<= selects the smallest. Left rows are always preserved, unmatched rows are NULL-padded, NULL keys do not match, and ties are nondeterministic.

  • What is the high-level algorithm used by the ASOF join operator?

The initial operator uses an ordered merge: co-partition and order both inputs, then advance the right cursor monotonically while retaining the latest eligible candidate. This is O(L + R) after ordering with bounded join state. Other implementations could later share the same logical semantics.

Does this direction make sense?

Thanks for explaining the rationale! I’m interested in helping further with this feature, now I need some time to think it through.

@Xuanwo

Xuanwo commented Jul 23, 2026

Copy link
Copy Markdown
Member Author

For @alamb

It is really helpful to see the design running end to end

Yes! That's exactly why I started a full end-to-end demo PR first. Proving that it works and using it as a starting point to split tasks makes much more sense nowadays.

Is there any chance you can break this one up into smaller PRs for easier review:

Sure, will happy to do this.


For @xudong963

If we agree with the PR direction, then it's better to split into a couple of small PRs to make it easier to move forward.

Let's go.


For @2010YOUY01

Thanks for explaining the rationale! I’m interested in helping further with this feature, now I need some time to think it through.

Thank you @2010YOUY01! I will start a series of stack PRs and get them merged one by one. This will allow you to have more time for thinking 😆

@2010YOUY01

2010YOUY01 commented Jul 23, 2026

Copy link
Copy Markdown
Contributor

okay I got some idea how to implement it. Here are some thoughts:

Semantics to implement

  • What are the precise semantics of an ASOF join? I suspect that different systems, such as DuckDB and Snowflake, may differ slightly, so we should first define the semantics rigorously.

I propose Snowflake-style semantics: each left row selects the closest eligible right row within an optional equality-key group. >/>= selects the greatest right key; </<= selects the smallest. Left rows are always preserved, unmatched rows are NULL-padded, NULL keys do not match, and ties are nondeterministic.

I think it's a good idea.

DuckDB distinguishes between inner and outer ASOF joins, while Snowflake supports only one variant: left outer.

I don't fully understand why Snowflake made this decision, but perhaps it is the most common pattern in practice. In any case, it would be preferable to start simple.

Regarding ASOF join conditions, both DuckDB and Snowflake support a conjunction consisting of:

  • Exactly one inequality condition
  • Zero or more equality conditions

For example:

SELECT t.*, p.price
FROM trades t
ASOF JOIN prices p
ON t.symbol = p.symbol AND t.when >= p.when; -- 1 ie cond + 1 eq cond

We probably want to support the same join conditions.

Algorithm to implement

  • What is the high-level algorithm used by the ASOF join operator?

The initial operator uses an ordered merge: co-partition and order both inputs, then advance the right cursor monotonically while retaining the latest eligible candidate. This is O(L + R) after ordering with bounded join state. Other implementations could later share the same logical semantics.

Let's call this idea the repartition-based ASOF join. An alternative is the broadcast-based ASOF join, described below. I tend to think it's better to implement the broadcast-based algorithm in the first version, here are the reasons:

(TLDR: assuming its common to have workloads with only inequality condition, but no equality ASOF join condition, the broadcast approach can fully utilize all CPUs, while the repartition based approach can't)

# Broadcast ASOF Join

- Broadcast and sort the build side; repartition the probe side using batch-level round-robin partitioning, then sort each resulting partition.
- Perform a monotonic scan independently within each probe partition.

I think both approaches will be useful in the long term, as they target different workload patterns.

  • repartition-based: More efficient for large number of equality groups, no skew
  • broadcast-based: If there are no equality condition, or equality keys have very low cardinality, or equality key are heavily skewed, it performs better than repartition-based approach. (it seems the no-equality workloads are quite common)

For example, consider the join condition (t1.v1 < t2.v1) AND (t1.tag = t2.tag). The repartition-based algorithm requires both inputs to be repartitioned by tag before sorting and joining. If there are only two distinct tags, there will be only two effective partitions, leaving most CPU cores unused in a multicore environment. While the broadcast-based approach can fully utilize all the available CPU cores for the joining phase.

For workloads that can be perfectly repartitioned, the broadcast approach should not be much slower. Its main additional cost is maintaining and advancing more cursors over the broadcast side, which we could probably vectorize relatively easily.

Implementation plan

  1. Question: Are the current split PRs independently mergeable? 🤔 In other words, would merging the first one leave the main branch in a consistent and functional state, or are the PRs split primarily to make the implementation easier to read and review?

  2. Suggestion: Another useful principle might be to keep the first patch simple and defer aggressive low-level optimizations when they make the code substantially harder to follow.

  3. Suggestion: It would be great to write a top-level design document explaining the approach to readers without much background knowledge. It could start with an example query, explain the semantics, show the query-plan shape, and finally describe the algorithm inside the ASOF join operator. It should also summarize the major design decisions. This could be written as a top-level comment. Here is a good reference:

    /// `PiecewiseMergeJoinExec` is a join execution plan that only evaluates single range filter and show much

@xudong963

Copy link
Copy Markdown
Member

FYI, my teammate @jonathanc-n is also working on the AsOf join implementation in our fork repo massive-com#65. We really want the feature to be done, which will speed up some materialization cases in our internal. He might have some thoughts about the topic and the future collaboration.

@Xuanwo

Xuanwo commented Jul 27, 2026

Copy link
Copy Markdown
Member Author

Thanks @2010YOUY01, this is really helpful!

An alternative is the broadcast-based ASOF join, described below. I tend to think it's better to implement the broadcast-based algorithm in the first version, here are the reasons:

I initially chose the repartitioned merge to keep the join state bounded, but your point about equality-free joins collapsing to a single partition is convincing.

Let’s start with the broadcast-based ASOF join. We can keep the repartition-based algorithm as a follow-up for large right inputs or well-distributed equality keys, then use benchmarks to guide strategy selection.

Question: Are the current split PRs independently mergeable? 🤔 In other words, would merging the first one leave the main branch in a consistent and functional state, or are the PRs split primarily to make the implementation easier to read and review?

The stack PRs are intended to be mergeable in dependency order, with each step leaving main consistent. I’ll rework the physical PR first, update the dependent PRs afterward, and add the top-level design comment you suggested.

@Xuanwo

Xuanwo commented Jul 27, 2026

Copy link
Copy Markdown
Member Author

FYI, my teammate @jonathanc-n is also working on the AsOf join implementation in our fork repo massive-com#65. We really want the feature to be done, which will speed up some materialization cases in our internal. He might have some thoughts about the topic and the future collaboration.

Happy to know! Looking forward to collaborating with @jonathanc-n again (after iceberg-rust 😆).

Xuanwo added 13 commits July 27, 2026 16:22
# Conflicts:
#	datafusion/expr/src/logical_plan/plan.rs
# Conflicts:
#	datafusion/proto/src/physical_plan/mod.rs
# Conflicts:
#	benchmarks/src/asof.rs
#	datafusion/core/tests/sql/joins.rs
#	datafusion/expr/src/logical_plan/plan.rs
#	datafusion/physical-plan/benches/asof_join.rs
#	datafusion/physical-plan/src/joins/asof_join.rs
#	datafusion/proto/src/physical_plan/mod.rs
#	datafusion/substrait/tests/cases/serialize.rs
# Conflicts:
#	datafusion/proto/tests/cases/roundtrip_logical_plan.rs
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

auto detected api change Auto detected API change core Core DataFusion crate documentation Improvements or additions to documentation logical-expr Logical plan and expressions optimizer Optimizer rules physical-plan Changes to the physical-plan crate proto Related to proto crate sql SQL Planner sqllogictest SQL Logic Tests (.slt) substrait Changes to the substrait crate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

ASOF join support / Specialize Range Joins

5 participants