david-mollitor-db opened a new pull request, #58888:
URL: https://github.com/apache/spark/pull/58888
### What changes were proposed in this pull request?
`SortMergeAsOfJoinScanner`
(`sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala`)
materialized the best match with `bestMatch = rightRow.copy()` while
scanning the buffered right
group. Under a backward (last-match-wins) join this copies every
as-of-satisfying candidate and
discards all but the last; forward/nearest copies on each improvement. This
reuses a single
detached holder for the best match, copying the winning row's bytes in place:
- Add a scanner-scoped `UnsafeRow bestMatchRow` and a growable backing
`byte[] bestMatchBuffer`.
- A `retainBestMatch(row)` helper `Platform.copyMemory`s the row into the
(grown-as-needed) buffer
and points the holder at it.
- Both `bestMatch = rightRow.copy()` sites (`findBestBackwardForward`,
`findBestForwardNearest`)
become `bestMatch = retainBestMatch(rightRow)`.
This turns O(candidates) allocations into ~O(1): for fixed-width right rows
the buffer is allocated
once and reused; variable-width grows to the max and reuses.
### Why are the changes needed?
`SortMergeAsOfJoinScanner` has no whole-stage codegen, so its inner scan is
always interpreted.
JFR profiling of `AsOfJoinBenchmark` (sort-merge cases isolated) showed
`UnsafeRow.copy()` at ~18%
of CPU and ~61% of allocation (`UnsafeRow` 36% + `byte[]` 25%): each
`copy()` allocates a fresh
`UnsafeRow` and `byte[]`. After the change `UnsafeRow.copy()` drops to ~0.8%
of CPU (replaced by a
~2.5% `Platform.copyMemory`), the `UnsafeRow` allocation class leaves the
hot-class list, and
`byte[]` allocation drops from ~25% to ~5.5%.
A copy is still required (not just retaining the reference): the right
group's
`ExternalAppendOnlyUnsafeRowArray` returns a stable instance from its
in-memory iterator but a
single reused `UnsafeRow` from its spill iterator, and the buffer is
cleared/refilled between left
rows. The reused holder keeps that safety (it owns a detached `byte[]`)
while removing the
per-candidate allocation. The holder's footprint is one row's width and is
scanner-scoped (one
scanner per partition, freed at task completion), so it is not
row-count-proportional and needs no
shrink logic.
### Does this PR introduce _any_ user-facing change?
No. Results are identical; the holder is consumed by
`resultProjection(...).copy()` in `findNext`
before the next scan can overwrite it.
### How was this patch tested?
Existing tests pass: `SortMergeAsOfJoinSuite`, `DataFrameAsOfJoinSuite`,
`AsOfJoinSQLSuite`, and
`AsOfJoinSortMergeSQLSuite`. The suite's existing "spill to disk" cases
(backward/forward/nearest/
left outer) exercise the reused-instance spill iterator -- the exact path
where the detached copy
is mandatory. Before/after JFR on `AsOfJoinBenchmark` confirms the
allocation reduction.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac
This pull request and its description were written by Isaac.
--
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]
