You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/03 07:02:08 UTC
(spark) branch master updated: [SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence joins
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 966cddae0cac [SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence joins
966cddae0cac is described below
commit 966cddae0cac828f7912021c828d72c0cd108ea5
Author: James Xu <xu...@gmail.com>
AuthorDate: Mon Aug 3 15:01:51 2026 +0800
[SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence 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 <xu...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/expressions/ExprUtils.scala | 38 ++
.../org/apache/spark/sql/internal/SQLConf.scala | 14 +
.../sql/catalyst/expressions/ExprUtilsSuite.scala | 58 +++
.../spark/sql/execution/joins/HashJoin.scala | 234 ++++++++++---
.../sql/execution/joins/JoinCodegenSupport.scala | 43 ++-
.../joins/SortMergeJoinEvaluatorFactory.scala | 50 ++-
.../sql/execution/joins/SortMergeJoinExec.scala | 119 ++++++-
.../joins/StreamedSideJoinCondition.scala | 68 ++++
.../sql/execution/joins/ExistenceJoinSuite.scala | 388 ++++++++++++++++++---
.../SplitStreamedSideJoinConditionSuite.scala | 245 +++++++++++++
10 files changed, 1119 insertions(+), 138 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
index 6c9ea6d656b2..f3f2988abddc 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch, TypeCheckSuccess}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.plans.logical.Aggregate
+import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, CharVarcharUtils}
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase, QueryExecutionErrors}
import org.apache.spark.sql.internal.types.{AbstractMapType, StringTypeWithCollation}
@@ -220,4 +221,41 @@ object ExprUtils extends EvalHelper with QueryErrorsBase {
a.groupingExpressions.foreach(checkValidGroupingExprs)
a.aggregateExpressions.foreach(checkValidAggregateExpression)
}
+
+ /**
+ * Returns true if `e` is safe to evaluate unconditionally, i.e. on rows where the
+ * original plan would not have evaluated it: evaluating it must not raise an error and
+ * must not change the query result. This is the check to use when relocating an
+ * expression out of a short-circuited position, e.g. moving the base out of the taken
+ * branch of an If/CaseWhen, or hoisting a streamed-side join conjunct above the probe
+ * so it also runs for streamed rows that have no match.
+ *
+ * Only a whitelist of total, deterministic expressions qualifies:
+ * - leaves: attribute references and literals;
+ * - total accessors: GetStructField, GetArrayStructFields and GetMapValue never throw.
+ * Note that GetArrayItem/ElementAt are NOT included: they throw on invalid ordinals
+ * when ANSI mode is on;
+ * - logic/predicates: And, Or, Not, comparisons, IsNull, IsNotNull, IsNaN, NullIf,
+ * Coalesce, In and InSet are total boolean functions.
+ * Anything else (arithmetic, casts, string functions, UDFs, nested IF/CASE WHEN, ...)
+ * conservatively returns false. On top of the whitelist, the expression must be
+ * deterministic and must not contain subqueries.
+ *
+ * Note: this deliberately does not rely on [[Expression.throwable]], which is opt-in
+ * metadata that most expressions do not override. A throwing ScalaUDF with
+ * non-throwing children, for example, reports non-throwable.
+ */
+ def canEvaluateUnconditionally(e: Expression): Boolean =
+ e.deterministic && !e.containsPattern(PLAN_EXPRESSION) &&
+ canEvaluateUnconditionallyInternal(e)
+
+ private def canEvaluateUnconditionallyInternal(e: Expression): Boolean = e match {
+ case _: AttributeReference | _: Literal => true
+ case _: GetStructField | _: GetArrayStructFields | _: GetMapValue =>
+ e.children.forall(canEvaluateUnconditionallyInternal)
+ case _: And | _: Or | _: Not | _: BinaryComparison | _: IsNull | _: IsNotNull |
+ _: IsNaN | _: NullIf | _: Coalesce | _: In | _: InSet =>
+ e.children.forall(canEvaluateUnconditionallyInternal)
+ case _ => false
+ }
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 217f6f68a7b5..11379a31604f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -980,6 +980,18 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val SPLIT_STREAMED_SIDE_JOIN_CONDITION =
+ buildConf("spark.sql.join.splitStreamedSideJoinCondition")
+ .internal()
+ .doc("When true, split join conditions for LeftAnti, LeftOuter, RightOuter, and " +
+ "ExistenceJoin by referenced side, evaluating streamed-side-only conjuncts before " +
+ "the hash-bucket or merge walk. This avoids evaluating expensive streamed-side-only " +
+ "predicates once per matched buffered row.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
val SORT_MERGE_AS_OF_JOIN_ENABLED =
buildConf("spark.sql.join.sortMergeAsOfJoin.enabled")
.doc("When true, use a dedicated sort-merge physical operator for AS-OF joins " +
@@ -8757,6 +8769,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def preferSortMergeJoin: Boolean = getConf(PREFER_SORTMERGEJOIN)
+ def splitStreamedSideJoinCondition: Boolean = getConf(SPLIT_STREAMED_SIDE_JOIN_CONDITION)
+
def sortMergeAsOfJoinEnabled: Boolean = getConf(SORT_MERGE_AS_OF_JOIN_ENABLED)
def sqlAsOfJoinEnabled: Boolean = getConf(SQL_ASOF_JOIN_ENABLED)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExprUtilsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExprUtilsSuite.scala
new file mode 100644
index 000000000000..c0ac407b0432
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExprUtilsSuite.scala
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.types.{ArrayType, IntegerType, LongType, StringType, StructField, StructType}
+
+class ExprUtilsSuite extends SparkFunSuite {
+
+ private val a = AttributeReference("a", LongType)()
+
+ test("canEvaluateUnconditionally: whitelisted total expressions") {
+ assert(ExprUtils.canEvaluateUnconditionally(a))
+ assert(ExprUtils.canEvaluateUnconditionally(Literal(1L)))
+ assert(ExprUtils.canEvaluateUnconditionally(
+ And(LessThan(a, Literal(5L)), IsNotNull(a))))
+ assert(ExprUtils.canEvaluateUnconditionally(
+ GetStructField(
+ AttributeReference("s", StructType(StructField("f1", IntegerType) :: Nil))(), 0)))
+ assert(ExprUtils.canEvaluateUnconditionally(Coalesce(Seq(a, Literal(0L)))))
+ assert(ExprUtils.canEvaluateUnconditionally(In(a, Seq(Literal(1L), Literal(2L)))))
+ }
+
+ test("canEvaluateUnconditionally: expressions that can throw are excluded") {
+ // Arithmetic can throw (overflow, div-by-zero in ANSI mode) even though it does not
+ // override `throwable` and inherits false from its non-throwing children, which is
+ // why the whitelist is used instead of the throwable flag.
+ val arithmetic = LessThan(Add(a, Literal(1L)), Literal(5L))
+ assert(!arithmetic.throwable)
+ assert(!ExprUtils.canEvaluateUnconditionally(arithmetic))
+ assert(!ExprUtils.canEvaluateUnconditionally(EqualTo(Remainder(a, Literal(3L)), Literal(0L))))
+ assert(!ExprUtils.canEvaluateUnconditionally(Cast(a, StringType)))
+ // GetArrayItem/ElementAt can throw on invalid ordinals in ANSI mode.
+ val arr = AttributeReference("arr", ArrayType(IntegerType))()
+ assert(!ExprUtils.canEvaluateUnconditionally(GetArrayItem(arr, Literal(0))))
+ // A whitelisted predicate over a non-whitelisted child is still excluded.
+ assert(!ExprUtils.canEvaluateUnconditionally(IsNotNull(Add(a, Literal(1L)))))
+ }
+
+ test("canEvaluateUnconditionally: non-deterministic expressions are excluded") {
+ assert(!ExprUtils.canEvaluateUnconditionally(LessThan(Rand(Literal(0L)), Literal(0.5))))
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala
index 9df791aa8de0..901ae14bcdcc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala
@@ -152,6 +152,25 @@ trait HashJoin extends JoinCodegenSupport {
(r: InternalRow) => true
}
+ /**
+ * For join types that preserve all streamed rows, split the condition into
+ * streamed-only and rest. The streamed-only part can be evaluated once per streamed row
+ * before probing the hash table.
+ */
+ protected lazy val (streamedOnlyCondition, restCondition):
+ (Option[Expression], Option[Expression]) = {
+ StreamedSideJoinCondition.split(
+ condition, joinType, streamedPlan, conf.splitStreamedSideJoinCondition)
+ }
+
+ @transient protected[this] lazy val boundStreamedOnlyCondition = streamedOnlyCondition.map {
+ Predicate.create(_, streamedPlan.output).eval _
+ }.getOrElse((_: InternalRow) => true)
+
+ @transient protected[this] lazy val boundRestCondition = restCondition.map {
+ Predicate.create(_, streamedPlan.output ++ buildPlan.output).eval _
+ }.getOrElse((_: InternalRow) => true)
+
protected def createResultProjection(): (InternalRow) => InternalRow = joinType match {
case LeftExistence(_) =>
UnsafeProjection.create(output, output)
@@ -205,8 +224,14 @@ trait HashJoin extends JoinCodegenSupport {
streamedIter.map { currentRow =>
val rowKey = keyGenerator(currentRow)
joinedRow.withLeft(currentRow)
- val matched = hashedRelation.getValue(rowKey)
- if (matched != null && boundCondition(joinedRow.withRight(matched))) {
+ // If streamed-only condition is false/null, the full condition can never be true,
+ // so the row is emitted with null build side (no probe needed).
+ val matched = if (boundStreamedOnlyCondition(currentRow)) {
+ hashedRelation.getValue(rowKey)
+ } else {
+ null
+ }
+ if (matched != null && boundRestCondition(joinedRow.withRight(matched))) {
joinedRow
} else {
joinedRow.withRight(nullRow)
@@ -216,13 +241,17 @@ trait HashJoin extends JoinCodegenSupport {
streamedIter.flatMap { currentRow =>
val rowKey = keyGenerator(currentRow)
joinedRow.withLeft(currentRow)
- val buildIter = hashedRelation.get(rowKey)
+ val buildIter = if (boundStreamedOnlyCondition(currentRow)) {
+ hashedRelation.get(rowKey)
+ } else {
+ null
+ }
new RowIterator {
private var found = false
override def advanceNext(): Boolean = {
while (buildIter != null && buildIter.hasNext) {
val nextBuildRow = buildIter.next()
- if (boundCondition(joinedRow.withRight(nextBuildRow))) {
+ if (boundRestCondition(joinedRow.withRight(nextBuildRow))) {
if (found && singleJoin) {
throw QueryExecutionErrors.scalarSubqueryReturnsMultipleRows();
}
@@ -279,19 +308,23 @@ trait HashJoin extends JoinCodegenSupport {
if (hashedRelation.keyIsUnique) {
streamIter.map { current =>
val key = joinKeys(current)
- lazy val matched = hashedRelation.getValue(key)
- val exists = !key.anyNull && matched != null &&
- (condition.isEmpty || boundCondition(joinedRow(current, matched)))
+ val exists = !key.anyNull && boundStreamedOnlyCondition(current) && {
+ val matched = hashedRelation.getValue(key)
+ matched != null && (restCondition.isEmpty ||
+ boundRestCondition(joinedRow(current, matched)))
+ }
result.setBoolean(0, exists)
joinedRow(current, result)
}
} else {
streamIter.map { current =>
val key = joinKeys(current)
- lazy val buildIter = hashedRelation.get(key)
- val exists = !key.anyNull && buildIter != null && (condition.isEmpty || buildIter.exists {
- (row: InternalRow) => boundCondition(joinedRow(current, row))
- })
+ val exists = !key.anyNull && boundStreamedOnlyCondition(current) && {
+ val buildIter = hashedRelation.get(key)
+ buildIter != null && (restCondition.isEmpty || buildIter.exists {
+ (row: InternalRow) => boundRestCondition(joinedRow(current, row))
+ })
+ }
result.setBoolean(0, exists)
joinedRow(current, result)
}
@@ -313,16 +346,21 @@ trait HashJoin extends JoinCodegenSupport {
streamIter.filter { current =>
val key = joinKeys(current)
lazy val matched = hashedRelation.getValue(key)
- key.anyNull || matched == null ||
- (condition.isDefined && !boundCondition(joinedRow(current, matched)))
+ // If streamed-only condition is false/null, the full condition can never be true,
+ // so the row is guaranteed emitted (no probe needed).
+ key.anyNull || !boundStreamedOnlyCondition(current) || matched == null ||
+ (restCondition.isDefined && !boundRestCondition(joinedRow(current, matched)))
}
} else {
streamIter.filter { current =>
val key = joinKeys(current)
lazy val buildIter = hashedRelation.get(key)
- key.anyNull || buildIter == null || (condition.isDefined && !buildIter.exists {
- row => boundCondition(joinedRow(current, row))
- })
+ // If streamed-only condition is false/null, the full condition can never be true,
+ // so the row is guaranteed emitted (no probe needed).
+ key.anyNull || !boundStreamedOnlyCondition(current) || buildIter == null ||
+ (restCondition.isDefined && !buildIter.exists {
+ row => boundRestCondition(joinedRow(current, row))
+ })
}
}
}
@@ -357,6 +395,46 @@ trait HashJoin extends JoinCodegenSupport {
}
}
+ /**
+ * Generates the code for evaluating a streamed-side-only condition on the given stream vars.
+ */
+ protected def genStreamedOnlyCondition(
+ ctx: CodegenContext,
+ expr: Expression,
+ streamVars: Seq[ExprCode]): ExprCode = {
+ ctx.currentVars = streamVars
+ val boundExpr = BindReferences.bindReference(expr, streamedPlan.output)
+ boundExpr.genCode(ctx)
+ }
+
+ /**
+ * Generates code evaluating the hoisted streamed-only condition, if any. Returns the
+ * generated code and the name of the boolean variable it declares, which holds whether the
+ * streamed row may satisfy the full join condition. Returns ("", "") when no streamed-only
+ * condition was hoisted.
+ *
+ * Callers must fold the boolean into their probe-skip condition instead of emitting a row
+ * and returning early: HashJoin's doConsume is inlined into the streamed producer's row
+ * loop, where a bare `return` would skip the producer's loop-cursor write-back (batching
+ * producers such as ColumnarToRowExec would reprocess the same batch). Folding also keeps
+ * a single consume site per streamed row, so lazy streamed variables are materialized in
+ * the same scope as all their uses.
+ */
+ protected def genStreamedOnlyCheck(
+ ctx: CodegenContext,
+ input: Seq[ExprCode]): (String, String) = {
+ streamedOnlyCondition match {
+ case Some(expr) =>
+ val ev = genStreamedOnlyCondition(ctx, expr, input)
+ val passed = ctx.freshName("streamedOnlyPassed")
+ (s"""
+ |${ev.code}
+ |boolean $passed = !${ev.isNull} && ${ev.value};
+ """.stripMargin, passed)
+ case None => ("", "")
+ }
+ }
+
override def doProduce(ctx: CodegenContext): String = {
streamedPlan.asInstanceOf[CodegenSupport].produce(ctx, this)
}
@@ -457,14 +535,11 @@ trait HashJoin extends JoinCodegenSupport {
val buildVars = genOneSideJoinVars(ctx, matched, buildPlan, setDefaultValue = true)
val numOutput = metricTerm(ctx, "numOutputRows")
- // filter the output via condition. When there is no condition, skip the `conditionPassed`
- // variable and the wrapping `if (!conditionPassed)` / `if (conditionPassed)` branches that
- // would always be dead / unconditional.
- val hasCondition = condition.isDefined
- val conditionPassed = if (hasCondition) ctx.freshName("conditionPassed") else ""
- val checkCondition = if (hasCondition) {
- val expr = condition.get
- // evaluate the variables from build side that used by condition
+ // Evaluate the rest of the condition (cross-side conjuncts) inside the match loop.
+ val hasRestCondition = restCondition.isDefined
+ val conditionPassed = if (hasRestCondition) ctx.freshName("conditionPassed") else ""
+ val checkCondition = if (hasRestCondition) {
+ val expr = restCondition.get
val eval = evaluateRequiredVariables(buildPlan.output, buildVars, expr.references)
ctx.currentVars = input ++ buildVars
val ev =
@@ -486,8 +561,19 @@ trait HashJoin extends JoinCodegenSupport {
case BuildRight => input ++ buildVars
}
+ // Fold the hoisted streamed-only predicate (if any) into the probe below: a streamed row
+ // that fails it cannot match, so the probe is skipped and the regular path emits the row
+ // with a null build side. See genStreamedOnlyCheck for why this must not be an early
+ // emit + return.
+ val (streamedOnlyCheckCode, streamedOnlyPassed) = genStreamedOnlyCheck(ctx, input)
+ val skipProbe = if (streamedOnlyPassed.nonEmpty) {
+ s"$anyNull || !$streamedOnlyPassed"
+ } else {
+ anyNull
+ }
+
if (keyIsUnique) {
- val resetWhenConditionFails = if (hasCondition) {
+ val resetWhenConditionFails = if (hasRestCondition) {
s"""
|if (!$conditionPassed) {
| $matched = null;
@@ -501,8 +587,9 @@ trait HashJoin extends JoinCodegenSupport {
s"""
|// generate join key for stream side
|${keyEv.code}
+ |$streamedOnlyCheckCode
|// find matches from HashedRelation
- |UnsafeRow $matched = $anyNull ? null: (UnsafeRow)$relationTerm.getValue(${keyEv.value});
+ |UnsafeRow $matched = $skipProbe ? null: (UnsafeRow)$relationTerm.getValue(${keyEv.value});
|${checkCondition.trim}
|$resetWhenConditionFails
|$numOutput.add(1);
@@ -525,13 +612,15 @@ trait HashJoin extends JoinCodegenSupport {
}
val (conditionGuardOpen, conditionGuardClose) =
- if (hasCondition) (s"if ($conditionPassed) {", "}") else ("", "")
+ if (hasRestCondition) (s"if ($conditionPassed) {", "}") else ("", "")
s"""
|// generate join key for stream side
|${keyEv.code}
+ |$streamedOnlyCheckCode
|// find matches from HashRelation
- |$iteratorCls $matches = $anyNull ? null : ($iteratorCls)$relationTerm.get(${keyEv.value});
+ |$iteratorCls $matches = $skipProbe ?
+ | null : ($iteratorCls)$relationTerm.get(${keyEv.value});
|boolean $found = false;
|// the last iteration of this loop is to emit an empty row if there is no matched rows.
|while ($matches != null && $matches.hasNext() || !$found) {
@@ -617,7 +706,23 @@ trait HashJoin extends JoinCodegenSupport {
}
val (keyEv, anyNull) = genStreamSideJoinKey(ctx, input)
- val (matched, checkCondition, _) = getJoinCondition(ctx, input, streamedPlan, buildPlan)
+ val (matched, checkCondition, _) = restCondition match {
+ case Some(expr) => getJoinCondition(ctx, expr, input, streamedPlan, buildPlan, None)
+ case None =>
+ val dummy = ctx.freshName("matched")
+ (dummy, "", Nil)
+ }
+
+ // Fold the hoisted streamed-only predicate (if any) into the probe below: a streamed row
+ // that fails it is guaranteed to be emitted, so the probe is skipped and the regular
+ // !found path emits the row. See genStreamedOnlyCheck for why this must not be an early
+ // emit + return.
+ val (streamedOnlyCheckCode, streamedOnlyPassed) = genStreamedOnlyCheck(ctx, input)
+ val probeAllowed = if (streamedOnlyPassed.nonEmpty) {
+ s"!($anyNull) && $streamedOnlyPassed"
+ } else {
+ s"!($anyNull)"
+ }
if (keyIsUnique) {
val found = ctx.freshName("found")
@@ -625,8 +730,9 @@ trait HashJoin extends JoinCodegenSupport {
|boolean $found = false;
|// generate join key for stream side
|${keyEv.code}
+ |$streamedOnlyCheckCode
|// Check if the key has nulls.
- |if (!($anyNull)) {
+ |if ($probeAllowed) {
| // Check if the HashedRelation exists.
| UnsafeRow $matched = (UnsafeRow)$relationTerm.getValue(${keyEv.value});
| if ($matched != null) {
@@ -649,8 +755,9 @@ trait HashJoin extends JoinCodegenSupport {
|boolean $found = false;
|// generate join key for stream side
|${keyEv.code}
+ |$streamedOnlyCheckCode
|// Check if the key has nulls.
- |if (!($anyNull)) {
+ |if ($probeAllowed) {
| // Check if the HashedRelation exists.
| $iteratorCls $matches = ($iteratorCls)$relationTerm.get(${keyEv.value});
| if ($matches != null) {
@@ -682,21 +789,32 @@ trait HashJoin extends JoinCodegenSupport {
val matched = ctx.freshName("matched")
val buildVars = genOneSideJoinVars(ctx, matched, buildPlan, setDefaultValue = false)
- val checkCondition = if (condition.isDefined) {
- val expr = condition.get
- // evaluate the variables from build side that used by condition
- val eval = evaluateRequiredVariables(buildPlan.output, buildVars, expr.references)
- // filter the output via condition
- ctx.currentVars = input ++ buildVars
- val ev =
- BindReferences.bindReference(expr, streamedPlan.output ++ buildPlan.output).genCode(ctx)
- s"""
- |$eval
- |${ev.code}
- |$existsVar = !${ev.isNull} && ${ev.value};
- """.stripMargin
+
+ // Evaluate the rest of the condition (cross-side conjuncts) inside the match loop.
+ val checkCondition = restCondition match {
+ case Some(expr) =>
+ val eval = evaluateRequiredVariables(buildPlan.output, buildVars, expr.references)
+ ctx.currentVars = input ++ buildVars
+ val ev = BindReferences.bindReference(
+ expr, streamedPlan.output ++ buildPlan.output).genCode(ctx)
+ s"""
+ |$eval
+ |${ev.code}
+ |$existsVar = !${ev.isNull} && ${ev.value};
+ """.stripMargin
+ case None =>
+ s"$existsVar = true;"
+ }
+
+ // Fold the hoisted streamed-only predicate (if any) into the probe below: a streamed row
+ // that fails it yields exists = false, so the probe is skipped and the single emit below
+ // produces the row. See genStreamedOnlyCheck for why this must not be an early emit +
+ // return.
+ val (streamedOnlyCheckCode, streamedOnlyPassed) = genStreamedOnlyCheck(ctx, input)
+ val probeAllowed = if (streamedOnlyPassed.nonEmpty) {
+ s"!($anyNull) && $streamedOnlyPassed"
} else {
- s"$existsVar = true;"
+ s"!($anyNull)"
}
val resultVar = input ++ Seq(ExprCode.forNonNullValue(
@@ -704,13 +822,16 @@ trait HashJoin extends JoinCodegenSupport {
if (keyIsUnique) {
s"""
+ |boolean $existsVar = false;
|// generate join key for stream side
|${keyEv.code}
+ |$streamedOnlyCheckCode
|// find matches from HashedRelation
- |UnsafeRow $matched = $anyNull ? null: (UnsafeRow)$relationTerm.getValue(${keyEv.value});
- |boolean $existsVar = false;
- |if ($matched != null) {
- | $checkCondition
+ |if ($probeAllowed) {
+ | UnsafeRow $matched = (UnsafeRow)$relationTerm.getValue(${keyEv.value});
+ | if ($matched != null) {
+ | $checkCondition
+ | }
|}
|$numOutput.add(1);
|${consume(ctx, resultVar)}
@@ -719,15 +840,18 @@ trait HashJoin extends JoinCodegenSupport {
val matches = ctx.freshName("matches")
val iteratorCls = classOf[Iterator[UnsafeRow]].getName
s"""
+ |boolean $existsVar = false;
|// generate join key for stream side
|${keyEv.code}
+ |$streamedOnlyCheckCode
|// find matches from HashRelation
- |$iteratorCls $matches = $anyNull ? null : ($iteratorCls)$relationTerm.get(${keyEv.value});
- |boolean $existsVar = false;
- |if ($matches != null) {
- | while (!$existsVar && $matches.hasNext()) {
- | UnsafeRow $matched = (UnsafeRow) $matches.next();
- | $checkCondition
+ |if ($probeAllowed) {
+ | $iteratorCls $matches = ($iteratorCls)$relationTerm.get(${keyEv.value});
+ | if ($matches != null) {
+ | while (!$existsVar && $matches.hasNext()) {
+ | UnsafeRow $matched = (UnsafeRow) $matches.next();
+ | $checkCondition
+ | }
| }
|}
|$numOutput.add(1);
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
index 6496f9a0006e..056ddca2072d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.joins
-import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference, Expression}
import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
@@ -40,6 +40,25 @@ trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
streamPlan: SparkPlan,
buildPlan: SparkPlan,
buildRow: Option[String] = None): (String, String, Seq[ExprCode]) = {
+ if (condition.isDefined) {
+ getJoinCondition(ctx, condition.get, streamVars, streamPlan, buildPlan, buildRow)
+ } else {
+ val buildSideRow = buildRow.getOrElse(ctx.freshName("buildRow"))
+ val buildVars = genOneSideJoinVars(ctx, buildSideRow, buildPlan, setDefaultValue = false)
+ (buildSideRow, "", buildVars)
+ }
+ }
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows from an explicit expression.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ conditionExpr: Expression,
+ streamVars: Seq[ExprCode],
+ streamPlan: SparkPlan,
+ buildPlan: SparkPlan,
+ buildRow: Option[String]): (String, String, Seq[ExprCode]) = {
val buildSideRow = buildRow.getOrElse(ctx.freshName("buildRow"))
val buildVars = genOneSideJoinVars(ctx, buildSideRow, buildPlan, setDefaultValue = false)
// We want to evaluate the passed streamVars. However, evaluation modifies the contained
@@ -47,25 +66,21 @@ trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
// full outer join will want to evaluate streamVars in a different scope than the
// condition check). Because of this, we first make a copy.
val streamVars2 = streamVars.map(_.copy())
- val checkCondition = if (condition.isDefined) {
- val expr = condition.get
- // evaluate the variables that are used by the condition
- val eval = evaluateRequiredVariables(streamPlan.output ++ buildPlan.output,
- streamVars2 ++ buildVars, expr.references)
+ // evaluate the variables that are used by the condition
+ val eval = evaluateRequiredVariables(streamPlan.output ++ buildPlan.output,
+ streamVars2 ++ buildVars, conditionExpr.references)
- // filter the output via condition
- ctx.currentVars = streamVars2 ++ buildVars
- val ev =
- BindReferences.bindReference(expr, streamPlan.output ++ buildPlan.output).genCode(ctx)
- val skipRow = s"${ev.isNull} || !${ev.value}"
+ // filter the output via condition
+ ctx.currentVars = streamVars2 ++ buildVars
+ val ev = BindReferences.bindReference(
+ conditionExpr, streamPlan.output ++ buildPlan.output).genCode(ctx)
+ val skipRow = s"${ev.isNull} || !${ev.value}"
+ val checkCondition =
s"""
|$eval
|${ev.code}
|if (!($skipRow))
""".stripMargin
- } else {
- ""
- }
(buildSideRow, checkCondition, buildVars)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinEvaluatorFactory.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinEvaluatorFactory.scala
index 2b6a19dfa8a8..cf358960c710 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinEvaluatorFactory.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinEvaluatorFactory.scala
@@ -37,7 +37,9 @@ class SortMergeJoinEvaluatorFactory(
sizeInBytesSpillThreshold: Long,
numOutputRows: SQLMetric,
spillSize: SQLMetric,
- onlyBufferFirstMatchedRow: Boolean)
+ onlyBufferFirstMatchedRow: Boolean,
+ streamedOnlyCondition: Option[Expression] = None,
+ restCondition: Option[Expression] = None)
extends PartitionEvaluatorFactory[InternalRow, InternalRow] {
override def createEvaluator(): PartitionEvaluator[InternalRow, InternalRow] =
new SortMergeJoinEvaluator
@@ -136,10 +138,17 @@ class SortMergeJoinEvaluatorFactory(
spillSize,
cleanupResources)
val rightNullRow = new GenericInternalRow(right.output.length)
+ val boundStreamedOnly: InternalRow => Boolean = streamedOnlyCondition.map {
+ Predicate.create(_, left.output).eval _
+ }.getOrElse((_: InternalRow) => true)
+ val boundRest: InternalRow => Boolean = restCondition.map {
+ Predicate.create(_, left.output ++ right.output).eval _
+ }.getOrElse((_: InternalRow) => true)
new LeftOuterIterator(
smjScanner,
rightNullRow,
- boundCondition,
+ boundStreamedOnly,
+ boundRest,
resultProj,
numOutputRows).toScala
@@ -156,10 +165,17 @@ class SortMergeJoinEvaluatorFactory(
spillSize,
cleanupResources)
val leftNullRow = new GenericInternalRow(left.output.length)
+ val boundStreamedOnly: InternalRow => Boolean = streamedOnlyCondition.map {
+ Predicate.create(_, right.output).eval _
+ }.getOrElse((_: InternalRow) => true)
+ val boundRest: InternalRow => Boolean = restCondition.map {
+ Predicate.create(_, left.output ++ right.output).eval _
+ }.getOrElse((_: InternalRow) => true)
new RightOuterIterator(
smjScanner,
leftNullRow,
- boundCondition,
+ boundStreamedOnly,
+ boundRest,
resultProj,
numOutputRows).toScala
@@ -217,6 +233,12 @@ class SortMergeJoinEvaluatorFactory(
}.toScala
case LeftAnti =>
+ val boundStreamedOnly: InternalRow => Boolean = streamedOnlyCondition.map {
+ Predicate.create(_, left.output).eval _
+ }.getOrElse((_: InternalRow) => true)
+ val boundRest: InternalRow => Boolean = restCondition.map {
+ Predicate.create(_, left.output ++ right.output).eval _
+ }.getOrElse((_: InternalRow) => true)
new RowIterator {
private[this] var currentLeftRow: InternalRow = _
private[this] val smjScanner = new SortMergeJoinScanner(
@@ -236,6 +258,11 @@ class SortMergeJoinEvaluatorFactory(
override def advanceNext(): Boolean = {
while (smjScanner.findNextOuterJoinRows()) {
currentLeftRow = smjScanner.getStreamedRow
+ if (!boundStreamedOnly(currentLeftRow)) {
+ // streamed-only predicate is false/null -> full condition is false -> emit row
+ numOutputRows += 1
+ return true
+ }
val currentRightMatches = smjScanner.getBufferedMatches
if (currentRightMatches == null || currentRightMatches.length == 0) {
numOutputRows += 1
@@ -245,7 +272,7 @@ class SortMergeJoinEvaluatorFactory(
val rightMatchesIterator = currentRightMatches.generateIterator()
while (!found && rightMatchesIterator.hasNext) {
joinRow(currentLeftRow, rightMatchesIterator.next())
- if (boundCondition(joinRow)) {
+ if (boundRest(joinRow)) {
found = true
}
}
@@ -261,6 +288,12 @@ class SortMergeJoinEvaluatorFactory(
}.toScala
case j: ExistenceJoin =>
+ val boundStreamedOnly: InternalRow => Boolean = streamedOnlyCondition.map {
+ Predicate.create(_, left.output).eval _
+ }.getOrElse((_: InternalRow) => true)
+ val boundRest: InternalRow => Boolean = restCondition.map {
+ Predicate.create(_, left.output ++ right.output).eval _
+ }.getOrElse((_: InternalRow) => true)
new RowIterator {
private[this] var currentLeftRow: InternalRow = _
private[this] val result: InternalRow = new GenericInternalRow(Array[Any](null))
@@ -281,13 +314,20 @@ class SortMergeJoinEvaluatorFactory(
override def advanceNext(): Boolean = {
while (smjScanner.findNextOuterJoinRows()) {
currentLeftRow = smjScanner.getStreamedRow
+ if (!boundStreamedOnly(currentLeftRow)) {
+ // streamed-only predicate is false/null -> full condition is false ->
+ // exists=false
+ result.setBoolean(0, false)
+ numOutputRows += 1
+ return true
+ }
val currentRightMatches = smjScanner.getBufferedMatches
var found = false
if (currentRightMatches != null && currentRightMatches.length > 0) {
val rightMatchesIterator = currentRightMatches.generateIterator()
while (!found && rightMatchesIterator.hasNext) {
joinRow(currentLeftRow, rightMatchesIterator.next())
- if (boundCondition(joinRow)) {
+ if (boundRest(joinRow)) {
found = true
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala
index 51604cdfedf1..92445fb3b554 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala
@@ -121,6 +121,17 @@ case class SortMergeJoinExec(
}
}
+ /**
+ * For join types that preserve all streamed rows, split the condition into
+ * streamed-only and rest. The streamed-only part can be evaluated once per streamed row
+ * before walking the buffered matches.
+ */
+ private lazy val (streamedOnlyCondition, restCondition):
+ (Option[Expression], Option[Expression]) = {
+ StreamedSideJoinCondition.split(
+ condition, joinType, streamedPlan, conf.splitStreamedSideJoinCondition)
+ }
+
protected override def doExecute(): RDD[InternalRow] = {
val numOutputRows = longMetric("numOutputRows")
val spillSize = longMetric("spillSize")
@@ -140,7 +151,9 @@ case class SortMergeJoinExec(
sizeInBytesSpillThreshold,
numOutputRows,
spillSize,
- onlyBufferFirstMatchedRow
+ onlyBufferFirstMatchedRow,
+ streamedOnlyCondition,
+ restCondition
)
if (conf.usePartitionEvaluator) {
left.execute().zipPartitionsWithEvaluator(right.execute(), evaluatorFactory)
@@ -513,7 +526,28 @@ case class SortMergeJoinExec(
s"SortMergeJoin.doProduce should not take $x as the JoinType")
}
- val (streamedBeforeLoop, condCheck, loadStreamed) = if (condition.isDefined) {
+ // Generate streamed-only condition check for join types that preserve streamed rows.
+ val (streamedOnlyPre, streamedOnlyGuard) =
+ if (streamedOnlyCondition.isDefined) {
+ ctx.currentVars = streamedVars
+ val ev = BindReferences.bindReference(
+ streamedOnlyCondition.get, streamedPlan.output).genCode(ctx)
+ val isNullVar = ctx.freshName("streamedOnlyIsNull")
+ val valueVar = ctx.freshName("streamedOnlyValue")
+ val pre =
+ s"""
+ |${ev.code}
+ |boolean $isNullVar = ${ev.isNull};
+ |boolean $valueVar = ${ev.value};
+ """.stripMargin
+ (pre, Some(s"!$isNullVar && $valueVar"))
+ } else {
+ ("", None)
+ }
+
+ val conditionForCodegen = if (streamedOnlyGuard.isDefined) restCondition else condition
+
+ val (streamedBeforeLoop, condCheck, loadStreamed) = if (conditionForCodegen.isDefined) {
// Split the code of creating variables based on whether it's used by condition or not.
val loaded = ctx.freshName("loaded")
val (streamedBefore, streamedAfter) = splitVarsByCondition(streamedOutput, streamedVars)
@@ -521,7 +555,7 @@ case class SortMergeJoinExec(
// Generate code for condition
ctx.currentVars = streamedVars ++ bufferedVars
val cond = BindReferences.bindReference(
- condition.get, streamedPlan.output ++ bufferedPlan.output).genCode(ctx)
+ conditionForCodegen.get, streamedPlan.output ++ bufferedPlan.output).genCode(ctx)
// Evaluate the columns those used by condition before loop
val before = joinType match {
case LeftAnti =>
@@ -572,10 +606,14 @@ case class SortMergeJoinExec(
(evaluateVariables(streamedVars), "", "")
}
- val beforeLoop =
+ val existsVarDecl = existsVar.map(v => s"boolean $v = false;").getOrElse("")
+
+ val beforeLoopWithoutGuard =
s"""
|${streamedVarDecl.mkString("\n")}
|${streamedBeforeLoop.trim}
+ |$streamedOnlyPre
+ |$existsVarDecl
|scala.collection.Iterator<UnsafeRow> $iterator = $matches.generateIterator();
""".stripMargin
val outputRow =
@@ -583,10 +621,40 @@ case class SortMergeJoinExec(
|$numOutput.add(1);
|${consume(ctx, resultVars)}
""".stripMargin
+ val guardOutputRow = joinType match {
+ case LeftOuter | RightOuter =>
+ val defaultBufferedVars =
+ genOneSideJoinVars(ctx, bufferedRow, bufferedPlan, setDefaultValue = true)
+ val guardResultVars = joinType match {
+ case RightOuter => defaultBufferedVars ++ streamedVars
+ case _ => streamedVars ++ defaultBufferedVars
+ }
+ s"""
+ |$numOutput.add(1);
+ |${consume(ctx, guardResultVars)}
+ """.stripMargin
+ case _ => outputRow
+ }
val findNextJoinRows = s"$findNextJoinRowsFuncName($streamedInput, $bufferedInput)"
val thisPlan = ctx.addReferenceObj("plan", this)
val eagerCleanup = s"$thisPlan.cleanupResources();"
+ // For join types with a streamed-only guard, prepend the guard to beforeLoop
+ // so the row is emitted before the inner loop.
+ val beforeLoop = streamedOnlyGuard match {
+ case Some(guard) =>
+ s"""
+ |$beforeLoopWithoutGuard
+ |if (!($guard)) {
+ | InternalRow $bufferedRow = null;
+ | $loadStreamed
+ | $guardOutputRow
+ | continue;
+ |}
+ """.stripMargin
+ case None => beforeLoopWithoutGuard
+ }
+
val doJoin = joinType match {
case _: InnerLike =>
val cleanedFlag =
@@ -781,7 +849,6 @@ case class SortMergeJoinExec(
|while ($streamedInput.hasNext()) {
| $findNextJoinRows;
| $beforeLoop
- | boolean $exists = false;
|
| while (!$exists && $matchIterator.hasNext()) {
| InternalRow $bufferedRow = (InternalRow) $matchIterator.next();
@@ -1299,11 +1366,13 @@ private[joins] class SortMergeJoinScanner(
private class LeftOuterIterator(
smjScanner: SortMergeJoinScanner,
rightNullRow: InternalRow,
- boundCondition: InternalRow => Boolean,
+ boundStreamedOnly: InternalRow => Boolean,
+ boundRest: InternalRow => Boolean,
resultProj: InternalRow => InternalRow,
numOutputRows: SQLMetric)
extends OneSideOuterIterator(
- smjScanner, rightNullRow, boundCondition, resultProj, numOutputRows) {
+ smjScanner, rightNullRow, boundStreamedOnly, boundRest,
+ resultProj, numOutputRows) {
protected override def setStreamSideOutput(row: InternalRow): Unit = joinedRow.withLeft(row)
protected override def setBufferedSideOutput(row: InternalRow): Unit = joinedRow.withRight(row)
@@ -1315,10 +1384,13 @@ private class LeftOuterIterator(
private class RightOuterIterator(
smjScanner: SortMergeJoinScanner,
leftNullRow: InternalRow,
- boundCondition: InternalRow => Boolean,
+ boundStreamedOnly: InternalRow => Boolean,
+ boundRest: InternalRow => Boolean,
resultProj: InternalRow => InternalRow,
numOutputRows: SQLMetric)
- extends OneSideOuterIterator(smjScanner, leftNullRow, boundCondition, resultProj, numOutputRows) {
+ extends OneSideOuterIterator(
+ smjScanner, leftNullRow, boundStreamedOnly, boundRest,
+ resultProj, numOutputRows) {
protected override def setStreamSideOutput(row: InternalRow): Unit = joinedRow.withRight(row)
protected override def setBufferedSideOutput(row: InternalRow): Unit = joinedRow.withLeft(row)
@@ -1336,14 +1408,17 @@ private class RightOuterIterator(
*
* @param smjScanner a scanner that streams rows and buffers any matching rows
* @param bufferedSideNullRow the default row to return when a streamed row has no matches
- * @param boundCondition an additional filter condition for buffered rows
+ * @param boundStreamedOnly a predicate evaluated on the streamed row only
+ * @param boundRest a predicate evaluated on the joined (left ++ right) row for the residual
+ * condition, bound to the physical row order produced by this iterator
* @param resultProj how the output should be projected
* @param numOutputRows an accumulator metric for the number of rows output
*/
private abstract class OneSideOuterIterator(
smjScanner: SortMergeJoinScanner,
bufferedSideNullRow: InternalRow,
- boundCondition: InternalRow => Boolean,
+ boundStreamedOnly: InternalRow => Boolean,
+ boundRest: InternalRow => Boolean,
resultProj: InternalRow => InternalRow,
numOutputRows: SQLMetric) extends RowIterator {
@@ -1368,12 +1443,15 @@ private abstract class OneSideOuterIterator(
rightMatchesIterator = null
if (smjScanner.findNextOuterJoinRows()) {
setStreamSideOutput(smjScanner.getStreamedRow)
- if (smjScanner.getBufferedMatches.isEmpty) {
+ if (!boundStreamedOnly(smjScanner.getStreamedRow)) {
+ // Streamed-only predicate is false/null -> full condition is false -> emit null-padded row.
+ setBufferedSideOutput(bufferedSideNullRow)
+ } else if (smjScanner.getBufferedMatches.isEmpty) {
// There are no matching rows in the buffer, so return the null row
setBufferedSideOutput(bufferedSideNullRow)
} else {
- // Find the next row in the buffer that satisfied the bound condition
- if (!advanceBufferUntilBoundConditionSatisfied()) {
+ // Find the next row in the buffer that satisfied the rest condition
+ if (!advanceBufferUntilRestConditionSatisfied()) {
setBufferedSideOutput(bufferedSideNullRow)
}
}
@@ -1385,10 +1463,10 @@ private abstract class OneSideOuterIterator(
}
/**
- * Advance to the next row in the buffer that satisfies the bound condition.
+ * Advance to the next row in the buffer that satisfies the rest condition.
* @return whether there is such a row in the current buffer.
*/
- private def advanceBufferUntilBoundConditionSatisfied(): Boolean = {
+ private def advanceBufferUntilRestConditionSatisfied(): Boolean = {
var foundMatch: Boolean = false
if (rightMatchesIterator == null) {
rightMatchesIterator = smjScanner.getBufferedMatches.generateIterator()
@@ -1396,13 +1474,18 @@ private abstract class OneSideOuterIterator(
while (!foundMatch && rightMatchesIterator.hasNext) {
setBufferedSideOutput(rightMatchesIterator.next())
- foundMatch = boundCondition(joinedRow)
+ foundMatch = boundRest(joinedRow)
}
foundMatch
}
override def advanceNext(): Boolean = {
- val r = advanceBufferUntilBoundConditionSatisfied() || advanceStream()
+ // Only walk the buffered matches if we are in the middle of iterating them for the
+ // current streamed row. If the iterator is null, advanceStream() has just emitted a
+ // null-padded row (either no matches or the streamed-only predicate was false), so we
+ // must move to the next streamed row rather than re-create the match iterator.
+ val r = (rightMatchesIterator != null && advanceBufferUntilRestConditionSatisfied()) ||
+ advanceStream()
if (r) numOutputRows += 1
r
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/StreamedSideJoinCondition.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/StreamedSideJoinCondition.scala
new file mode 100644
index 000000000000..e5e5778e5d2f
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/StreamedSideJoinCondition.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{And, Expression, ExprUtils, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, JoinType, LeftAnti, LeftOuter, RightOuter}
+import org.apache.spark.sql.execution.SparkPlan
+
+/**
+ * Helper for splitting a join condition into a streamed-side-only part and the remaining part,
+ * for join types that preserve all streamed rows.
+ *
+ * The split is valid because 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. The streamed row outcome is therefore already determined before any probe: it is
+ * emitted for outer/existence joins, and emitted for left anti joins when no match exists.
+ */
+private[joins] object StreamedSideJoinCondition extends PredicateHelper {
+
+ /**
+ * Splits `condition` into conjuncts that reference only the streamed side and the remaining
+ * conjuncts, when `splitEnabled` is true and the join type preserves all streamed rows.
+ * The streamed-only part can be evaluated once per streamed row before probing/walking the
+ * buffered matches. Returns `(None, condition)` unchanged otherwise.
+ *
+ * Only conjuncts that can be evaluated unconditionally are hoisted (see
+ * [[ExprUtils.canEvaluateUnconditionally]]): the hoisted part runs for every streamed
+ * row, including rows that have no buffered match and would never evaluate the conjunct
+ * otherwise, so hoisting a conjunct that can throw 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. A whitelist of total expression families is
+ * used instead of [[Expression.throwable]], which is opt-in metadata that UDFs such as
+ * ScalaUDF do not override, so a throwing UDF would report non-throwable.
+ */
+ def split(
+ condition: Option[Expression],
+ joinType: JoinType,
+ streamedPlan: SparkPlan,
+ splitEnabled: Boolean): (Option[Expression], Option[Expression]) = {
+ val supported = joinType match {
+ case LeftAnti | LeftOuter | RightOuter | _: ExistenceJoin => true
+ case _ => false
+ }
+ if (condition.isDefined && splitEnabled && supported) {
+ val conjuncts = splitConjunctivePredicates(condition.get)
+ val (streamedOnly, rest) = conjuncts.partition(p =>
+ ExprUtils.canEvaluateUnconditionally(p) && p.references.subsetOf(streamedPlan.outputSet))
+ (streamedOnly.reduceOption(And), rest.reduceOption(And))
+ } else {
+ (None, condition)
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/ExistenceJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/ExistenceJoinSuite.scala
index 428d29f2989a..cabc2a1c34c6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/ExistenceJoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/ExistenceJoinSuite.scala
@@ -177,6 +177,43 @@ class ExistenceJoinSuite extends SharedSparkSession {
}
}
+ // Condition: a = c (equi-key) AND b < 3.0 (left-only) AND d < 4.0 (right-only)
+ private lazy val mixedResidualCondition = {
+ And(
+ And(
+ EqualTo(left.col("a").expr, right.col("c").expr),
+ LessThan(left.col("b").expr, Literal(3.0))),
+ LessThan(right.col("d").expr, Literal(4.0)))
+ }
+
+ // Condition: a = c (equi-key) AND d < 4.0 (right-only)
+ private lazy val rightOnlyResidualCondition = {
+ And(
+ EqualTo(left.col("a").expr, right.col("c").expr),
+ LessThan(right.col("d").expr, Literal(4.0)))
+ }
+
+ // Condition: a = c (equi-key) AND b < 3.0 (left-only): the residual conjunct references
+ // only the streamed side, so the split hoists it entirely and restCondition is None.
+ private lazy val leftOnlyResidualCondition = {
+ And(
+ EqualTo(left.col("a").expr, right.col("c").expr),
+ LessThan(left.col("b").expr, Literal(3.0)))
+ }
+
+ protected def testWithSplitStreamedSideCondOnAndOff(
+ testName: String)(f: String => Unit): Unit = {
+ Seq("false", "true").foreach { configValue =>
+ testWithWholeStageCodegenOnAndOff(
+ s"$testName (splitStreamedSideJoinCondition=$configValue)") { _ =>
+ withSQLConf(
+ SQLConf.SPLIT_STREAMED_SIDE_JOIN_CONDITION.key -> configValue) {
+ f(configValue)
+ }
+ }
+ }
+ }
+
// Note: the input dataframes and expression must be evaluated lazily because
// the SQLContext should be used only within a test to keep SQL tests stable
private def testExistenceJoin(
@@ -196,16 +233,22 @@ class ExistenceJoinSuite extends SharedSparkSession {
val existsAttr = AttributeReference("exists", BooleanType, false)()
val leftSemiPlus = ExistenceJoin(existsAttr)
def createLeftSemiPlusJoin(join: SparkPlan): SparkPlan = {
- val output = join.output.dropRight(1)
- val condition = if (joinType == LeftSemi) {
- existsAttr
- } else {
- Not(existsAttr)
+ joinType match {
+ case LeftSemi =>
+ val output = join.output.dropRight(1)
+ ProjectExec(output, FilterExec(existsAttr, join))
+ case LeftAnti =>
+ val output = join.output.dropRight(1)
+ ProjectExec(output, FilterExec(Not(existsAttr), join))
+ case _ =>
+ // Only LeftSemi/LeftAnti can be expressed by filtering an ExistenceJoin result.
+ join
}
- ProjectExec(output, FilterExec(condition, join))
}
- testWithWholeStageCodegenOnAndOff(s"$testName using ShuffledHashJoin") { _ =>
+ val testSemiPlusWrapper = joinType == LeftSemi || joinType == LeftAnti
+
+ testWithSplitStreamedSideCondOnAndOff(s"$testName using ShuffledHashJoin") { _ =>
extractJoinParts().foreach { case (_, leftKeys, rightKeys, boundCondition, _, _, _, _) =>
withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
@@ -214,17 +257,19 @@ class ExistenceJoinSuite extends SharedSparkSession {
leftKeys, rightKeys, joinType, BuildRight, boundCondition, left, right)),
expectedAnswer,
sortAnswers = true)
- checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
- EnsureRequirements.apply(
- createLeftSemiPlusJoin(ShuffledHashJoinExec(
- leftKeys, rightKeys, leftSemiPlus, BuildRight, boundCondition, left, right))),
- expectedAnswer,
- sortAnswers = true)
+ if (testSemiPlusWrapper) {
+ checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
+ EnsureRequirements.apply(
+ createLeftSemiPlusJoin(ShuffledHashJoinExec(
+ leftKeys, rightKeys, leftSemiPlus, BuildRight, boundCondition, left, right))),
+ expectedAnswer,
+ sortAnswers = true)
+ }
}
}
}
- testWithWholeStageCodegenOnAndOff(s"$testName using BroadcastHashJoin") { _ =>
+ testWithSplitStreamedSideCondOnAndOff(s"$testName using BroadcastHashJoin") { _ =>
extractJoinParts().foreach { case (_, leftKeys, rightKeys, boundCondition, _, _, _, _) =>
withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
@@ -233,17 +278,19 @@ class ExistenceJoinSuite extends SharedSparkSession {
leftKeys, rightKeys, joinType, BuildRight, boundCondition, left, right)),
expectedAnswer,
sortAnswers = true)
- checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
- EnsureRequirements.apply(
- createLeftSemiPlusJoin(BroadcastHashJoinExec(
- leftKeys, rightKeys, leftSemiPlus, BuildRight, boundCondition, left, right))),
- expectedAnswer,
- sortAnswers = true)
+ if (testSemiPlusWrapper) {
+ checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
+ EnsureRequirements.apply(
+ createLeftSemiPlusJoin(BroadcastHashJoinExec(
+ leftKeys, rightKeys, leftSemiPlus, BuildRight, boundCondition, left, right))),
+ expectedAnswer,
+ sortAnswers = true)
+ }
}
}
}
- testWithWholeStageCodegenOnAndOff(s"$testName using SortMergeJoin") { _ =>
+ testWithSplitStreamedSideCondOnAndOff(s"$testName using SortMergeJoin") { _ =>
extractJoinParts().foreach { case (_, leftKeys, rightKeys, boundCondition, _, _, _, _) =>
withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
@@ -251,45 +298,56 @@ class ExistenceJoinSuite extends SharedSparkSession {
SortMergeJoinExec(leftKeys, rightKeys, joinType, boundCondition, left, right)),
expectedAnswer,
sortAnswers = true)
+ if (testSemiPlusWrapper) {
+ checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
+ EnsureRequirements.apply(
+ createLeftSemiPlusJoin(SortMergeJoinExec(
+ leftKeys, rightKeys, leftSemiPlus, boundCondition, left, right))),
+ expectedAnswer,
+ sortAnswers = true)
+ }
+ }
+ }
+ }
+
+ Seq("false", "true").foreach { configValue =>
+ test(s"$testName using BroadcastNestedLoopJoin build left" +
+ s" (splitStreamedSideJoinCondition=$configValue)") {
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1",
+ SQLConf.SPLIT_STREAMED_SIDE_JOIN_CONDITION.key -> configValue) {
checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
EnsureRequirements.apply(
- createLeftSemiPlusJoin(SortMergeJoinExec(
- leftKeys, rightKeys, leftSemiPlus, boundCondition, left, right))),
+ BroadcastNestedLoopJoinExec(left, right, BuildLeft, joinType, condition)),
expectedAnswer,
sortAnswers = true)
+ if (testSemiPlusWrapper) {
+ checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
+ EnsureRequirements.apply(
+ createLeftSemiPlusJoin(BroadcastNestedLoopJoinExec(
+ left, right, BuildLeft, leftSemiPlus, condition))),
+ expectedAnswer,
+ sortAnswers = true)
+ }
}
}
}
- test(s"$testName using BroadcastNestedLoopJoin build left") {
- withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
- checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
- EnsureRequirements.apply(
- BroadcastNestedLoopJoinExec(left, right, BuildLeft, joinType, condition)),
- expectedAnswer,
- sortAnswers = true)
- checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
- EnsureRequirements.apply(
- createLeftSemiPlusJoin(BroadcastNestedLoopJoinExec(
- left, right, BuildLeft, leftSemiPlus, condition))),
- expectedAnswer,
- sortAnswers = true)
- }
- }
-
- testWithWholeStageCodegenOnAndOff(s"$testName using BroadcastNestedLoopJoin build right") { _ =>
+ testWithSplitStreamedSideCondOnAndOff(
+ s"$testName using BroadcastNestedLoopJoin build right") { _ =>
withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
EnsureRequirements.apply(
BroadcastNestedLoopJoinExec(left, right, BuildRight, joinType, condition)),
expectedAnswer,
sortAnswers = true)
- checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
- EnsureRequirements.apply(
- createLeftSemiPlusJoin(BroadcastNestedLoopJoinExec(
- left, right, BuildRight, leftSemiPlus, condition))),
- expectedAnswer,
- sortAnswers = true)
+ if (testSemiPlusWrapper) {
+ checkAnswer2(leftRows, rightRows, (left: SparkPlan, right: SparkPlan) =>
+ EnsureRequirements.apply(
+ createLeftSemiPlusJoin(BroadcastNestedLoopJoinExec(
+ left, right, BuildRight, leftSemiPlus, condition))),
+ expectedAnswer,
+ sortAnswers = true)
+ }
}
}
}
@@ -401,4 +459,242 @@ class ExistenceJoinSuite extends SharedSparkSession {
Some(And(EqualTo(left.col("a").expr, rightUniqueKey.col("c").expr),
LessThan(left.col("b").expr, rightUniqueKey.col("d").expr))),
Seq(Row(1, 2.0), Row(1, 2.0), Row(3, 3.0), Row(null, null), Row(null, 5.0), Row(6, null)))
+
+ // ---- Tests for streamed-side-only residual predicate hoisting ----
+
+ // LeftAnti: rows where b >= 3.0 OR no right match with d < 4.0
+ // (1, 2.0): no c=1 match -> emitted
+ // (2, 1.0): match exists -> dropped
+ // (3, 3.0): b=3.0 >= 3.0 -> emitted
+ // (null, null): null key -> emitted
+ // (null, 5.0): b=5.0 >= 3.0 -> emitted
+ // (6, null): b=null -> emitted
+ testExistenceJoin(
+ "test mixed residual condition for left anti join",
+ LeftAnti,
+ left,
+ right,
+ Some(mixedResidualCondition),
+ Seq(Row(1, 2.0), Row(1, 2.0), Row(3, 3.0), Row(null, null), Row(null, 5.0), Row(6, null)))
+
+ // LeftOuter: rows where b < 3.0 and right match with d < 4.0 get matched; others emitted as
+ // null-padded. Equi-matches: (2,1.0)-(2,3.0), (3,3.0)-(3,2.0), (6,null)-(6,null).
+ // For (2,1.0): b<3.0 true, right d=3.0<4.0 true -> matched output (2,1.0,2,3.0).
+ // Both sides contain two rows with the matching key, so the Cartesian product emits 4 rows.
+ // For (3,3.0): b<3.0 false -> emitted as (3,3.0,null,null)
+ // For (6,null): b<3.0 null -> emitted as (6,null,null,null)
+ // For (1,2.0): no equi-match -> emitted as (1,2.0,null,null)
+ // Null keys are emitted as null-padded.
+ testExistenceJoin(
+ "test mixed residual condition for left outer join",
+ LeftOuter,
+ left,
+ right,
+ Some(mixedResidualCondition),
+ Seq(
+ Row(1, 2.0, null, null),
+ Row(1, 2.0, null, null),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(3, 3.0, null, null),
+ Row(null, null, null, null),
+ Row(null, 5.0, null, null),
+ Row(6, null, null, null)))
+
+ // ExistenceJoin with the same mixed residual condition: exists=true only for (2,1.0).
+ testExistenceJoin(
+ "test mixed residual condition for existence join",
+ ExistenceJoin(AttributeReference("exists", BooleanType, false)()),
+ left,
+ right,
+ Some(mixedResidualCondition),
+ Seq(
+ Row(1, 2.0, false),
+ Row(1, 2.0, false),
+ Row(2, 1.0, true),
+ Row(2, 1.0, true),
+ Row(3, 3.0, false),
+ Row(null, null, false),
+ Row(null, 5.0, false),
+ Row(6, null, false)))
+
+ // Inner join is outside the split whitelist: nothing is hoisted and the results must be
+ // identical with the split config on and off.
+ testExistenceJoin(
+ "test mixed residual condition for inner join",
+ Inner,
+ left,
+ right,
+ Some(mixedResidualCondition),
+ Seq(
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0)))
+
+ // Right-only residual: nothing is hoistable (the only residual conjunct references the
+ // non-streamed side), so d < 4.0 must stay in the join condition and the results must be
+ // identical with the split config on and off. Compared to the mixed condition above, the
+ // b < 3.0 conjunct is gone, so (3, 3.0) now matches.
+
+ // LeftAnti: (2, 1.0) and (3, 3.0) match -> dropped; (6, null) has a key match but d is
+ // null -> no qualifying match -> emitted.
+ testExistenceJoin(
+ "test right-only residual condition for left anti join",
+ LeftAnti,
+ left,
+ right,
+ Some(rightOnlyResidualCondition),
+ Seq(Row(1, 2.0), Row(1, 2.0), Row(null, null), Row(null, 5.0), Row(6, null)))
+
+ // LeftOuter: (2, 1.0) matches both right (2, 3.0) rows (2 left rows -> 4 output rows);
+ // (3, 3.0) matches (3, 2.0); everything else is emitted null-padded, including (6, null)
+ // whose key match has d = null.
+ testExistenceJoin(
+ "test right-only residual condition for left outer join",
+ LeftOuter,
+ left,
+ right,
+ Some(rightOnlyResidualCondition),
+ Seq(
+ Row(1, 2.0, null, null),
+ Row(1, 2.0, null, null),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(3, 3.0, 3, 2.0),
+ Row(null, null, null, null),
+ Row(null, 5.0, null, null),
+ Row(6, null, null, null)))
+
+ // ExistenceJoin with the same right-only residual condition: exists=true for (2, 1.0) and
+ // (3, 3.0).
+ testExistenceJoin(
+ "test right-only residual condition for existence join",
+ ExistenceJoin(AttributeReference("exists", BooleanType, false)()),
+ left,
+ right,
+ Some(rightOnlyResidualCondition),
+ Seq(
+ Row(1, 2.0, false),
+ Row(1, 2.0, false),
+ Row(2, 1.0, true),
+ Row(2, 1.0, true),
+ Row(3, 3.0, true),
+ Row(null, null, false),
+ Row(null, 5.0, false),
+ Row(6, null, false)))
+
+ // Left-only residual: the only residual conjunct references the streamed side, so the
+ // split hoists it entirely and restCondition is None. These cases cover the fully-hoisted
+ // shape (SortMergeJoinExec codegen's conditionForCodegen = None branch and the
+ // interpreted always-true rest walk). The expected answers happen to match the
+ // mixed-condition ones above because d < 4.0 is true for every equi-match in the
+ // fixtures; the point of these cases is the restCondition = None code shape.
+
+ // LeftAnti: (2, 1.0) matches -> dropped; (3, 3.0) fails the hoisted b < 3.0; (6, null)
+ // evaluates it to null; null keys never match.
+ testExistenceJoin(
+ "test left-only residual condition for left anti join",
+ LeftAnti,
+ left,
+ right,
+ Some(leftOnlyResidualCondition),
+ Seq(Row(1, 2.0), Row(1, 2.0), Row(3, 3.0), Row(null, null), Row(null, 5.0), Row(6, null)))
+
+ // LeftOuter: (2, 1.0) matches both right (2, 3.0) rows (2 left rows -> 4 output rows);
+ // everything else is emitted null-padded, including (3, 3.0) which fails the hoisted
+ // b < 3.0 and (6, null) for which it is null.
+ testExistenceJoin(
+ "test left-only residual condition for left outer join",
+ LeftOuter,
+ left,
+ right,
+ Some(leftOnlyResidualCondition),
+ Seq(
+ Row(1, 2.0, null, null),
+ Row(1, 2.0, null, null),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(3, 3.0, null, null),
+ Row(null, null, null, null),
+ 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).
+ testExistenceJoin(
+ "test left-only residual condition for existence join",
+ ExistenceJoin(AttributeReference("exists", BooleanType, false)()),
+ left,
+ right,
+ Some(leftOnlyResidualCondition),
+ Seq(
+ Row(1, 2.0, false),
+ Row(1, 2.0, false),
+ Row(2, 1.0, true),
+ Row(2, 1.0, true),
+ Row(3, 3.0, false),
+ Row(null, null, false),
+ Row(null, 5.0, false),
+ Row(6, null, false)))
+
+ // Inner join is outside the split whitelist: with a fully streamed-side-only residual,
+ // nothing may be hoisted and the results must be identical with the split config on and
+ // off.
+ testExistenceJoin(
+ "test left-only residual condition for inner join",
+ Inner,
+ left,
+ right,
+ Some(leftOnlyResidualCondition),
+ Seq(
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0)))
+
+ // RightOuter streams the right side, so here d < 4.0 is the streamed-side-only conjunct
+ // and IS hoistable (the mirror of the left-streamed cases above). The hash joins use
+ // BuildLeft so that the right side is the streamed one; right rows with no qualifying
+ // left match are emitted null-padded.
+ testWithSplitStreamedSideCondOnAndOff(
+ "test right-only residual condition for right outer join") { _ =>
+ val join = Join(left.logicalPlan, right.logicalPlan,
+ Inner, Some(rightOnlyResidualCondition), JoinHint.NONE)
+ ExtractEquiJoinKeys.unapply(join).foreach {
+ case (_, leftKeys, rightKeys, boundCondition, _, _, _, _) =>
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
+ val expected = Seq(
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(2, 1.0, 2, 3.0),
+ Row(3, 3.0, 3, 2.0),
+ Row(null, null, 4, 1.0),
+ Row(null, null, null, null),
+ Row(null, null, null, 5.0),
+ Row(null, null, 6, null))
+ checkAnswer2(left, right, (l: SparkPlan, r: SparkPlan) =>
+ EnsureRequirements.apply(
+ SortMergeJoinExec(leftKeys, rightKeys, RightOuter, boundCondition, l, r)),
+ expected, sortAnswers = true)
+ checkAnswer2(left, right, (l: SparkPlan, r: SparkPlan) =>
+ EnsureRequirements.apply(
+ BroadcastHashJoinExec(
+ leftKeys, rightKeys, RightOuter, BuildLeft, boundCondition, l, r)),
+ expected, sortAnswers = true)
+ checkAnswer2(left, right, (l: SparkPlan, r: SparkPlan) =>
+ EnsureRequirements.apply(
+ ShuffledHashJoinExec(
+ leftKeys, rightKeys, RightOuter, BuildLeft, boundCondition, l, r)),
+ expected, sortAnswers = true)
+ }
+ }
+ }
+
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/SplitStreamedSideJoinConditionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/SplitStreamedSideJoinConditionSuite.scala
new file mode 100644
index 000000000000..ec3f8d3c8c43
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/SplitStreamedSideJoinConditionSuite.scala
@@ -0,0 +1,245 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.{DataFrame, QueryTest}
+import org.apache.spark.sql.catalyst.expressions.{Add, And, LessThan, Literal, Rand}
+import org.apache.spark.sql.catalyst.plans.LeftOuter
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for streamed-side join condition hoisting in whole-stage codegen. The
+ * generated-code shape under test requires all of the following:
+ * - spark.sql.join.splitStreamedSideJoinCondition = true
+ * - a join type that preserves streamed rows (LeftOuter, RightOuter, LeftAnti, ExistenceJoin)
+ * - 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 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 uses continue in its own loop instead of returning from an inlined consumer.
+ * ExistenceJoinSuite covers those join implementations functionally, with the
+ * config on and off.
+ * - a batching streamed producer (vectorized parquet scan -> ColumnarToRowExec) whose loop
+ * cursor is only written back after the inlined consume code
+ * - a streamed column (pad) not referenced by the join keys/condition, so
+ * WholeStageCodegenExec.consume does not wrap the join's doConsume in a function
+ * - a residual filter that references pad (pad % 2 = 0 is not pushed into the scan), so pad
+ * is materialized before the join's doConsume and the generated code compiles
+ *
+ * Expected results are produced by running the same query with the hoisting disabled. The
+ * LIMIT below ORDER BY on the hoisting-enabled run is load-bearing: an early-returning guard
+ * makes the generated code reprocess the same batch forever; the limit is what makes the
+ * query terminate (with wrong results).
+ */
+class SplitStreamedSideJoinConditionSuite extends QueryTest with SharedSparkSession {
+
+ // streamed_t: (id, a, pad) with id 0..7, a = id, pad = id * 10
+ // build_t: (bid, b) with bid 0..3, b = bid * 100
+ private def withStreamedAndBuildTables(f: => Unit): Unit = {
+ withTempPath { path =>
+ spark.range(8).selectExpr("id", "id AS a", "id * 10 AS pad")
+ .write.parquet(path.getCanonicalPath + "/streamed")
+ spark.range(4).selectExpr("id AS bid", "id * 100 AS b")
+ .write.parquet(path.getCanonicalPath + "/build")
+ spark.read.parquet(path.getCanonicalPath + "/streamed")
+ .createOrReplaceTempView("streamed_t")
+ spark.read.parquet(path.getCanonicalPath + "/build")
+ .createOrReplaceTempView("build_t")
+ f
+ }
+ }
+
+ // Compares the query with hoisting enabled against the same query with hoisting disabled,
+ // sorting both sides by all columns.
+ private def checkDataFrame(query: => DataFrame): Unit = {
+ val expected = withSQLConf(SQLConf.SPLIT_STREAMED_SIDE_JOIN_CONDITION.key -> "false") {
+ sortByAllColumns(query).collect().toSeq
+ }
+ val df = query
+ assert(df.queryExecution.executedPlan.toString.contains("BroadcastHashJoin"),
+ s"expected a broadcast hash join in ${df.queryExecution.executedPlan}")
+ checkAnswer(sortByAllColumns(df.limit(100)), expected)
+ }
+
+ private def sortByAllColumns(df: DataFrame): DataFrame = {
+ val cols = df.columns.toSeq
+ df.orderBy(cols.head, cols.tail: _*)
+ }
+
+ private def testWithCodegenOnAndOff(testName: String)(
+ queryDf: => DataFrame,
+ codegenOnly: Boolean = false): Unit = {
+ testWithWholeStageCodegenOnAndOff(testName) { _ =>
+ val confs = Seq(
+ SQLConf.SPLIT_STREAMED_SIDE_JOIN_CONDITION.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") ++
+ (if (codegenOnly) Seq(SQLConf.CODEGEN_FACTORY_MODE.key -> "CODEGEN_ONLY") else Nil)
+ withSQLConf(confs: _*) {
+ withStreamedAndBuildTables {
+ checkDataFrame(queryDf)
+ }
+ }
+ }
+ }
+
+ // Streamed-only conjuncts that can throw or are non-deterministic must stay in the
+ // residual: hoisting evaluates the hoisted part for every streamed row, including rows
+ // that have no buffered match and would never evaluate the conjunct otherwise.
+
+ test("split does not hoist conjuncts that can throw") {
+ val streamed = spark.range(8).selectExpr("id", "id AS a").queryExecution.executedPlan
+ val a = streamed.output.find(_.name == "a").get
+ val hoistable = LessThan(a, Literal(5L))
+ // Add can throw on overflow in ANSI mode but inherits throwable = false from its
+ // non-throwing children. It must stay in the residual, which is why a !throwable
+ // gate is not sufficient and the hoisted part is restricted to expression families
+ // with a reliable non-throwing contract.
+ val unmarkedThrowable = LessThan(Add(a, Literal(1L)), Literal(5L))
+ assert(!unmarkedThrowable.throwable)
+ val (streamedOnly, rest) = StreamedSideJoinCondition.split(
+ Some(And(hoistable, unmarkedThrowable)), LeftOuter, streamed, splitEnabled = true)
+ assert(streamedOnly.contains(hoistable))
+ assert(rest.contains(unmarkedThrowable))
+ }
+
+ test("split does not hoist non-deterministic conjuncts") {
+ val streamed = spark.range(8).selectExpr("id", "id AS a").queryExecution.executedPlan
+ val a = streamed.output.find(_.name == "a").get
+ val hoistable = LessThan(a, Literal(5L))
+ val nonDeterministic = LessThan(Rand(Literal(0L)), Literal(0.5))
+ val (streamedOnly, rest) = StreamedSideJoinCondition.split(
+ Some(And(hoistable, nonDeterministic)), LeftOuter, streamed, splitEnabled = true)
+ assert(streamedOnly.contains(hoistable))
+ assert(rest.contains(nonDeterministic))
+ }
+
+ // The streamed-only condition a < 5 is false for ids 5..7, so the hoisted guard fires on
+ // those rows. pad is selected but not referenced by the join.
+
+ testWithCodegenOnAndOff(
+ "left outer join, unique build key")(
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ s.id, s.a, s.pad, b.bid
+ |FROM (SELECT * FROM streamed_t WHERE pad % 2 = 0) s
+ |LEFT OUTER JOIN (SELECT DISTINCT bid FROM build_t) b
+ |ON s.id = b.bid AND s.a < 5
+ """.stripMargin))
+
+ testWithCodegenOnAndOff(
+ "left outer join, non-unique build key")(
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ s.id, s.a, s.pad, b.bid, b.b
+ |FROM (SELECT * FROM streamed_t WHERE pad % 2 = 0) s
+ |LEFT OUTER JOIN (SELECT bid, b FROM build_t UNION ALL SELECT bid, b FROM build_t) b
+ |ON s.id = b.bid AND s.a < 5
+ """.stripMargin))
+
+ // The "compiles without pre-materialized column" variants drop the residual filter, so pad
+ // stays lazy through the join and is only materialized by the join's own consume code. They
+ // exercise how the generated guard / probe code scopes that materialization: if pad's
+ // declaration lands in a different (nested) scope than its uses, the generated code does
+ // not compile. CODEGEN_ONLY turns the silent fallback to interpreted execution into an
+ // error so these tests actually assert compilation succeeds.
+ testWithCodegenOnAndOff(
+ "left outer join compiles without pre-materialized column")(
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ s.id, s.a, s.pad, b.bid, b.b
+ |FROM streamed_t s
+ |LEFT OUTER JOIN build_t b
+ |ON s.id = b.bid AND s.a < 5
+ """.stripMargin),
+ codegenOnly = true)
+
+ testWithCodegenOnAndOff(
+ "right outer join")(
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ b.bid, b.b, s.id, s.a, s.pad
+ |FROM build_t b
+ |RIGHT OUTER JOIN (SELECT * FROM streamed_t WHERE pad % 2 = 0) s
+ |ON b.bid = s.id AND s.a < 5
+ """.stripMargin))
+
+ testWithCodegenOnAndOff(
+ "left anti join")(
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ s.id, s.a, s.pad
+ |FROM (SELECT * FROM streamed_t WHERE pad % 2 = 0) s
+ |LEFT ANTI JOIN build_t b
+ |ON s.id = b.bid AND s.a < 5
+ """.stripMargin))
+
+ testWithCodegenOnAndOff(
+ "left anti join compiles without pre-materialized column")(
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ s.id, s.a, s.pad
+ |FROM streamed_t s
+ |LEFT ANTI JOIN build_t b
+ |ON s.id = b.bid AND s.a < 5
+ """.stripMargin),
+ codegenOnly = true)
+
+ // The EXISTS flag is projected (not filtered on) so guard-fired rows (ids 5..7) are
+ // emitted via the guard path with e = false, making any replay of those rows visible as
+ // duplicates. Filtering on EXISTS instead would drop them and mask a premature return in
+ // the guard: the stage buffer stays empty, BufferedRowIterator.hasNext treats that as
+ // end-of-stream, and since the abandoned rows produce no output anyway, the query would
+ // return the right answer for the wrong reason.
+ testWithCodegenOnAndOff(
+ "existence join")(
+ spark.sql("""
+ |SELECT s.id, s.a, s.pad,
+ | EXISTS (SELECT /*+ BROADCAST(b) */ 1 FROM build_t b WHERE b.bid = s.id AND s.a < 5) AS e
+ |FROM (SELECT * FROM streamed_t WHERE pad % 2 = 0) s
+ """.stripMargin))
+
+ testWithCodegenOnAndOff(
+ "existence join compiles without pre-materialized column")(
+ spark.sql("""
+ |SELECT s.id, s.a, s.pad,
+ | EXISTS (SELECT /*+ BROADCAST(b) */ 1 FROM build_t b WHERE b.bid = s.id AND s.a < 5) AS e
+ |FROM streamed_t s
+ """.stripMargin),
+ codegenOnly = true)
+
+ // Left outer join whose ON clause has two streamed-side-only conjuncts: a hoistable
+ // a < 5 and a real registered UDF that returns true when its input is true and throws
+ // when it is false or null. The UDF conjunct holds for the matched rows (ids 0..3) and
+ // throws for the unmatched ones (ids 4..7), so it throws exactly when it is wrongly
+ // hoisted and evaluated before probing. A ScalaUDF does not override `throwable`, so
+ // the old !throwable gate hoisted it; the whitelist keeps it in the residual.
+ private def throwingUdfJoin: DataFrame = {
+ spark.udf.register("throw_unless_true", (v: Boolean) =>
+ if (v) true else throw new RuntimeException("throw_unless_true evaluated to false"))
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ s.id, s.a, b.bid, b.b
+ |FROM streamed_t s
+ |LEFT OUTER JOIN build_t b
+ |ON s.id = b.bid AND s.a < 5 AND throw_unless_true(s.a < 4)
+ """.stripMargin)
+ }
+
+ // End-to-end: the throwing UDF conjunct must not run for streamed rows 4..7, which have
+ // no buffered match. The hoistable conjunct a < 5 keeps the hoisted guard on the path
+ // so the test proves the UDF conjunct is excluded from it, not that hoisting is off.
+ // Before the fix, rows 4..7 evaluate the hoisted UDF conjunct and throw.
+ testWithCodegenOnAndOff(
+ "throwing UDF conjunct is not evaluated for unmatched rows")(
+ throwingUdfJoin)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org