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]
