From d5ecff018640c8b09f0e56645a49920a1f40cc47 Mon Sep 17 00:00:00 2001 From: James Xu Date: Tue, 7 Jul 2026 23:46:33 +0800 Subject: [PATCH 1/2] [SPARK-58066][SQL] Hoist streamed-side predicates in outer/anti/existence 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). --- .../apache/spark/sql/internal/SQLConf.scala | 14 ++ .../spark/sql/execution/joins/HashJoin.scala | 228 ++++++++++++++---- .../execution/joins/JoinCodegenSupport.scala | 43 ++-- .../joins/SortMergeJoinEvaluatorFactory.scala | 66 ++++- .../execution/joins/SortMergeJoinExec.scala | 125 ++++++++-- .../execution/joins/ExistenceJoinSuite.scala | 203 ++++++++++++---- 6 files changed, 543 insertions(+), 136 deletions(-) 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 1ca873d7da1a1..831c8876f5102 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 @@ -897,6 +897,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.SESSION) + .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 " + @@ -8274,6 +8286,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 enableRadixSort: Boolean = getConf(RADIX_SORT_ENABLED) 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 9df791aa8de0c..ba87165e96c71 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 @@ -41,7 +41,7 @@ private[joins] case class HashedRelationInfo( keyIsUnique: Boolean, isEmpty: Boolean) -trait HashJoin extends JoinCodegenSupport { +trait HashJoin extends JoinCodegenSupport with PredicateHelper { assert(leftKeys.forall(key => UnsafeRowUtils.isBinaryStable(key.dataType))) assert(rightKeys.forall(key => UnsafeRowUtils.isBinaryStable(key.dataType))) @@ -152,6 +152,34 @@ 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]) = { + if (condition.isDefined && conf.splitStreamedSideJoinCondition && + (joinType match { + case LeftAnti | LeftOuter | RightOuter | _: ExistenceJoin => true + case _ => false + })) { + val conjuncts = splitConjunctivePredicates(condition.get) + val (streamedOnly, rest) = conjuncts.partition(_.references.subsetOf(streamedPlan.outputSet)) + (streamedOnly.reduceOption(And), rest.reduceOption(And)) + } else { + (None, condition) + } + } + + @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 +233,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 +250,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 +317,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 +355,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 || matched == null || !boundStreamedOnlyCondition(current) || + (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 || buildIter == null || !boundStreamedOnlyCondition(current) || + (restCondition.isDefined && !buildIter.exists { + row => boundRestCondition(joinedRow(current, row)) + }) } } } @@ -357,6 +404,18 @@ 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) + } + override def doProduce(ctx: CodegenContext): String = { streamedPlan.asInstanceOf[CodegenSupport].produce(ctx, this) } @@ -457,14 +516,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 +542,32 @@ trait HashJoin extends JoinCodegenSupport { case BuildRight => input ++ buildVars } + // Variables for the guard-false path: emit the streamed row with null build side. + // They must be separate ExprCode instances so that the shared resultVars used in the + // main path are not consumed/cleared by the guard-false consume. + val defaultBuildVars = genOneSideJoinVars(ctx, matched, buildPlan, setDefaultValue = true) + val defaultResultVars = buildSide match { + case BuildLeft => defaultBuildVars ++ input + case BuildRight => input ++ defaultBuildVars + } + + // Early-return guard: if streamed-only predicate is false/null, full condition is false + // and the row is emitted with null build side, so skip the probe entirely. + val streamedOnlyGuard = streamedOnlyCondition.map { expr => + val ev = genStreamedOnlyCondition(ctx, expr, input) + s""" + |${ev.code} + |if (${ev.isNull} || !${ev.value}) { + | UnsafeRow $matched = null; + | $numOutput.add(1); + | ${consume(ctx, defaultResultVars)} + | return; + |} + """.stripMargin + }.getOrElse("") + if (keyIsUnique) { - val resetWhenConditionFails = if (hasCondition) { + val resetWhenConditionFails = if (hasRestCondition) { s""" |if (!$conditionPassed) { | $matched = null; @@ -501,6 +581,7 @@ trait HashJoin extends JoinCodegenSupport { s""" |// generate join key for stream side |${keyEv.code} + |$streamedOnlyGuard |// find matches from HashedRelation |UnsafeRow $matched = $anyNull ? null: (UnsafeRow)$relationTerm.getValue(${keyEv.value}); |${checkCondition.trim} @@ -525,11 +606,12 @@ 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} + |$streamedOnlyGuard |// find matches from HashRelation |$iteratorCls $matches = $anyNull ? null : ($iteratorCls)$relationTerm.get(${keyEv.value}); |boolean $found = false; @@ -617,7 +699,26 @@ 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) + } + + // Early-return guard: if streamed-only predicate is false/null, full condition is false + // and the row is guaranteed emitted, so skip the probe entirely. + val streamedOnlyGuard = streamedOnlyCondition.map { expr => + val ev = genStreamedOnlyCondition(ctx, expr, input) + s""" + |${ev.code} + |if (${ev.isNull} || !${ev.value}) { + | $numOutput.add(1); + | ${consume(ctx, input)} + | return; + |} + """.stripMargin + }.getOrElse("") if (keyIsUnique) { val found = ctx.freshName("found") @@ -625,6 +726,7 @@ trait HashJoin extends JoinCodegenSupport { |boolean $found = false; |// generate join key for stream side |${keyEv.code} + |$streamedOnlyGuard |// Check if the key has nulls. |if (!($anyNull)) { | // Check if the HashedRelation exists. @@ -649,6 +751,7 @@ trait HashJoin extends JoinCodegenSupport { |boolean $found = false; |// generate join key for stream side |${keyEv.code} + |$streamedOnlyGuard |// Check if the key has nulls. |if (!($anyNull)) { | // Check if the HashedRelation exists. @@ -682,35 +785,55 @@ 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) + + // 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;" + } + + // Early-return guard: if streamed-only predicate is false/null, full condition is false + // and the exists flag is false, so skip the probe entirely. + val existsFalseResultVar = input ++ Seq(ExprCode.forNonNullValue( + JavaCode.variable(existsVar, BooleanType))) + val streamedOnlyGuard = streamedOnlyCondition.map { expr => + val ev = genStreamedOnlyCondition(ctx, expr, input) s""" - |$eval |${ev.code} - |$existsVar = !${ev.isNull} && ${ev.value}; + |if (${ev.isNull} || !${ev.value}) { + | $existsVar = false; + | $numOutput.add(1); + | ${consume(ctx, existsFalseResultVar)} + | return; + |} """.stripMargin - } else { - s"$existsVar = true;" - } + }.getOrElse("") val resultVar = input ++ Seq(ExprCode.forNonNullValue( JavaCode.variable(existsVar, BooleanType))) if (keyIsUnique) { s""" + |boolean $existsVar = false; |// generate join key for stream side |${keyEv.code} + |$streamedOnlyGuard |// find matches from HashedRelation - |UnsafeRow $matched = $anyNull ? null: (UnsafeRow)$relationTerm.getValue(${keyEv.value}); - |boolean $existsVar = false; - |if ($matched != null) { - | $checkCondition + |if (!($anyNull)) { + | UnsafeRow $matched = (UnsafeRow)$relationTerm.getValue(${keyEv.value}); + | if ($matched != null) { + | $checkCondition + | } |} |$numOutput.add(1); |${consume(ctx, resultVar)} @@ -719,15 +842,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} + |$streamedOnlyGuard |// 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 (!($anyNull)) { + | $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 6496f9a0006e2..056ddca2072d7 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 2b6a19dfa8a8d..c9ff5b32fb1cc 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 @@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.joins import org.apache.spark.{PartitionEvaluator, PartitionEvaluatorFactory} import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, GenericInternalRow, JoinedRow, Predicate, Projection, RowOrdering, UnsafeProjection, UnsafeRow} -import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter, InnerLike, JoinType, LeftAnti, LeftOuter, LeftSemi, RightOuter} +import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter, InnerLike, JoinType, LeftAnti, LeftExistence, LeftOuter, LeftSemi, RightOuter} import org.apache.spark.sql.execution.{ExternalAppendOnlyUnsafeRowArray, RowIterator, SparkPlan} import org.apache.spark.sql.execution.metric.SQLMetric @@ -37,13 +37,29 @@ 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 private class SortMergeJoinEvaluator extends PartitionEvaluator[InternalRow, InternalRow] { + private[this] val streamedPlan = joinType match { + case _: InnerLike | LeftOuter | FullOuter | LeftExistence(_) => left + case RightOuter => right + case x => throw new IllegalArgumentException( + s"SortMergeJoinEvaluatorFactory.streamedPlan should not take $x as the JoinType") + } + + private[this] val bufferedPlan = joinType match { + case _: InnerLike | LeftOuter | FullOuter | LeftExistence(_) => right + case RightOuter => left + case x => throw new IllegalArgumentException( + s"SortMergeJoinEvaluatorFactory.bufferedPlan should not take $x as the JoinType") + } + private def cleanupResources(): Unit = { IndexedSeq(left, right).foreach(_.cleanupResources()) } @@ -136,10 +152,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 +179,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 +247,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 +272,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 +286,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 +302,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 +328,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 51604cdfedf1c..d46c577c8d3ad 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 @@ -43,7 +43,7 @@ case class SortMergeJoinExec( condition: Option[Expression], left: SparkPlan, right: SparkPlan, - isSkewJoin: Boolean = false) extends ShuffledJoin { + isSkewJoin: Boolean = false) extends ShuffledJoin with PredicateHelper { override lazy val metrics = Map( "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"), @@ -121,6 +121,26 @@ 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]) = { + if (condition.isDefined && conf.splitStreamedSideJoinCondition && + (joinType match { + case LeftAnti | LeftOuter | RightOuter | _: ExistenceJoin => true + case _ => false + })) { + val conjuncts = splitConjunctivePredicates(condition.get) + val (streamedOnly, rest) = conjuncts.partition(_.references.subsetOf(streamedPlan.outputSet)) + (streamedOnly.reduceOption(And), rest.reduceOption(And)) + } else { + (None, condition) + } + } + protected override def doExecute(): RDD[InternalRow] = { val numOutputRows = longMetric("numOutputRows") val spillSize = longMetric("spillSize") @@ -140,7 +160,9 @@ case class SortMergeJoinExec( sizeInBytesSpillThreshold, numOutputRows, spillSize, - onlyBufferFirstMatchedRow + onlyBufferFirstMatchedRow, + streamedOnlyCondition, + restCondition ) if (conf.usePartitionEvaluator) { left.execute().zipPartitionsWithEvaluator(right.execute(), evaluatorFactory) @@ -513,7 +535,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 +564,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 +615,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 $iterator = $matches.generateIterator(); """.stripMargin val outputRow = @@ -583,10 +630,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 +858,6 @@ case class SortMergeJoinExec( |while ($streamedInput.hasNext()) { | $findNextJoinRows; | $beforeLoop - | boolean $exists = false; | | while (!$exists && $matchIterator.hasNext()) { | InternalRow $bufferedRow = (InternalRow) $matchIterator.next(); @@ -1299,11 +1375,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 +1393,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 +1417,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 +1452,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 +1472,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 +1483,13 @@ 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() + val r = advanceBufferUntilRestConditionSatisfied() || advanceStream() if (r) numOutputRows += 1 r } 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 428d29f2989a2..15558e9b790aa 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,35 @@ class ExistenceJoinSuite extends SharedSparkSession { } } + // Condition: a = c (equi-key) AND b < 3.0 (left-only) AND d < 4.0 (right-only) + private lazy val leftOnlyResidualCondition = { + 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))) + } + + 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 +225,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 +249,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 +270,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 +290,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 +451,65 @@ 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 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: 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 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))) + } From fd4f4482a85b12e19bc326cd31e5b0eb6de7ba59 Mon Sep 17 00:00:00 2001 From: James Xu Date: Fri, 10 Jul 2026 07:09:02 +0800 Subject: [PATCH 2/2] [SPARK-58066][SQL] Fix SMJ outer iterator for split streamed condition 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. --- .../spark/sql/execution/joins/SortMergeJoinExec.scala | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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 d46c577c8d3ad..fd62808e5fcb7 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 @@ -1489,7 +1489,12 @@ private abstract class OneSideOuterIterator( } override def advanceNext(): Boolean = { - val r = advanceBufferUntilRestConditionSatisfied() || 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 }