From f46e28b8cc46fde5e71cf4dad8887e4c2257ed92 Mon Sep 17 00:00:00 2001 From: David Mollitor Date: Thu, 17 Sep 2026 03:10:13 +0000 Subject: [PATCH] [MINOR][SQL] Hoist loop-invariant left row out of the AS-OF join inner scan ### What changes were proposed in this pull request? In `SortMergeAsOfJoinScanner` (`sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala`), the per-left-row scan over the buffered right group (`findBestBackwardForward` and `findBestForwardNearest`) rebinds both sides of the shared `JoinedRow` on every iteration via `joinedRow.withLeft(leftRow).withRight(rightRow)`. The left row is fixed for the duration of the scan, so this binds `joinedRow.withLeft(leftRow)` once before the loop and keeps only `joinedRow.withRight(rightRow)` inside it. ### Why are the changes needed? The inner scan runs once per buffered right row for every left row, so `withLeft` was called redundantly on each iteration with an argument that never changes within a scan. JFR profiling of `AsOfJoinBenchmark` showed `JoinedRow.withLeft`/`withRight` among the hotter frames in the (interpreted) scanner. Binding the left side once removes the redundant per-row call. This is behavior-preserving: `JoinedRow.withLeft` only stores the left-row reference, nothing mutates the left side within a scan, and `findNext` re-binds both sides before projecting the output row. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing tests pass: `SortMergeAsOfJoinSuite`, `DataFrameAsOfJoinSuite`, `AsOfJoinSQLSuite`, and `AsOfJoinSortMergeSQLSuite` (89 tests). ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Isaac Co-authored-by: Isaac --- .../spark/sql/execution/joins/SortMergeAsOfJoinExec.scala | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala index 5bdfbca78abfa..94cea8a81ab71 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala @@ -386,9 +386,11 @@ private[joins] class SortMergeAsOfJoinScanner( var bestMatch: InternalRow = null val iter = rightGroupBuffer.generateIterator() + // `leftRow` is fixed for this scan, so bind it once; only the right side changes per row. + joinedRow.withLeft(leftRow) while (iter.hasNext) { val rightRow = iter.next() - joinedRow.withLeft(leftRow).withRight(rightRow) + joinedRow.withRight(rightRow) val asOfSatisfied = boundAsOfCond.eval(joinedRow) if (asOfSatisfied != null && asOfSatisfied.asInstanceOf[Boolean]) { @@ -418,9 +420,11 @@ private[joins] class SortMergeAsOfJoinScanner( var bestDistance: Any = null val iter = rightGroupBuffer.generateIterator() + // `leftRow` is fixed for this scan, so bind it once; only the right side changes per row. + joinedRow.withLeft(leftRow) while (iter.hasNext) { val rightRow = iter.next() - joinedRow.withLeft(leftRow).withRight(rightRow) + joinedRow.withRight(rightRow) val asOfSatisfied = boundAsOfCond.eval(joinedRow) if (asOfSatisfied != null && asOfSatisfied.asInstanceOf[Boolean]) {