Skip to content

feat: support partitioned execution for ASOF joins - #26592

Closed
Kristofferjoha wants to merge 6 commits into
apache:mainfrom
Kristofferjoha:feat/asof-partitioned-execution
Closed

feat: support partitioned execution for ASOF joins#26592
Kristofferjoha wants to merge 6 commits into
apache:mainfrom
Kristofferjoha:feat/asof-partitioned-execution

Conversation

@Kristofferjoha

@Kristofferjoha Kristofferjoha commented Oct 22, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Closes #26478.

Rationale for this change

ASOF joins currently collect the complete right input into a non-spillable reservation. Queries fail when that input exceeds the memory pool, even when upstream sorts can spill. Each left partition also scans the collected right input independently.

What changes are included in this PR?

  • Add AsOfJoinMode::Partitioned, requiring both inputs to be co-partitioned on equality keys and sorted within each partition. Each left partition streams its corresponding right partition.
  • Account for retained streaming buffers in the memory pool, including candidate matches and pending output rows, without counting shared buffers twice.
  • Preserve the mode through projection, child replacement, and protobuf/JSON serialization. Older serialized plans default to broadcast.
  • Add the opt-in datafusion.optimizer.repartition_asof_joins option, defaulting to false. Joins without equality keys continue to use broadcast execution.

The join itself does not spill: upstream sorts handle spilling, and the join still needs memory for its active working set. Skewed keys can reduce parallelism.

Existing ASOF SQL benchmarks compared with local main (6c1727fcd), using four partitions and 12 measured samples per case after warm-up:

Case Main median Partitioned median
Q02: balanced equality keys 183.55 ms 108.14 ms
Q03: wide payload 63.86 ms 46.93 ms
Q06: skewed equality keys 175.70 ms 335.76 ms
Q11: large right input 2687.45 ms 1045.61 ms

What is the testing strategy for this PR?

  • SQL logic tests cover all four match operators, composite and computed equality keys, NULLs, empty inputs, ties, and small batches.
  • SQL integration tests check strategy selection, input distribution and ordering, plan display, and query results.
  • A memory-limit integration test verifies that partitioned execution succeeds with spilled sort output under a budget that rejects broadcast collection.
  • Unit tests cover partition selection, mode preservation, buffer accounting, bounded streaming, and reservation cleanup on completion, cancellation, and error.
  • Protobuf and JSON tests cover mode round trips, compatibility with missing mode fields, and invalid serialized modes.

Are there any user-facing changes?

Yes. Enable partitioned execution for ASOF joins with equality keys using:

SET datafusion.optimizer.repartition_asof_joins = true;

The default execution strategy remains broadcast. The SQL guide and generated configuration reference document the option.

AI-assisted: I used an AI coding assistant while preparing this PR. I have reviewed and understand all changes.

@github-actions github-actions Bot added documentation Improvements or additions to documentation core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) common Related to common crate proto Related to proto crate physical-plan Changes to the physical-plan crate labels Oct 22, 2026
@Kristofferjoha
Kristofferjoha marked this pull request as ready for review October 22, 2026 17:08
@github-actions

