[SPARK-59716][SQL] Skip the per-match UnsafeRow copy when the AS-OF join right-side buffer is in memory

### What changes were proposed in this pull request?

`SortMergeAsOfJoinScanner` rescans the buffered right-side group once per left row, and did
`bestMatch = rightRow.copy()` for every right row that improved on the match so far, although only
the final one is ever used.

That copy is only needed when `ExternalAppendOnlyUnsafeRowArray` has switched to its spillable
backing store, whose iterator re-points a single `UnsafeRow` on every `next()`. While the buffer is
in memory the iterator returns the distinct rows it stores, so a retained match stays valid.

This PR exposes that as `ExternalAppendOnlyUnsafeRowArray.isSpillBacked` and copies a match only
when it is true. The flag is read per scan rather than cached, because `clear()` drops the spillable
backing store between equi-key groups.

### Why are the changes needed?

The copies are pure overhead on the common, non-spilled path, and there are `O(matches)` of them per
left row.

`AsOfJoinBenchmark`, `Best Time(ms)` of the "Sort-merge AS-OF join" case, from the regenerated
result files in this PR. `Improvement` is the reduction in best time.

`AS-OF Join (left=10000, right=10000, groups=100)`

| JDK | Before | After | Improvement |
|---|---|---|---|
| 17 | 53 | 44 | 17.0% |
| 21 | 56 | 49 | 12.5% |
| 25 | 53 | 53 | 0.0% |

`AS-OF Join (left=10000, right=10000, groups=10)`

| JDK | Before | After | Improvement |
|---|---|---|---|
| 17 | 170 | 110 | 35.3% |
| 21 | 181 | 120 | 33.7% |
| 25 | 180 | 122 | 32.2% |

`AS-OF Join no equi-key (left=10000, right=10000)`

| JDK | Before | After | Improvement |
|---|---|---|---|
| 17 | 1378 | 804 | 41.7% |
| 21 | 1466 | 870 | 40.7% |
| 25 | 1457 | 942 | 35.3% |

The `groups=100` case buffers only ~100 right rows per group, so the scan is a small part of an
end-to-end time dominated by the shuffle and sort, and the result sits inside its own stdev either
way. The larger-group cases, where the per-left-row scan actually dominates, improve by 32-42%.

Reusing one `UnsafeRow` and copying the winning row's bytes into it was measured too, and did not
help: it removes the allocation but keeps the memory copy, which is what costs here.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Pass the CIs.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 5

Closes #58888

Closes #58971 from dongjoon-hyun/SPARK-59716.

Authored-by: Dongjoon Hyun <dongjoon@apache.org>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: dongjoon-hyun

sql/core/benchmarks/AsOfJoinBenchmark-jdk21-results.txt

@@ -6,21 +6,21 @@ OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join (left=10000, right=10000, groups=100):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 --------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                            37055          37454         564          0.0     3705548.9       1.0X
-Sort-merge AS-OF join                                        56             80          20          0.2        5593.8     662.4X
+Correlated subquery (baseline)                            39014          39335         454          0.0     3901384.8       1.0X
+Sort-merge AS-OF join                                        49             69          15          0.2        4941.2     789.6X
 
 OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join (left=10000, right=10000, groups=10):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 -------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                           37594          37599           6          0.0     3759403.2       1.0X
-Sort-merge AS-OF join                                      181            192          11          0.1       18128.5     207.4X
+Correlated subquery (baseline)                           40047          40121         104          0.0     4004730.2       1.0X
+Sort-merge AS-OF join                                      120            124           3          0.1       11967.3     334.6X
 
 OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join no equi-key (left=10000, right=10000):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 --------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                            24398          24433          50          0.0     2439779.8       1.0X
-Sort-merge AS-OF join                                      1466           1473          10          0.0      146583.2      16.6X
+Correlated subquery (baseline)                            23948          24018          99          0.0     2394812.4       1.0X
+Sort-merge AS-OF join                                       870            876           7          0.0       86954.2      27.5X
 
 

sql/core/benchmarks/AsOfJoinBenchmark-jdk25-results.txt

