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

Is your feature request related to a problem or challenge?

AsOfJoinExec requires a single right partition and collects the whole right input into one reservation (collect_right_input). Because that reservation can't spill, any ASOF join fails with "Resources exhausted" once the right side is larger than the memory pool, even though the sorts below it can spill.

#23738 describes this trade-off ("the complete right input must fit in memory, and its rows may be scanned once per left partition") and says "a repartitioned ASOF implementation can be added later as a separate physical strategy".

Example query ran with datafusion-cli -m 6g -f repro.sql:

WITH left_input AS (
  SELECT value % 1000 AS key, value / 1000 + 1 AS ts
  FROM range(10000000)
),
right_input AS (
  SELECT value % 1000 AS key, value / 1000 AS ts, repeat('y', 64) AS payload
  FROM range(50000000)
)
SELECT count(*) AS n, sum(length(r.payload)) AS bytes
FROM left_input l
ASOF JOIN right_input r MATCH_CONDITION (l.ts >= r.ts)
ON l.key = r.key;

This gives the following error: "Failed to allocate additional 672.0 KB for AsOfJoinInput with 3.7 GB already allocated for this reservation"

Describe the solution you'd like

Add a partitioned execution mode for AsOfJoinExec when ON equality keys are present:

1. Require both inputs to be HashPartitioned on the equality keys and sorted within each partition (similar to SortMergeJoinExec).
2. In execute(), stream right partition i directly into AsOfJoinStream / InputCursor for left partition i instead of collecting and broadcasting the entire right side.

Since InputCursor already walks the right stream forward-only, streaming it per partition bounds the join's memory usage to the current batches + candidate row, while letting the upstream SortExec handle spilling. It also avoids re-scanning the entire right side once per left partition.

We can keep the current broadcast behavior for joins without equality keys (or when configured/chosen for small or skewed inputs) and expose the partitioned strategy via a config option / PartitionMode first.

Describe alternatives you've considered

We could make AsOfJoinInput spillable while keeping the broadcast design, but the right side would still funnel through a single partition (SortPreservingMergeExec) and every left partition would have to re-read the spilled right data from disk.

Additional context

I'm happy to work on a PR for this if the approach sounds good

Willingness to contribute

I plan to contribute to this feature independently