cloud-fan commented on code in PR #60029:
URL: https://github.com/apache/spark/pull/60029#discussion_r4487098856
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -161,17 +162,22 @@ private[joins] class SortMergeAsOfJoinScanner(
private val joinedOutput = leftOutput ++ rightOutput
private val joinedRow = new JoinedRow()
// Use nullable bound references so outer-join null padding is safe even when
- // right attributes are NOT NULL in the catalog schema.
+ // right attributes are NOT NULL in the catalog schema. Every output row
reuses the
+ // projection's row.
private val resultProjection = {
val nullableRefs = joinedOutput.zipWithIndex.map { case (attr, i) =>
BoundReference(i, attr.dataType, nullable = true)
}
UnsafeProjection.create(nullableRefs, joinedOutput)
}
- private val boundAsOfCond = bindReference(asOfCondition, joinedOutput)
- private val boundResidualCond =
- residualCondition.map(bindReference(_, joinedOutput))
+ // Compiled, so a check runs generated code instead of walking the
expression tree. Both skip
Review Comment:
**Nit (P3):** These comments describe only the generated predicate.
`ShortCircuitPredicate` inherits `createObject`, which returns
`InterpretedPredicate` under the interpreted-only factory mode that the new
tests run, and after a failed compile in the default mode. That object walks
the expression tree. With `spark.sql.subexpressionElimination.enabled` on, the
default, it does apply subexpression elimination: it wraps repeated parts in
`ExpressionProxy`, and it still short-circuits only because each proxy is
evaluated lazily, on first access. So "Compiled" and "Both skip subexpression
elimination" here, and "Compiles a predicate without subexpression elimination"
on `ShortCircuitPredicate`, hold only for the generated path. Separately, "a
task that checks no right row compiles nothing" overstates: such a task still
compiles the scanner's `UnsafeProjection`s, and its key ordering when there are
equi-keys. Only the two conditions are lazy.
Suggested wording: the generated predicates skip subexpression elimination;
the interpreted fallback keeps it but evaluates a shared part only when an
operand reaches it, so AND, OR, and IF still short-circuit; and a task that
checks no right row compiles neither condition. The current `y <> 0 AND x / y >
1 AND x / y < 10` example can stay.
**Verification:**
- **Inspection:** The comments on the two scanner conditions and on the
predicate factory state that the generated predicates skip subexpression
elimination, that the interpreted fallback keeps it but evaluates a shared part
only when an operand reaches it, and that a task which checks no right row
builds neither condition, and the edit touches only comments.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -161,17 +162,22 @@ private[joins] class SortMergeAsOfJoinScanner(
private val joinedOutput = leftOutput ++ rightOutput
private val joinedRow = new JoinedRow()
// Use nullable bound references so outer-join null padding is safe even when
- // right attributes are NOT NULL in the catalog schema.
+ // right attributes are NOT NULL in the catalog schema. Every output row
reuses the
+ // projection's row.
private val resultProjection = {
val nullableRefs = joinedOutput.zipWithIndex.map { case (attr, i) =>
BoundReference(i, attr.dataType, nullable = true)
}
UnsafeProjection.create(nullableRefs, joinedOutput)
}
- private val boundAsOfCond = bindReference(asOfCondition, joinedOutput)
+ // Compiled once, so a check runs generated code instead of walking the
expression tree.
+ // The scan checks the as-of condition for every right row it reads. Its
repeated parts are
+ // the match keys, which its first comparison always computes, so sharing
them is safe.
+ private val boundAsOfCond = Predicate.create(asOfCondition, joinedOutput)
Review Comment:
Thanks, this looks right. The as-of condition is now built with
`ShortCircuitPredicate` as well, so a part that a MATCH_CONDITION operand
repeats behind a guard is no longer evaluated ahead of the guard. The new
IF-guard SQL test runs with compiled and interpreted conditions and with
subexpression elimination on and off, and the new scanner test shows the
repeated part never runs behind a FALSE guard.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4473971655","thread_id":"inline:4473971655","verdict_sha256":"85bf1c8d1fd8d643a81103b117280af1185e50dad5ccd873de5bf6de3bfc5b34"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -161,17 +162,22 @@ private[joins] class SortMergeAsOfJoinScanner(
private val joinedOutput = leftOutput ++ rightOutput
private val joinedRow = new JoinedRow()
// Use nullable bound references so outer-join null padding is safe even when
- // right attributes are NOT NULL in the catalog schema.
+ // right attributes are NOT NULL in the catalog schema. Every output row
reuses the
+ // projection's row.
private val resultProjection = {
val nullableRefs = joinedOutput.zipWithIndex.map { case (attr, i) =>
BoundReference(i, attr.dataType, nullable = true)
}
UnsafeProjection.create(nullableRefs, joinedOutput)
}
- private val boundAsOfCond = bindReference(asOfCondition, joinedOutput)
+ // Compiled once, so a check runs generated code instead of walking the
expression tree.
+ // The scan checks the as-of condition for every right row it reads. Its
repeated parts are
+ // the match keys, which its first comparison always computes, so sharing
them is safe.
+ private val boundAsOfCond = Predicate.create(asOfCondition, joinedOutput)
Review Comment:
Thanks, both conditions are now lazy, so a task whose scan never checks a
right row builds neither of them. The new scanner test confirms this for an
empty left side and an empty right side, using a condition that fails to
compile.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4473971684","thread_id":"inline:4473971684","verdict_sha256":"85bf1c8d1fd8d643a81103b117280af1185e50dad5ccd873de5bf6de3bfc5b34"}
-->
--
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]
