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..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 @@ -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,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/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))) + }