LukaZdravic opened a new pull request, #59957:
URL: https://github.com/apache/spark/pull/59957
### What changes were proposed in this pull request?
`SortMergeAsOfJoinExec` gets a cursor scan. For a backward or forward join,
it walks the right side once and keeps only the best right row so far. It does
not buffer the right-side group.
The operator checks once per query whether the join can use it. It needs all
of these:
- A backward or forward join, not nearest.
- No residual join condition left after pushdown.
- An as-of condition that compares the two sort keys, one on each side, of
the same type.
- A scalar key type, not STRUCT or ARRAY.
Every other join keeps the buffered scan. The new internal config
`spark.sql.join.sortMergeAsOfJoin.cursorScan.enabled` (default `true`) turns
the cursor scan off. The new metric `number of cursor scan tasks` counts the
tasks that use it.
### Why are the changes needed?
The buffered scan saves each right-side group, then rereads it from the
start for every left row. So a group with N left rows and M right rows can cost
up to N x M condition checks, and a large group spills to disk. Both sides are
already sorted by the as-of key, so the match only moves forward as the left
key grows. The cursor scan costs about N + M checks.
```sql
SELECT * FROM trades t LEFT ASOF JOIN quotes q
MATCH_CONDITION (t.ts >= q.ts) ON t.sym = q.sym
```
`AsOfJoinBenchmark`, 10,000 rows on each side, best time on my machine:
- 100 groups: 40 ms -> 34 ms
- 10 groups: 121 ms -> 30 ms
- No equi-key: 885 ms -> 18 ms
### Does this PR introduce _any_ user-facing change?
No. Query results stay the same. The join node in the SQL UI shows one new
metric.
### How was this patch tested?
New `SortMergeAsOfJoinScannerSuite` tests check that the cursor scan checks
the as-of condition only twice per left row, and which conditions
`supportsCursorScan` accepts and rejects.
New `AsOfJoinSortMergeSQLSuite` tests run each query with the cursor scan
off and on, and check both against the same expected rows. With a spill
threshold of one row, the cursor run must spill nothing. They cover ties, NULL
keys, scalar types, NaN and -0.0, expression sides, STRING against INT, mixed
TIMESTAMP_LTZ and TIMESTAMP_NTZ across a DST gap, and the joins that must keep
the buffered scan.
The existing DataFrame `joinAsOf` tests in `SortMergeAsOfJoinSuite` now run
with the cursor scan on. Its spill tests turn the cursor scan off, since it
never buffers.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
--
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]