@@ -6,21 +6,21 @@ OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join (left=10000, right=10000, groups=100):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 --------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                            40348          40452         148          0.0     4034755.7       1.0X
-Sort-merge AS-OF join                                        53             72          12          0.2        5328.1     757.3X
+Correlated subquery (baseline)                            39527          39582          77          0.0     3952677.7       1.0X
+Sort-merge AS-OF join                                        53             66          12          0.2        5296.1     746.3X
 
 OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join (left=10000, right=10000, groups=10):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 -------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                           38920          38926           9          0.0     3891989.2       1.0X
-Sort-merge AS-OF join                                      180            194          10          0.1       18010.8     216.1X
+Correlated subquery (baseline)                           38087          38181         132          0.0     3808724.6       1.0X
+Sort-merge AS-OF join                                      122            134          10          0.1       12168.4     313.0X
 
 OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join no equi-key (left=10000, right=10000):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 --------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                            24916          24985          98          0.0     2491579.0       1.0X
-Sort-merge AS-OF join                                      1457           1466          13          0.0      145736.3      17.1X
+Correlated subquery (baseline)                            26714          26883         239          0.0     2671359.5       1.0X
+Sort-merge AS-OF join                                       942            959          15          0.0       94248.7      28.3X
 
 

sql/core/benchmarks/AsOfJoinBenchmark-results.txt

@@ -6,21 +6,21 @@ OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join (left=10000, right=10000, groups=100):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 --------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                            37227          37450         315          0.0     3722681.0       1.0X
-Sort-merge AS-OF join                                        53             65           9          0.2        5338.2     697.4X
+Correlated subquery (baseline)                            37503          37642         196          0.0     3750289.2       1.0X
+Sort-merge AS-OF join                                        44             56           9          0.2        4397.5     852.8X
 
 OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join (left=10000, right=10000, groups=10):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 -------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                           37115          37195         114          0.0     3711469.3       1.0X
-Sort-merge AS-OF join                                      170            177           4          0.1       17028.0     218.0X
+Correlated subquery (baseline)                           37895          38037         201          0.0     3789498.9       1.0X
+Sort-merge AS-OF join                                      110            117           5          0.1       10975.9     345.3X
 
 OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
 AMD EPYC 7763 64-Core Processor
 AS-OF Join no equi-key (left=10000, right=10000):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 --------------------------------------------------------------------------------------------------------------------------------
-Correlated subquery (baseline)                            25422          25490          96          0.0     2542213.0       1.0X
-Sort-merge AS-OF join                                      1378           1379           2          0.0      137830.4      18.4X
+Correlated subquery (baseline)                            23963          24038         107          0.0     2396300.0       1.0X
+Sort-merge AS-OF join                                       804            809           5          0.0       80401.4      29.8X
 
 

sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala

@@ -99,6 +99,17 @@ class ExternalAppendOnlyUnsafeRowArray(
 
   def isEmpty: Boolean = numRows == 0
 
+  /**
+   * Whether this array has switched to the [[UnsafeExternalSorter]] backing store (which may
+   * not have written anything to disk yet, see [[spillSize]] for that).
+   *
+   * This decides the row ownership contract of [[generateIterator]]: while it is false the
+   * iterator yields the distinct [[UnsafeRow]]s stored in the in-memory buffer, while once it
+   * is true the iterator re-points a single [[UnsafeRow]] on every `next()`, so a caller that
+   * retains a row past the current iteration step has to copy it.
+   */
+  def isSpillBacked: Boolean = spillableArray != null
+
   /**
    * Total number of bytes that has been spilled into disk so far.
    */

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala

@@ -376,6 +376,18 @@ private[joins] class SortMergeAsOfJoinScanner(
     }
   }
 
