[SPARK-56887][SQL][FOLLOWUP] Add spill support and extend ShuffledJoin in SortMergeAsOfJoinExec#56595
Open
sarutak wants to merge 2 commits into
Open
[SPARK-56887][SQL][FOLLOWUP] Add spill support and extend ShuffledJoin in SortMergeAsOfJoinExec#56595sarutak wants to merge 2 commits into
sarutak wants to merge 2 commits into
Conversation
…orward scan unification
Refactor SortMergeAsOfJoinExec per review feedback:
1. Extend ShuffledJoin instead of BaseJoinExec
- Inherits standard requiredChildDistribution, outputPartitioning,
and output for all join types
- Override requiredChildDistribution only for the AllTuples case
(no equi-keys)
- supportCodegen = false (codegen is future work)
2. Replace ArrayBuffer with ExternalAppendOnlyUnsafeRowArray
- 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 scan to forward-only (remove findBestRightToLeft)
- Backward joins: forward scan keeping the last as-of-satisfying
row (monotone condition guarantees correctness + early termination
on true->false transition)
- Forward/Nearest joins: forward scan with distance-based early
termination (unchanged logic)
- Both are compatible with ExternalAppendOnlyUnsafeRowArray's
forward-only generateIterator()
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Follow-up to #55912. Refactors
SortMergeAsOfJoinExecto address review feedback:Extend
ShuffledJoininstead ofBaseJoinExec: Inherits standardrequiredChildDistribution,outputPartitioning, andoutputhandling consistent with other shuffled join operators.Replace
ArrayBufferwithExternalAppendOnlyUnsafeRowArray: 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 samespark.sql.sortMergeJoinExecBuffer.*configs asSortMergeJoinExec.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).Add
spillSizemetric: Reports spill bytes at task completion, matchingSortMergeJoinExec's observability.Why are the changes needed?
The original operator used an in-memory
ArrayBufferthat could OOM on skewed equi-key groups (flagged in #55912 review). The reverse scan for Backward joins prevented usingExternalAppendOnlyUnsafeRowArray(forward-only iterator). As noted in the review, the reverse scan is unnecessary since a forward scan keeping the last match produces the same result for monotone predicates, and extendingShuffledJoinprovides maintenance parity withSortMergeJoinExec.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 passDataFrameAsOfJoinSuite: all 11 tests passAsOfJoinBenchmark: no performance regression (AMD EPYC 7763, 10K×10K rows, 100 groups):Was this patch authored or co-authored using generative AI tooling?
Kiro CLI / Claude