[jira] [Assigned] (SPARK-58066) [SQL] Hoist streamed-side residual predicates for left outer, left anti, right outer and existence joins

["Wenchen Fan (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-58066?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Wenchen Fan reassigned SPARK-58066:
-----------------------------------

    Assignee: James Xu

>  [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
>            Assignee: 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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org