[Kristofferjoha]

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]

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

[Kristofferjoha]

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.