github-actions Bot commented Oct 22, 2026 •
edited
Loading

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 v55.1.0 (current)
       Built [  48.430s] (current)
     Parsing datafusion v55.1.0 (current)
      Parsed [   0.030s] (current)
    Building datafusion v55.1.0 (baseline)
       Built [  47.578s] (baseline)
     Parsing datafusion v55.1.0 (baseline)
      Parsed [   0.030s] (baseline)
    Checking datafusion v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.806s] 229 checks: 229 pass, 31 skip
     Summary no semver update required
    Finished [  98.485s] datafusion
    Building datafusion-common v55.1.0 (current)
       Built [  28.600s] (current)
     Parsing datafusion-common v55.1.0 (current)
      Parsed [   0.066s] (current)
    Building datafusion-common v55.1.0 (baseline)
       Built [  28.742s] (baseline)
     Parsing datafusion-common v55.1.0 (baseline)
      Parsed [   0.066s] (baseline)
    Checking datafusion-common v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.871s] 229 checks: 228 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.51.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field OptimizerOptions.repartition_asof_joins in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:1620

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  59.226s] datafusion-common
    Building datafusion-physical-plan v55.1.0 (current)
       Built [  33.603s] (current)
     Parsing datafusion-physical-plan v55.1.0 (current)
      Parsed [   0.197s] (current)
    Building datafusion-physical-plan v55.1.0 (baseline)
       Built [  33.453s] (baseline)
     Parsing datafusion-physical-plan v55.1.0 (baseline)
      Parsed [   0.198s] (baseline)
    Checking datafusion-physical-plan v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.857s] 229 checks: 229 pass, 31 skip
     Summary no semver update required
    Finished [  69.680s] datafusion-physical-plan
    Building datafusion-proto v55.1.0 (current)
       Built [  45.196s] (current)
     Parsing datafusion-proto v55.1.0 (current)
      Parsed [   0.011s] (current)
    Building datafusion-proto v55.1.0 (baseline)
       Built [  44.876s] (baseline)
     Parsing datafusion-proto v55.1.0 (baseline)
      Parsed [   0.012s] (baseline)
    Checking datafusion-proto v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.130s] 229 checks: 229 pass, 31 skip
     Summary no semver update required
    Finished [  91.255s] datafusion-proto
    Building datafusion-proto-models v55.1.0 (current)
       Built [  21.700s] (current)
     Parsing datafusion-proto-models v55.1.0 (current)
      Parsed [   0.138s] (current)
    Building datafusion-proto-models v55.1.0 (baseline)
       Built [  21.727s] (baseline)
     Parsing datafusion-proto-models v55.1.0 (baseline)
      Parsed [   0.141s] (baseline)
    Checking datafusion-proto-models v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   2.347s] 229 checks: 228 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.51.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field AsOfJoinExecNode.mode in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:2801
  field AsOfJoinExecNode.mode in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:2801

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  47.228s] datafusion-proto-models

@github-actions github-actions Bot added the auto detected api change Auto detected API change label Oct 22, 2026
@Kristofferjoha

Kristofferjoha commented Oct 22, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

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

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

This is my first contribution. I am unsure as to how to handle this.

@alamb

alamb commented Oct 29, 2026

Copy link
Copy Markdown
Contributor

Thanks for the contribution! I am closing this under DataFusion's AI Dump policy.

https://datafusion.apache.org/contributor-guide/ai-policy.html#what-is-an-ai-dump-and-why-it-is-not-helpful

This has the hallmarks of AI code that was not reviewed by the author: a 1,400-line new join mode with config and proto changes and no design discussion before implementation. Please open an issue to discuss the design first; we would welcome that.

If you would like to continue, please open a focused PR in your own words that you have reviewed and understand, and we are happy to help with review.

@alamb alamb closed this Oct 29, 2026
@Kristofferjoha Kristofferjoha mentioned this pull request Oct 29, 2026
Open
@2010YOUY01

2010YOUY01 commented Oct 29, 2026

Copy link
Copy Markdown
Member

@Kristofferjoha Thank you for the contribution. Sorry if our judgment about AI contributions has become too aggressive given the current challenges. I don't think this contribution is an AI dump, judging from the writeup.

However, personally I won't prioritize reviewing this PR, this might be too ambitious for a new contributor. This work requires a deep understanding of the entire system, so I'd prefer to review it only when it comes from contributors I've worked with before.

Sorry if this sounds overly conservative, but we haven't figured out a better approach given the current volume of AI-generated contributions. I'd suggest starting with smaller contributions to build trust with existing maintainers before taking on such a large feature in the future. (until we've figured out a better approach 😮‍💨 )

@Kristofferjoha

Kristofferjoha commented Oct 30, 2026

Copy link
Copy Markdown
Contributor Author

@2010YOUY01 Makes sense. I'll start with some smaller stuff to learn without it coming at the expense of the maintainers' time.

@Kristofferjoha Kristofferjoha mentioned this pull request Oct 30, 2026
Open
@alamb

alamb commented Oct 30, 2026

Copy link
Copy Markdown
Contributor

Yes, thank you for understanding. We are really struggling to keep up with the influx of PRs and I agree some may be being closed to aggressively

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

auto detected api change Auto detected API change common Related to common crate core Core DataFusion crate documentation Improvements or additions to documentation physical-plan Changes to the physical-plan crate proto Related to proto crate sqllogictest SQL Logic Tests (.slt)

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

Support hash-partitioned execution in AsOfJoinExec for memory-bounded joins

3 participants

@Kristofferjoha @alamb @2010YOUY01