xumingming opened a new pull request, #57162:
URL: https://github.com/apache/spark/pull/57162
### 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.
```
SELECT /*+ BROADCAST(t2) */ t1.*
FROM t1 LEFT JOIN t2
ON t1.id = t2.id AND expensive_udf(t1.a) = 'ok'
```
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.
```
SELECT *
FROM t1 LEFT ANTI JOIN t2
ON t1.id = t2.id AND t1.category IN (1, 3, 5)
```
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.
--
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]
