[SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence joins#57162
Open
xumingming wants to merge 2 commits into
Open
[SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence joins#57162xumingming wants to merge 2 commits into
xumingming wants to merge 2 commits into
Conversation
…ence joins For join types that preserve streamed rows (LeftAnti, LeftOuter, RightOuter and ExistenceJoin), ON-clause conjuncts that reference only the streamed side were previously evaluated once per matched (streamed, buffered) pair inside the hash-bucket or merge walk. When such predicates contain expensive UDFs and are false for most rows, this wastes significant CPU. The optimization is based on the observation that, for these join types, if a streamed-only predicate S is FALSE/NULL for a streamed row, the full join condition S AND other(S,B) is FALSE/NULL for ANY buffered row B. Therefore the streamed row outcome is already determined before any probe: it is emitted for outer/existence joins and emitted for left anti joins when no match exists. This change splits the join condition into streamed-only and mixed-side parts. The streamed-only part is evaluated once per streamed row before entering the match loop; only the mixed-side remainder is evaluated inside the loop. When the residual condition is entirely streamed-side-only, the inner loop degenerates to a pure existence check. Guarded by the new SQL config spark.sql.join.splitStreamedSideJoinCondition (default false).
When spark.sql.join.splitStreamedSideJoinCondition is enabled, the non-codegen SortMergeJoin OneSideOuterIterator could emit extra rows for streamed rows whose streamed-only predicate is false. advanceStream() correctly emitted a null-padded row and reset the buffered-match iterator, but advanceNext() unconditionally re-created the iterator and checked only the residual condition, ignoring the false streamed-only predicate. Fix advanceNext() to only walk buffered matches when the iterator is already initialized for the current streamed row.
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.
What changes were proposed in this pull request?
For join types that preserve streamed rows (LeftAnti, LeftOuter, RightOuter and ExistenceJoin), ON-clause conjuncts that reference only the streamed side were previously evaluated once per matched (streamed, buffered) pair inside the hash-bucket or merge walk. When such predicates contain expensive UDFs and are false for most rows, this wastes significant CPU.
The optimization is based on the observation that, for these join types, if a streamed-only predicate S is FALSE/NULL for a streamed row, the full join condition S AND other(S,B) is FALSE/NULL for ANY buffered row B. Therefore the streamed row outcome is already determined before any probe: it is emitted for outer/existence joins and emitted for left anti joins when no match exists.
This change splits the join condition into streamed-only and mixed-side parts. The streamed-only part is evaluated once per streamed row before entering the match loop; only the mixed-side remainder is evaluated inside the loop. When the residual condition is entirely streamed-side-only, the inner loop degenerates to a pure existence check.
Guarded by the new SQL config
spark.sql.join.splitStreamedSideJoinCondition (default false).
Why are the changes needed?
For hash and sort-merge joins that preserve streamed-side rows (LeftAnti, LeftOuter, RightOuter and ExistenceJoin), ON-clause conjuncts that reference only the streamed side are currently evaluated repeatedly inside the match loop, once per candidate buffered row. When such predicates involve expensive UDFs or computations and are false for most streamed rows, this wastes CPU because the join result for those streamed rows is already determined before any buffered row is inspected.
Scenario 1: Expensive streamed-side filter in a left outer join.
If expensive_udf(t1.a) returns a value other than 'ok' for most rows of t1, the join condition can never be satisfied for those rows, yet the UDF is invoked for every matching row in t2.
Scenario 2: Left anti join with a streamed-side predicate.
Rows from t1 whose category is not in the allowed set are guaranteed to be emitted (they have no match by definition), but the current implementation still probes t2 for each of them.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
Unit test.
Was this patch authored or co-authored using generative AI tooling?
No.