[SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence joins by xumingming · Pull Request #57162 · apache/spark

xumingming · 2026-07-09T15:41:23Z

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.

xumingming · 2026-07-16T12:36:18Z

@cloud-fan Can you take a look at?

cloud-fan

1 blocking, 1 non-blocking, 2 nits.
The hoisting is semantically sound and the interpreted paths look correct, but the codegen early-return; in HashJoin's doConsume can escape the producer's row loop and produce wrong results / a hang on batching producers.

Correctness (1)

HashJoin.scala:561: streamed-only guard emits a bare return; inside doConsume, which can escape the producer's row loop when the join's consume code isn't wrapped in a function — see inline

Design / architecture (1)

HashJoin.scala:168: the condition-split block is duplicated verbatim in HashJoin and SortMergeJoinExec — see inline

Nits: 2 minor items (see inline comments).

Verification

Traced the split's semantics on the interpreted paths: a streamed-only conjunct that is FALSE or NULL makes the full S AND other not-TRUE for every buffered row, so emitting the streamed row directly (null-padded for outer/existence, kept for anti, exists=false for existence) matches the pre-split result under three-valued logic. Binding order streamedPlan.output ++ buildPlan.output matches the joinedRow layout for all reachable cases. The SMJ codegen path is safe because it owns its while loop and uses continue. The one gap is the HashJoin codegen return; (below): HashJoin is a codegen consumer, so its doConsume is inlined into the streamed producer's loop unless WholeStageCodegenExec wraps it — which requireAllOutput (output ⊆ parent.usedInputs) does not guarantee once a streamed column is not referenced by the keys/condition.

cloud-fan · 2026-07-16T13:58:02Z

+      s"""
+         |${ev.code}
+         |if (${ev.isNull} || !${ev.value}) {
+         |  UnsafeRow $matched = null;


This guard (and the equivalent ones in codegenAnti and codegenExistence) ends with a bare return;. But HashJoin is a codegen consumer — doProduce just delegates to streamedPlan.produce(ctx, this) — so this doConsume code is inlined into the streamed producer's row loop unless WholeStageCodegenExec.consume wraps it in a separate doConsume_N function. That wrapping requires requireAllOutput = output.forall(parent.usedInputs.contains), and usedInputs defaults to references (join keys + condition attrs), so any streamed column not referenced by the keys/condition makes it false — e.g. the SELECT t1.* ... ON t1.id = t2.id AND udf(t1.a) = ... shape from the PR description. When not wrapped, the return; exits the producer's processNext(); for a batching producer (ColumnarToRowExec vectorized scan, RangeExec) the loop cursor is only advanced after the inner loop, so the next processNext() reprocesses the same batch → duplicate rows and/or non-termination.

MergeRowsExec.doConsume handles exactly this by always wrapping its body in a function ("so that return statements exit this function instead of the outer produce loop"). Suggest doing the same here, or restructuring the guard to fall through (e.g. do { } while(false) + a flag) instead of return. The current tests don't catch it: they use parallelize (an RDD → non-batching InputAdapter) and conditions where every streamed column is referenced, so the wrapper is always inserted.

cloud-fan · 2026-07-16T13:58:02Z

+          case _ => false
+        })) {
+      val conjuncts = splitConjunctivePredicates(condition.get)
+      val (streamedOnly, rest) = conjuncts.partition(_.references.subsetOf(streamedPlan.outputSet))


This whole streamedOnlyCondition/restCondition split — the splitConjunctivePredicates + partition(subsetOf(streamedPlan.outputSet)) and the joinType match { LeftAnti | LeftOuter | RightOuter | _: ExistenceJoin } gate — is duplicated verbatim in SortMergeJoinExec, and both classes newly mix in PredicateHelper for it. A later change to the supported-join-type set (or the split rule) has to touch both copies in lockstep; missing one silently mis-applies the optimization for one join family. Consider hoisting this into a shared trait or helper.

cloud-fan · 2026-07-16T13:58:02Z

+  }
+
+  // Condition: a = c (equi-key) AND d < 4.0 (right-only)
+  private lazy val rightOnlyResidualCondition = {


rightOnlyResidualCondition is defined but never referenced (only leftOnlyResidualCondition is used). Either wire it into a test — a right-only-residual case would add useful coverage — or remove it.

cloud-fan

3 addressed, 0 remaining, 3 new. (2 newly introduced, 1 late catches.)
1 blocking, 0 non-blocking, 2 nits.

Correctness (1)

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/StreamedSideJoinCondition.scala:52: Hoisting throwable streamed-side predicates changes unmatched-row behavior. -- see inline

Nits: 2 minor items (see inline comments).

Verification

Traced the config-on/off condition flow through HashJoin and SortMergeJoinExec, including unique/non-unique hash relations, outer/anti/existence outcomes, and the batching-producer codegen path. Compared the new physical split with PushPredicateThroughJoin's established relocation safety gate and re-adjudicated all 49 contract claims and 185 text candidates.

cloud-fan · 2026-07-20T19:12:12Z

+    }
+    if (condition.isDefined && splitEnabled && supported) {
+      val conjuncts = splitConjunctivePredicates(condition.get)
+      val (streamedOnly, rest) = conjuncts.partition(_.references.subsetOf(streamedPlan.outputSet))


Please restrict the hoisted set to non-throwable conjuncts. With this reference-only partition, raise_error(left_col) runs before probing even when the key has no buffered match; previously that residual was never evaluated for the row, so enabling the config changes an outer/anti result into an exception. PushPredicateThroughJoin handles the analogous relocation with cond.deterministic && !cond.throwable.

Made the changes, created a testing udf: TestThrowableUDF to test the throwable behavior.

BTW: the throwable property seems not widely adopted, only one usage in the Sequence expression.

cloud-fan · 2026-07-20T19:12:12Z

+      Row(null, 5.0, null, null),
+      Row(6, null, null, null)))
+
+  // ExistenceJoin with the same left-only residual condition: exists=true only for (2,1.0).


This case passes mixedResidualCondition, which also includes d < 4, so the comment should call it the mixed residual condition.

cloud-fan · 2026-07-20T19:12:12Z

+ *  - a broadcast hash join, so the scan and the join share one whole-stage. Sort-merge and
+ *    shuffled hash joins are out of scope here: their streamed side crosses a Sort or an
+ *    exchange, which advances its cursor before running the inlined consume code, and the
+ *    sort-merge guard is folded into the match condition rather than emitted as an early


The sort-merge guard is a standalone pre-loop guard that emits and continues before the match loop; it is not folded into the match condition. The relevant contrast is that it continues its own loop instead of returning from an inlined consumer.

cloud-fan

2 addressed, 1 remaining, 0 new.
1 blocking, 0 non-blocking, 0 nits.

Remaining from prior review (1)

The !p.throwable gate does not make hoisting safe for the motivating UDF case. ScalaUDF does not override throwable, so it inherits children.exists(_.throwable) and an ordinary deterministic UDF whose function throws is classified as non-throwable. For an unmatched streamed key, this change evaluates that UDF before probing even though the old join never evaluated the residual, turning a valid outer/anti result into an exception. Please restrict hoisting to expression families with a reliable non-throwing contract (or first make UDF throwable metadata conservative) and add an end-to-end test using a real registered UDF that throws on an unmatched streamed row. -- existing thread

Verification

Traced config-on/off behavior through HashJoin and SortMergeJoin interpreted and code-generated paths, including null keys, unique/non-unique build relations, failed hoisted predicates, residual-free probes, and outer/anti/existence output. Verified the current exception-safety gate against Expression.throwable and ScalaUDF: ScalaUDF overrides determinism but not throwable, so a deterministic registered UDF with literal/attribute children is treated as non-throwable even when its function body throws.

…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.

Address review comments on the whole-stage codegen shape of the hoisted streamed-side condition: - Extract the split logic shared by HashJoin and SortMergeJoinExec into StreamedSideJoinCondition.split (new file). - Replace the early emit + return guards in HashJoin's codegenOuter, codegenAnti and codegenExistence with the hoisted predicate folded into the probe-skip condition. The guard's bare return exited the streamed producer's processNext(); for batching producers (ColumnarToRowExec, RangeExec) the loop cursor is only written back after the inlined consume code, so the same batch was reprocessed, producing duplicate rows or non-termination. Folding also keeps a single consume site per streamed row, so lazy streamed variables are materialized in the same scope as all their uses. - Add SplitStreamedSideJoinConditionSuite covering left outer (unique and non-unique build key), left anti, existence and right outer joins over a vectorized parquet scan, using the same query with hoisting disabled as the result oracle. - Extend ExistenceJoinSuite with right-only residual condition and inner-join no-op coverage; rename leftOnlyResidualCondition to mixedResidualCondition.

- Change spark.sql.join.splitStreamedSideJoinCondition's binding policy to NOT_APPLICABLE: it is a physical-execution flag that does not affect view/UDF resolution. - Remove the dead streamedPlan/bufferedPlan vals from SortMergeJoinEvaluatorFactory and the now-unused LeftExistence import. - Reorder the interpreted anti join filter so a failed hoisted streamed-only predicate skips the hash probe, making the "(no probe needed)" comment accurate. - Add fully-hoisted (restCondition = None) test cases to ExistenceJoinSuite, covering SMJ codegen's conditionForCodegen = None branch and the interpreted always-true rest walk.

Restrict streamed-side predicate hoisting to deterministic, non-throwable conjuncts. The hoisted part runs for every streamed row, including rows with no buffered match that would never evaluate the conjunct otherwise, so hoisting a throwable conjunct could turn a valid outer/anti result into an exception, and hoisting a non-deterministic conjunct would change how many times it is evaluated per streamed row. Tests: add TestThrowableUDF (a test expression that throws when evaluated to false/null), unit tests for the split() exclusions, an end-to-end test proving the throwable conjunct is not evaluated for unmatched rows, and convert the suite queries from SQL strings to DataFrames so the throwable conjunct can be planted in the join condition.

Column.expr is no longer a member of the public Column interface since the Spark 4 Column rework; it is provided by RichColumn via the toRichColumn implicit. Import testImplicits.toRichColumn in SplitStreamedSideJoinConditionSuite, matching InnerJoinSuite and ExistenceJoinSuite.

The !throwable gate did not make streamed-side conjunct hoisting safe: throwable is opt-in metadata that most expressions do not override, so a deterministic ScalaUDF whose function throws inherits non-throwable from its children and would be hoisted above the probe, turning a valid outer/anti result into an exception for unmatched streamed rows. Add ExprUtils.canEvaluateUnconditionally, a shared catalyst helper that accepts only deterministic, subquery-free expressions from a whitelist of total families (attribute/literal leaves, GetStructField/ GetArrayStructFields/GetMapValue accessors, and boolean logic/ comparisons), and use it in StreamedSideJoinCondition.split instead of the !throwable check. Tests: replace the test-only TestThrowableUDF with an end-to-end test using a real registered UDF that throws on an unmatched streamed row (verified to throw under the old gate), cover the unmarked-throwable arithmetic case in the split unit test, and add ExprUtilsSuite for the whitelist itself.

cloud-fan

1 addressed, 0 remaining, 0 new.
0 blocking, 0 non-blocking, 0 nits.
The current revision resolves the prior exception-safety concern and preserves join semantics across the affected execution paths.

Verification

I traced the split from the SQLConf gate through both physical join families. A false or null hoisted conjunct makes the full conjunctive residual not true for every buffered row, so skipping the probe reaches the same null-padded, anti-preserved, or exists=false outcome as exhausting the old match loop. I also verified that the whitelist rejects the motivating throwing-UDF shape, HashJoin no longer exits an inlined producer loop early, and SortMergeJoin's guard continues only its own streamed-row loop.

…ence joins ### 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. Closes #57162 from xumingming/hoist-streamed-side-join-predicates. Authored-by: James Xu <xumingmingv@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit 966cdda) Signed-off-by: Wenchen Fan <wenchen@databricks.com>

cloud-fan · 2026-08-03T07:02:30Z

Merge Summary:

merged into master 966cdda
merged into branch-4.x 4fdaf14

Posted by merge_spark_pr.py

cloud-fan reviewed Jul 16, 2026

View reviewed changes

xumingming force-pushed the hoist-streamed-side-join-predicates branch 2 times, most recently from 19de9f6 to fb956a0 Compare July 20, 2026 15:10

cloud-fan reviewed Jul 20, 2026

View reviewed changes

cloud-fan reviewed Jul 24, 2026

View reviewed changes

xumingming added 7 commits August 3, 2026 07:56

xumingming force-pushed the hoist-streamed-side-join-predicates branch from dc98d9e to 9e95594 Compare August 2, 2026 23:56

cloud-fan approved these changes Aug 3, 2026

View reviewed changes

cloud-fan closed this in 966cdda Aug 3, 2026

Navigation Menu

Uh oh!

[SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence joins - #57162

[SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence joins#57162
xumingming wants to merge 7 commits into
apache:masterfrom
xumingming:hoist-streamed-side-join-predicates

xumingming commented Jul 9, 2026

Uh oh!

xumingming commented Jul 16, 2026

Uh oh!

cloud-fan left a comment

Uh oh!

cloud-fan Jul 16, 2026

Uh oh!

cloud-fan Jul 16, 2026

Uh oh!

cloud-fan Jul 16, 2026

Uh oh!

cloud-fan left a comment

Uh oh!

cloud-fan Jul 20, 2026

Uh oh!

xumingming Jul 21, 2026

Uh oh!

cloud-fan Jul 20, 2026

Uh oh!

cloud-fan Jul 20, 2026

Uh oh!

cloud-fan left a comment

Uh oh!

cloud-fan left a comment

Uh oh!

cloud-fan commented Aug 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

xumingming commented Jul 9, 2026

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

xumingming commented Jul 16, 2026

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Correctness (1)

Design / architecture (1)

Verification

Uh oh!

cloud-fan Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Correctness (1)

Verification

Uh oh!

cloud-fan Jul 20, 2026

Choose a reason for hiding this comment

Uh oh!

xumingming Jul 21, 2026

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jul 20, 2026

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jul 20, 2026

Choose a reason for hiding this comment

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Remaining from prior review (1)

Verification

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Verification

Uh oh!

cloud-fan commented Aug 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants