Kristofferjoha opened a new issue, #26478:
URL: https://github.com/apache/datafusion/issues/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`:
```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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
