[
https://issues.apache.org/jira/browse/SPARK-58066?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58066:
-----------------------------------
Labels: pull-request-available (was: )
> [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
> Priority: Major
> Labels: pull-request-available
>
> 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]
