cloud-fan commented on code in PR #60029:
URL: https://github.com/apache/spark/pull/60029#discussion_r4473971697
##########
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)
+ // The residual can guard a part that throws, as in `y <> 0 AND x / y > 1`.
Review Comment:
**Nit (P3):** This example has no repeated part, so it does not show the
hazard `ShortCircuitPredicate` prevents. Subexpression elimination only hoists
a part used more than once, so `Predicate.create` would keep `x / y` behind `y
<> 0` here. The hazard needs the guarded part to repeat, as in `y <> 0 AND x /
y > 1 AND x / y < 10`, which is the shape the new tests use. Using that
example, and saying the hazard is about repeated parts, keeps a reader from
concluding the special factory is unnecessary.
This extends an [existing
discussion](https://github.com/apache/spark/pull/60029#discussion_r4446872445).
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** Both compiled conditions are built eagerly in the
scanner constructor, which runs for every task, but only the scan loops over a
non-empty right group read them. A task whose left partition is empty, or whose
left keys match no right group, therefore generates two predicates, and on a
fresh executor compiles them, without ever evaluating either; the base branch
only bound the conditions there. For example, with AQE off and 200 shuffle
partitions over a handful of keys, most tasks pay this for nothing. Declaring
both as `lazy val`s, as `nearestKeys` already is a few lines below, removes the
cost without changing per-row work, and `HashJoin` and
`BroadcastNestedLoopJoinExec` hold their compiled conditions the same way. If
the per-row check should not test the lazy initialization, a scan can copy them
into locals once it knows its right group is non-empty.
**Verification:**
- **Behavior:** A scanner-level test shows that a scanner over an empty left
input, and one whose left keys match no right group, never builds the as-of
predicate, for example with an as-of condition whose code generation fails
under the codegen-only factory mode, while a scanner that reaches a non-empty
right group does build it.
- **Compatibility:** Existing sort-merge ASOF results stay unchanged across
codegen factory modes, including backward, forward, nearest, tolerance,
residual, and left outer cases.
- **Inspection:** Both compiled conditions are lazy and are first read only
when a scan checks a right row.
##########
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:
**Blocking (P1):** The as-of condition still goes through
`Predicate.create`, so subexpression elimination hoists any part that a
MATCH_CONDITION operand repeats behind a guard. The comment above this line
assumes the only repeated parts are the match keys, but an operand can be any
deterministic expression. With
`spark.sql.subexpressionElimination.skipForShortcutExpr` at its default of
false, `EquivalentExpressions` treats every AND/OR operand inside an `IF`
predicate or the first `CASE WHEN` condition as always evaluated, and the
generated `eval()` runs every common subexpression before the predicate body.
The base branch evaluated this condition interpreted, where `If` and `And` stop
at the guard.
For example, in ANSI mode with `spark.sql.join.asofJoin.enabled=true`:
```sql
SELECT t.id, r.v
FROM VALUES (1, 1, 6, 0, 10, 20) AS t(id, k, x, y, a, b)
ASOF JOIN VALUES (1, 15, 'm') AS r(k, ts, v)
MATCH_CONDITION (IF(t.y <> 0 AND t.x / t.y > 1 AND t.x / t.y < 5, t.a,
t.b) >= r.ts)
ON t.k = r.k
```
Reading the code, the base branch takes the `IF`'s false branch and returns
`(1, 'm')`, while this change computes `t.x / t.y` before the guard and fails
with `DIVIDE_BY_ZERO`. The left child sort does not fail first on the base
branch: with an equi-key its prefix covers only `t.k`, and its comparator
generates the operand without subexpression elimination. Any guarded part that
can throw (ANSI division or overflow, casts, `element_at`, `raise_error`) fails
the same way, and an expensive repeated part now runs for every scanned right
row. This affects joins with at least one equi-key, whether the condition comes
from a SQL MATCH_CONDITION or from a DataFrame `joinAsOf` key on the sort-merge
path; without equi-keys, the left sort's prefix already hoists the same part on
the base branch. I said in the last round that the as-of condition could stay
on `Predicate.create`; that was wrong.
This extends an [existing
discussion](https://github.com/apache/spark/pull/60029#discussion_r4446872445).
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -161,17 +161,20 @@ 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.
+ private val boundAsOfCond = Predicate.create(asOfCondition, joinedOutput)
private val boundResidualCond =
- residualCondition.map(bindReference(_, joinedOutput))
+ residualCondition.map(Predicate.create(_, joinedOutput))
Review Comment:
Thanks, the residual fix looks right: it now compiles without subexpression
elimination, so AND and OR keep short-circuiting, and the new tests cover the
AND, OR, and left-only guards with compiled and interpreted conditions. I was
wrong above that the as-of condition is safe on Predicate.create. This review
raises a new issue about subexpression elimination in the as-of condition, and
it also suggests a clearer example for the residual comment.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4446872445","thread_id":"inline:4446872445","verdict_sha256":"bd10262b8431ee4aa25a4e87c3445476b4010afc03821e5422f9dc0fc1397fea"}
-->
--
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]