+  /**
+   * Retains `rightRow` as the best match found so far.
+   *
+   * While the right-side buffer is held in memory, its iterator yields the distinct rows stored
+   * in the buffer and a match can be retained as is. Once the buffer is spill-backed the
+   * iterator re-points a single [[UnsafeRow]] on every `next()`, so the match has to be copied
+   * out before the scan advances. `needsCopy` is read once per scan by the callers.
+   */
+  private def retainMatch(rightRow: UnsafeRow, needsCopy: Boolean): UnsafeRow = {
+    if (needsCopy) rightRow.copy() else rightRow
+  }
+
   /**
    * Forward scan for Backward joins: last-match-wins.
    * Buffer is sorted ascending by as-of key. For left.t >= right.t,
@@ -385,6 +397,7 @@ private[joins] class SortMergeAsOfJoinScanner(
   private def findBestBackwardForward(leftRow: InternalRow): InternalRow = {
     var bestMatch: InternalRow = null
     val iter = rightGroupBuffer.generateIterator()
+    val needsCopy = rightGroupBuffer.isSpillBacked
 
     joinedRow.withLeft(leftRow)
     while (iter.hasNext) {
@@ -399,7 +412,7 @@ private[joins] class SortMergeAsOfJoinScanner(
         }
         if (residualSatisfied) {
           // Last match wins (closest right.t to left.t)
-          bestMatch = rightRow.copy()
+          bestMatch = retainMatch(rightRow, needsCopy)
         }
       } else if (bestMatch != null) {
         // as-of condition transitioned true -> false (monotone for Backward).
@@ -418,6 +431,7 @@ private[joins] class SortMergeAsOfJoinScanner(
     var bestMatch: InternalRow = null
     var bestDistance: Any = null
     val iter = rightGroupBuffer.generateIterator()
+    val needsCopy = rightGroupBuffer.isSpillBacked
 
     joinedRow.withLeft(leftRow)
     while (iter.hasNext) {
@@ -434,7 +448,7 @@ private[joins] class SortMergeAsOfJoinScanner(
           val distance = boundOrderExpr.eval(joinedRow)
           if (distance != null) {
             if (bestMatch == null || distanceOrdering.lt(distance, bestDistance)) {
-              bestMatch = rightRow.copy()
+              bestMatch = retainMatch(rightRow, needsCopy)
               bestDistance = distance
             } else {
               // Distance is increasing past the minimum. For Forward,

sql/core/src/test/scala/org/apache/spark/sql/SortMergeAsOfJoinSuite.scala

@@ -820,4 +820,41 @@ class SortMergeAsOfJoinSuite extends QueryTest
       )
     }
   }
+
+  test("backward join - in-memory group followed by a spill-backed one") {
+    // `clear()` drops the spillable backing store between equi-key groups, so one scanner can
+    // see an in-memory group (whose buffer iterator yields the distinct stored rows) and then a
+    // spill-backed one (whose iterator re-points a single UnsafeRow on every next(), so that a
+    // retained match has to be copied out). Group "A" has one right row and stays in memory,
+    // group "B" has three and spills; in "B" the match is followed by a non-matching row, so a
+    // match that was not copied out would be clobbered before it is emitted.
+    withSQLConf(
+      SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "1",
+      SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_SPILL_THRESHOLD.key -> "1") {
+      val leftSchema = StructType(
+        StructField("grp", StringType) ::
+          StructField("ts", IntegerType) :: Nil)
+      val rightSchema = StructType(
+        StructField("grp", StringType) ::
+          StructField("ts", IntegerType) ::
+          StructField("right_val", StringType) :: Nil)
+      // Values of differing lengths so a clobbered match stands out in the answer.
+      val bestVal = "b" * 40
+      val left = spark.createDataFrame(
+        List(Row("A", 8), Row("B", 8)).asJava, leftSchema)
+      val right = spark.createDataFrame(
+        List(Row("A", 3, "aa"), Row("B", 1, "x"), Row("B", 5, bestVal), Row("B", 12, "y")).asJava,
+        rightSchema)
+      checkAnswerAndSpill(
+        left.joinAsOf(
+          right, left.col("ts"), right.col("ts"), usingColumns = Seq("grp"),
+          joinType = "inner", tolerance = null,
+          allowExactMatches = true, direction = "backward"),
+        Seq(
+          Row("A", 8, "A", 3, "aa"),
+          Row("B", 8, "B", 5, bestVal)
+        )
+      )
+    }
+  }
 }