feat: support ASOF joins - #23738
Conversation
github-actions
Bot
added
sql
|
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 |
xudong963
left a comment
xudong963
left a comment
There was a problem hiding this comment.
Thanks for the PR, the implementation is a complete vertical slice, looks promising.
Some todos in my mind:
- 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.
- Add an
asof_join.sltcovering at least all four operators, grouped and ungrouped matching, nulls, USING, unmatched rows, invalid conditions, and EXPLAIN
github-actions
Bot
added
documentation
|
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:
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #23738 +/- ##
==========================================
- Coverage 81.31% 81.20% -0.12%
==========================================
Files 1117 1119 +2
Lines 395987 397457 +1470
Branches 395987 397457 +1470
==========================================
+ Hits 321993 322750 +757
- Misses 55177 55808 +631
- Partials 18817 18899 +82 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Thank you @2010YOUY01 joining the review. I agree we should settle the semantics and execution model first.
I propose Snowflake-style semantics: each left row selects the closest eligible right row within an optional equality-key group.
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
left a comment
alamb
left a comment
There was a problem hiding this comment.
looks neat -- I skimmed it quickly
| | LogicalPlan::Aggregate(_) | ||
| | LogicalPlan::Sort(_) | ||
| | LogicalPlan::Join(_) | ||
| | LogicalPlan::AsOfJoin(_) |
There was a problem hiding this comment.
Why does AsOfJoin get its own logical plan? Isn't the choice of join algorithm typically done via a physical plan?
There was a problem hiding this comment.
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.)
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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 🤔
There was a problem hiding this comment.
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
pricestable 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.
|
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:
|
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. |
Thanks for explaining the rationale! I’m interested in helping further with this feature, now I need some time to think it through. |
|
For @alamb
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.
Sure, will happy to do this. For @xudong963
Let's go. For @2010YOUY01
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 😆 |
|
okay I got some idea how to implement it. Here are some thoughts: Semantics to implement
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:
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 condWe probably want to support the same join conditions. Algorithm to implement
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) I think both approaches will be useful in the long term, as they target different workload patterns.
For example, consider the join condition 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
|
|
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. |
|
Thanks @2010YOUY01, this is really helpful!
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.
The stack PRs are intended to be mergeable in dependency order, with each step leaving |
Happy to know! Looking forward to collaborating with @jonathanc-n again (after iceberg-rust 😆). |
jonathanc-n
left a comment
jonathanc-n
left a comment
There was a problem hiding this comment.
Something I also ran into during my implementation was that for tiebreaks when several right rows in a group share the match value, the scan keeps advancing on the candidate, so you can get the last row in sort order which is arbitrary. This matches the behaviour in duckdb + snowflake but should probably be documented as a comment.
|
Thank you @jonathanc-n for the review, updated the full stack and this umbrella PR 😆 |
I think this split is a good idea. One issue is that I think it's better to use So I think it's likely we have to 1. first implement a simple v1 physical operator 2. Implement SQL interface, then add more tests and benchmark through SQL interface, and potentially do some fixes/changes in step 1. But in order to make progress on such a large feature, it's still a better choice I guess 🤔 I plan to continue reviewing #23828 (step1), and try to get it merged first. |
That makes sense. I have updated the stack and this umbrella PR. 😄 |
|
Progress update, for anyone interested in reviewing: #23828 is almost there, thanks to @Xuanwo! It includes only the basic physical operator. The next steps are:
|
This is great -- thank you @Xuanwo and @2010YOUY01 |
## Which issue does this PR close? - Part of #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>
7ce5dbd to
7da177a
Compare
6 participants
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 fullresponsibility 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/mainat0eecdfc44.#23828 has merged. The remaining PRs are intended to be independently mergeable
in dependency order, beginning with #24375 for floating-point equality keys,
then #23829 for logical planning and its frontend/serialization layers.
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 itsisolated fork-to-fork diff.
AsOfJoinExec, properties, statistics, metrics, unit tests, and conservative float-key rejectionjoin_asof,join_asof_using, prelude exports, and API testdfbenchworkloads and physical Criterion benchmarkWhich 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:
collected once, retained under a memory reservation, and shared immutably by
all left partitions. Each retained Arrow buffer is charged exactly once,
including when batches are zero-copy slices of the same allocation.
an independent merge cursor and scans the shared right-side batches, so output
partitioning is inherited from the left.
batch boundaries and output flushes. Group changes and EOF cannot reuse a
candidate from another equality group or drop the final eligible candidate.
<,<=,>, and>=, and NULL-pads unmatched right columns.feat: support floating-point ASOF equality keys #24375 then normalizes signed zero in the required equality-key ordering so
Float16, Float32, and Float64 sorting agrees with join equality.
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?
AsOfJoinExecand its state, properties, statistics,memory accounting, and metrics.
support.
closed until an ASOF extension is defined.
optimizer-inserted sort/repartition, wide and dictionary payloads, descending
successor matching, broadcast-side size asymmetry, skew, 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.shcargo clippy --all-targets --all-features -- -D warnings504 sqllogictest files
dfbench asofworkloads with one iteration and four partitionsThe benchmark runs validate the harness and expected row counts; they are not
presented as comparative performance claims because
maincannot plan theASOF workloads.
Are there any user-facing changes?
Yes. Users can construct ASOF joins through SQL,
LogicalPlanBuilder, andDataFrame. The initial contract is left-preserving, supports optionalequality keys plus one ordered match condition, and rejects unbounded inputs.
Float16, Float32, and Float64 equality keys treat signed zero consistently with
join equality. With
USING, wildcard output contains one unqualified equalitykey while both qualified input keys remain addressable. Existing join behavior
is unchanged.
Compatibility
The protobuf additions use new messages and append-only oneof/enum tags, so
existing wire tags are not reused. The generated Rust protobuf enums and the
public
LogicalPlanenum gain new variants; downstream exhaustive matches mustadd arms.
AsOfJoinis appended inLogicalPlanso existing variants retaintheir
PartialOrdordering, but the enum addition should still be reviewed as aRust source-compatibility break for a breaking DataFusion release.