Skip to content

[SPARK-59616][SQL] Avoid copying every candidate row in the AS-OF join best-match scan - #58888

Open
david-mollitor-db wants to merge 1 commit into
apache:masterfrom
david-mollitor-db:asof-bestmatch-reuse
Open

david-mollitor-db wants to merge 1 commit into
apache:masterfrom
david-mollitor-db:asof-bestmatch-reuse

Conversation

@david-mollitor-db

Copy link
Copy Markdown
Contributor

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.copyMemorys 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.

…n best-match scan

### 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

Co-authored-by: Isaac <no-reply@databricks.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant