James Xu created SPARK-58066:
--------------------------------
Summary: [SQL] Hoist streamed-side residual predicates for left
outer, left anti, right outer and existence joins
Key: SPARK-58066
URL: https://issues.apache.org/jira/browse/SPARK-58066
Project: Spark
Issue Type: Improvement
Components: Optimizer
Affects Versions: 4.3.0
Reporter: James Xu
h3. Problem:
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.*
{code:java}
SELECT /*+ BROADCAST(t2) */ t1.*
FROM t1 LEFT JOIN t2
ON t1.id = t2.id AND expensive_udf(t1.a) = 'ok'{code}
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.*
{code:java}
SELECT *
FROM t1 LEFT ANTI JOIN t2
ON t1.id = t2.id AND t1.category IN (1, 3, 5){code}
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.
h3. Root Cause:
In HashJoin and SortMergeJoinExec, the join condition is evaluated as a single
predicate over the joined row. Conjuncts that only reference the streamed side
are not separated from conjuncts that reference both sides, so streamed-side
evaluation is not hoisted out of the inner matching loop.
h3. Solution:
Split the join condition into streamed-only predicates and the remaining
mixed-side predicates. Evaluate the streamed-only part once per streamed row
before entering the match loop; use only the mixed-side part inside the loop.
If the residual condition is entirely streamed-side-only, the inner loop
degenerates to a pure existence check.
The algebraic basis is that for join types preserving streamed rows, if the
streamed-only conjunct is FALSE/NULL, the full condition is FALSE/NULL for any
buffered row, so the streamed row is emitted (or suppressed for anti joins)
without probing.
The change is guarded by a new SQL configuration
spark.sql.join.splitStreamedSideJoinCondition (default false) to allow safe
rollout.
h3. Expected Impact:
For workloads where streamed-side predicates filter out the majority of
streamed rows, this avoids redundant UDF invocations and redundant probe
operations. In synthetic benchmarks with expensive streamed-side predicates and
low match rates, wall-clock time is projected to drop significantly, while CPU
utilization for the predicate function decreases by up to the average number of
buffered matches per streamed row.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
