feat: support ASOF JOIN SQL - #23830
Conversation
github-actions
Bot
added
documentation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #23830 +/- ##
==========================================
+ Coverage 81.57% 81.68% +0.10%
==========================================
Files 1123 1124 +1
Lines 410789 413484 +2695
Branches 410789 413484 +2695
==========================================
+ Hits 335120 337752 +2632
+ Misses 55913 55834 -79
- Partials 19756 19898 +142 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
dff33b2 to
fe21f45
Compare
fe21f45 to
c1f3e95
Compare
github-actions
Bot
removed
physical-expr
# Conflicts: # datafusion/expr/src/logical_plan/builder.rs # datafusion/expr/src/logical_plan/plan.rs
github-actions
Bot
removed
optimizer
github-actions
Bot
removed
proto
|
cc @2010YOUY01, this PR is good to go! 🚀 Also CC @jayzhan211: would you like to follow this stack's review? Happy to ping you if you're interested. 😄 |
Of course! |
jayzhan211
left a comment
jayzhan211
left a comment
There was a problem hiding this comment.
Thanks @Xuanwo Minor issues left, others LGTM
There was a problem hiding this comment.
Thank you, this is great as always. Here is some sugestions:
Before merging
- I left a few suggestions on the logical plan APIs — curious what you think. If we can agree on that, let's update them before merging.
- Add some simple tests for the unparser.
Potential follow-ups
- Consolidate the Rust integration tests into
slt. - More
sltcoverage. Below is the coverage gap suggested by AI:
Click to expand
ASOF slt: extra coverage
-
Outer WHERE on a right column must stay above the join
Only correctness property with no guard. If filter pushdown later learns
about ASOF and pushes this into the right input, candidate selection changes.
SELECT l.id, r.val FROM asof_left l ASOF JOIN asof_right r
MATCH_CONDITION (l.ts >= r.ts) ON l.grp = r.grp WHERE r.val <> 'a6';
Expect: EXPLAIN shows Filter above AsOf Join; ids 2,4,5 (id 3 dropped, not remapped to a4). -
Projection pruning through the node
Existing EXPLAINs select every column, so they can't show pruning.
EXPLAIN SELECT l.id FROM asof_left l ASOF JOIN asof_right r
MATCH_CONDITION (l.ts >= r.ts) ON l.grp = r.grp;
Expect: TableScan: asof_right projection=[grp, ts] -
SELECT * with USING
Docs promise the key appears once; file only selects named columns.
SELECT * FROM asof_left l ASOF JOIN asof_right r
MATCH_CONDITION (l.ts >= r.ts) USING (grp);
Expect: 5 columns (id, grp, ts, ts, val), not 6. -
Expression operands
Every test uses bare columns; side-ownership check on expressions is untested.
MATCH_CONDITION (l.ts >= r.ts + INTERVAL '1 second')
ON lower(l.grp) = lower(r.grp) -
Reversed equality sides in ON
The main thing asof_join_on adds; only unit-tested in the builder.
ON r.grp = l.grp -- same result as l.grp = r.grp -
Ordered types beyond timestamp/int
String, DATE vs TIMESTAMP, float, tz/precision mismatch, dictionary keys.
ON l.grp = r.grp where l.grp is arrow_cast(grp, 'Dictionary(Int32, Utf8)')
Dictionary matters most: Parquet produces it. -
Empty inputs
ASOF JOIN (SELECT * FROM asof_right WHERE false) r ... -- all r.* NULL
FROM (SELECT * FROM asof_left WHERE false) l ... -- zero rows -
Duplicate left rows emitted once each
FROM (SELECT * FROM asof_left UNION ALL SELECT * FROM asof_left) l ...
Expect: 14 rows, each id twice with same match. -
Self join
FROM asof_left a ASOF JOIN asof_left b MATCH_CONDITION (a.ts > b.ts) ON a.grp = b.grp -
Composition through the planner
ASOF then inner join, GROUP BY over output, CREATE VIEW, WITH ... AS.
... ASOF JOIN asof_right r MATCH_CONDITION (...) ON ... JOIN asof_right r2 ON r.val = r2.val -
USING with multiple keys
USING (venue, grp) -
Unpinned error messages
Rust test only asserts "is_err"; none of these messages are checked anywhere.
MATCH_CONDITION (1 >= r.ts) -> left operand must reference only the left input
ON 1 = 1 -> must compare one left expression with one right
MATCH_CONDITION (l.ts >= r.ts AND ...) -> requires <, <=, >, or >=, found AND
ON l.grp = l.grp -> must compare one left expression with one right
ON l.grp = r.grp AND l.ts >= r.ts -> accepts only equality conditions combined with AND
SELECT ts ... USING (grp) -> Ambiguous reference to unqualified field ts
unbounded left table -> AsOfJoinExec requires bounded inputs
Note: NATURAL ASOF JOIN is rejected by sqlparser, so the planner branch is unreachable. -
EXPLAIN for the other two forms
USING shows constraint=Using; no-key form shows on=[] in the physical plan. -
Comment on deferred optimizations
Filter/limit pushdown into the left input and order reuse by ORDER BY
don't happen yet. A note saves the next person from thinking it's a bug.
| Ok(()) | ||
| } | ||
|
|
||
| fn register_asof_test_tables(ctx: &SessionContext) -> Result<()> { |
There was a problem hiding this comment.
Can we unify those tests into sqllogictests, this way I think it's easier to maintain. Unless there are some setups that is not possible to do in slt, we might want extra coverage here.
In order to move those tests checking partition properties, I think we can assert the end behavior (final plan shape), rather than internal properties.
There was a problem hiding this comment.
Agreed for the SQL-level cases. I kept the Rust tests that need multi-partition plans, unbounded inputs, or plan-to-SQL roundtrips. I’ll consolidate the remaining overlap in a follow-up.
|
|
||
| // Keep ASOF-specific locals out of the recursive plan unparser's stack frame. | ||
| #[inline(never)] | ||
| fn asof_join_to_sql( |
There was a problem hiding this comment.
I don't fully understand the unparser now, but my AI tool suggest it is good to go, with some cleanup advices:
AI review
The unparser is heavy relative to the join it mirrors. About 120 lines of branching on "already projected" versus not, and "needs a derived subquery" versus not, for each side. The regular join unparser has the same structural problem, so this is inherited rather than introduced. A cleaner alternative would be a small shared helper that takes a join input and returns both the relation and its projection items, used by both join kinds. That refactor is bigger than this PR should carry, but it would be a natural follow-up given the branch already touched the shared nesting helper.
There was a problem hiding this comment.
Will treat this change as a follow up
| MATCH_CONDITION (l.ts = r.ts) | ||
| ON l.grp = r.grp; | ||
|
|
||
| query error ASOF MATCH_CONDITION left operand must reference only the left input |
There was a problem hiding this comment.
Should this query get normalized internally, and make it valid? 🤔
There was a problem hiding this comment.
I’d keep this invalid. Snowflake defines MATCH_CONDITION operand order semantically, so the left input must remain on the left. ON equality operands are still accepted in either order.
There was a problem hiding this comment.
I see, that's the snowflake behavior.
|
Thank you @jayzhan211 @2010YOUY01 for the review! I have updated the builder API and added direct unparser coverage, including nested ASOF inputs. I kept the setup-dependent Rust tests here and will handle broader SLT consolidation separately. |
|
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 |
Sounds like a good plan! |
* feat: add ASOF join physical operator (apache#23828) ## Which issue does this PR close? - Part of apache#318. - Umbrella PR: apache#23738. - Follow-up for floating-point equality keys: apache#24375. ## 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 floating-point equality, 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. Floating-point equality keys are rejected in this base layer because Arrow's required sort order distinguishes `-0.0` from `+0.0` while join equality does not. apache#24375 adds the required ordering normalization as an independently reviewable layer. ## 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. - Reject floating-point equality keys until apache#24375 supplies a sort/equality contract that handles signed zero correctly. - Add physical operator tests covering match directions, equality groups, batch boundaries, unmatched rows, invalid contracts, shared-buffer memory accounting, multi-partition broadcast execution, and float-key rejection. ## 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 - FFI integration tests ## Are there any user-facing changes? This adds a new physical operator API. The base operator deliberately rejects floating-point equality keys; apache#24375 adds full Float16, Float32, and Float64 support. SQL and DataFrame APIs are left to later dependent PRs. --------- Co-authored-by: Yongting You <2010youy01@gmail.com> * feat: add ASOF join logical semantics (apache#23829) - Part of apache#318. - Umbrella PR: apache#23738. - Depends on apache#23828 (merged). This is the logical-planning layer of the ASOF JOIN stack. It defines the logical contract and planner behavior separately from the SQL frontend and serialization formats. logical layer. It no longer depends on the optional floating-point follow-up - Add `LogicalPlan::AsOfJoin`, `AsOfJoin`, and `AsOfMatch`. - Validate deterministic expressions, input ownership, supported match operators, equality-key types, and USING constraints. - Add `LogicalPlanBuilder` entry points and schema construction that preserves both qualified `USING` keys while exposing one unqualified wildcard key. - Integrate ASOF joins with tree transforms, display, type coercion, projection pruning, row bounds, and physical planning. - Plan logical ASOF joins to the broadcast-based `AsOfJoinExec` from - Fail closed at proto, SQL unparser, and Substrait boundaries until their owning stack layers add explicit support. - Defer ASOF-specific functional-dependency refinement to apache#24799 and filter pushdown to apache#24801 so each optimization can be reviewed independently. Yes: - `cargo fmt --all` - `cargo clippy --all-targets --all-features -- -D warnings` - `cargo test -p datafusion-expr min_rows_of_joins --all-features` - `cargo test -p datafusion-substrait asof_join_fails_closed_until_substrait_has_an_extension --all-features` - The extended workspace test command from the contributor guide This adds logical-plan and builder APIs for ASOF joins. SQL syntax, DataFrame APIs, and plan serialization are intentionally left to dependent stack PRs. Floating equality keys remain rejected by the merged physical operator unless the independent follow-up apache#24375 is also included. As with any new public `LogicalPlan` variant, downstream exhaustive matches must add an arm. The variant is appended so existing variants retain their `PartialOrd` ordering; maintainers should still treat the enum addition as a Rust source-compatibility break. This PR can be reviewed independently now that apache#23828 has merged. The optimization follow-ups apache#24799 and apache#24801 are not required by the core ASOF stack. * feat: support ASOF JOIN SQL (apache#23830) - Part of apache#318. - Umbrella PR: apache#23738. - Builds on apache#23829 and apache#23828, both merged. This is the SQL frontend layer of the ASOF JOIN stack. It adds syntax and unparsing on top of the merged physical and logical contracts. The PR now contains only the isolated SQL frontend diff. - Plan `ASOF JOIN ... MATCH_CONDITION (...)` with optional `ON` or `USING` equality keys. - Reject unsupported match shapes and non-equality `ON` predicates. - Unparse ASOF joins while preserving right-side candidate preselection and nested join scope. - Document the supported SQL syntax and semantics, including qualified `USING` keys, left-partitioned broadcast execution, the full-right memory requirement, repeated scans, and the absence of spill/repartitioned ASOF. - Add SQL integration and sqllogictest coverage for all four match directions, coercion, equality-free joins, USING, invalid contracts, EXPLAIN, boundedness, and optimized-plan round trips. - Verify the broadcast topology with a multi-partition left input: left partitioning is preserved while the right input is single-partitioned. Yes: - `cargo fmt --all` - `./ci/scripts/doc_prettier_check.sh --write --allow-dirty` - `cargo clippy --all-targets --all-features -- -D warnings` - `cargo test -p datafusion --test core_integration asof --all-features` - `cargo test -p datafusion-sqllogictest --test sqllogictests --all-features -- asof_join` - The extended workspace test command from the contributor guide Users can express Snowflake-style ASOF joins in SQL with `MATCH_CONDITION`, optional equality keys, and `<`, `<=`, `>`, or `>=` match directions. With `USING`, wildcard output exposes one unqualified key while both qualified input keys remain addressable. The user guide also documents the initial broadcast strategy and its memory/no-spill limitations. stack and does not depend on the optional floating-point follow-up apache#24375. --------- Co-authored-by: Xuanwo <github@xuanwo.io> Co-authored-by: Yongting You <2010youy01@gmail.com>
4 participants
Which issue does this PR close?
Rationale for this change
This is the SQL frontend layer of the ASOF JOIN stack. It adds syntax and
unparsing on top of the merged physical and logical contracts. The PR now
contains only the isolated SQL frontend diff.
What changes are included in this PR?
ASOF JOIN ... MATCH_CONDITION (...)with optionalONorUSINGequality keys.
ONpredicates.nested join scope.
USINGkeys, left-partitioned broadcast execution, the full-right memoryrequirement, repeated scans, and the absence of spill/repartitioned ASOF.
coercion, equality-free joins, USING, invalid contracts, EXPLAIN, boundedness,
and optimized-plan round trips.
partitioning is preserved while the right input is single-partitioned.
Are these changes tested?
Yes:
cargo fmt --all./ci/scripts/doc_prettier_check.sh --write --allow-dirtycargo clippy --all-targets --all-features -- -D warningscargo test -p datafusion --test core_integration asof --all-featurescargo test -p datafusion-sqllogictest --test sqllogictests --all-features -- asof_joinAre there any user-facing changes?
Users can express Snowflake-style ASOF joins in SQL with
MATCH_CONDITION, optional equality keys, and<,<=,>, or>=matchdirections. With
USING, wildcard output exposes one unqualified key whileboth qualified input keys remain addressable. The user guide also documents
the initial broadcast strategy and its memory/no-spill limitations.
#23829 and #23828 are merged. This is the next core layer in the ASOF stack and
does not depend on the optional floating-point follow-up #24375.