cloud-fan commented on code in PR #57162:
URL: https://github.com/apache/spark/pull/57162#discussion_r3596074874
##########
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 = {
Review Comment:
`rightOnlyResidualCondition` is defined but never referenced (only
`leftOnlyResidualCondition` is used). Either wire it into a test — a
right-only-residual case would add useful coverage — or remove it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala:
##########
@@ -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))
Review Comment:
This whole `streamedOnlyCondition`/`restCondition` split — the
`splitConjunctivePredicates` + `partition(subsetOf(streamedPlan.outputSet))`
and the `joinType match { LeftAnti | LeftOuter | RightOuter | _: ExistenceJoin
}` gate — is duplicated verbatim in `SortMergeJoinExec`, and both classes newly
mix in `PredicateHelper` for it. A later change to the supported-join-type set
(or the split rule) has to touch both copies in lockstep; missing one silently
mis-applies the optimization for one join family. Consider hoisting this into a
shared trait or helper.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala:
##########
@@ -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;
Review Comment:
This guard (and the equivalent ones in `codegenAnti` and `codegenExistence`)
ends with a bare `return;`. But `HashJoin` is a codegen *consumer* —
`doProduce` just delegates to `streamedPlan.produce(ctx, this)` — so this
`doConsume` code is inlined into the streamed producer's row loop unless
`WholeStageCodegenExec.consume` wraps it in a separate `doConsume_N` function.
That wrapping requires `requireAllOutput =
output.forall(parent.usedInputs.contains)`, and `usedInputs` defaults to
`references` (join keys + condition attrs), so any streamed column not
referenced by the keys/condition makes it false — e.g. the `SELECT t1.* ... ON
t1.id = t2.id AND udf(t1.a) = ...` shape from the PR description. When not
wrapped, the `return;` exits the producer's `processNext()`; for a batching
producer (`ColumnarToRowExec` vectorized scan, `RangeExec`) the loop cursor is
only advanced after the inner loop, so the next `processNext()` reprocesses the
same batch → duplicate rows and/or non-
termination.
`MergeRowsExec.doConsume` handles exactly this by always wrapping its body
in a function ("so that `return` statements exit this function instead of the
outer produce loop"). Suggest doing the same here, or restructuring the guard
to fall through (e.g. `do { } while(false)` + a flag) instead of `return`. The
current tests don't catch it: they use `parallelize` (an RDD → non-batching
`InputAdapter`) and conditions where every streamed column is referenced, so
the wrapper is always inserted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
