sarutak opened a new pull request, #56595:
URL: https://github.com/apache/spark/pull/56595

   Follow-up to #55912. Refactors `SortMergeAsOfJoinExec` to address review 
feedback:
   
   1. **Extend `ShuffledJoin`** instead of `BaseJoinExec`: Inherits standard 
`requiredChildDistribution`, `outputPartitioning`, and `output` handling 
consistent with other shuffled join operators.
   
   2. **Replace `ArrayBuffer` with `ExternalAppendOnlyUnsafeRowArray`**: The 
right-side group buffer now spills to disk when the in-memory threshold is 
exceeded, eliminating OOM for skewed equi-key groups. Uses the same 
`spark.sql.sortMergeJoinExecBuffer.*` configs as `SortMergeJoinExec`.
   
   3. **Unify to forward-only scan**: Removes `findBestRightToLeft` (which 
required indexed random access incompatible with spill-backed storage). 
Backward joins now use a forward scan keeping the last as-of-satisfying row, 
which produces the same result because the as-of predicate is monotone over the 
sorted buffer. Early termination is preserved (true→false transition for 
Backward, distance increase for Forward/Nearest).
   
   4. **Add `spillSize` metric**: Reports spill bytes at task completion, 
matching `SortMergeJoinExec`'s observability.
   
   ### Why are the changes needed?
   The original operator used an in-memory `ArrayBuffer` that could OOM on 
skewed equi-key groups. The reverse scan for Backward joins prevented using 
`ExternalAppendOnlyUnsafeRowArray` (forward-only iterator). The reverse scan is 
unnecessary since a forward scan keeping the last match produces the same 
result for monotone predicates.
   
   ### Does this PR introduce *any* user-facing change?
   
   No. Behavior is unchanged. The operator now spills to disk instead of OOMing 
for large groups.
   
   ### How was this patch tested?
   
   - `SortMergeAsOfJoinSuite`: all 26 tests pass
   - `DataFrameAsOfJoinSuite`: all 11 tests pass
   - `AsOfJoinBenchmark`: no performance regression (AMD EPYC 7763, 10K×10K 
rows, 100 groups):
   
   | JDK | Before (ms) | After (ms) | Speedup vs baseline |
   |-----|:-----------:|:----------:|:-------------------:|
   | 17  | 59          | 53         | 720× |
   | 21  | 62          | 68         | 562× |
   | 25  | 56          | 61         | 660× |
   
   ### Was this patch authored or co-authored using generative AI tooling?
   Kiro CLI / Claude


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

Reply via email to