cloud-fan commented on code in PR #60029:
URL: https://github.com/apache/spark/pull/60029#discussion_r4446872445
##########
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:
**Blocking (P1):** `Predicate.create` compiles the residual with
subexpression elimination, which is on by default. With
`spark.sql.subexpressionElimination.skipForShortcutExpr` at its default of
false, `EquivalentExpressions` treats every AND/OR operand as always evaluated.
A non-leaf subexpression that appears twice in later conjuncts is therefore
hoisted, and the generated `eval()` runs it before the guard conjunct. The old
interpreted `And.eval`/`Or.eval` never reached those operands once the guard
decided the result.
For example, with ANSI mode on (the default):
```sql
SELECT * FROM l ASOF JOIN r MATCH_CONDITION (l.t >= r.t)
ON l.k = r.k AND l.y <> r.y
AND l.x / (l.y - r.y) > 1 AND l.x / (l.y - r.y) < 10
```
On the base branch this returns rows. With this change it fails with
`DIVIDE_BY_ZERO` as soon as a candidate pair has `l.y = r.y`, because the
hoisted division runs before `l.y <> r.y`. A LEFT ASOF JOIN whose left-only
guard stays in the residual fails the same way, for example `l.kind = 'num' AND
CAST(r.v AS INT) >= l.lo AND CAST(r.v AS INT) <= l.hi` with
`CAST_INVALID_INPUT`. The same happens for any guarded expression that can
throw (ANSI overflow, `element_at`, `raise_error`). When the repeated
expression is expensive (regex, JSON, a UDF) and the guard is mostly false, the
check also does more work per scanned right row than before. `SQLQuerySuite`'s
SPARK-58211 test enables `skipForShortcutExpr` for exactly this shape.
This affects the default generated predicate only. The interpreted fallback
still short-circuits, because its subexpression proxies are evaluated lazily.
The as-of condition on line 175 is also fine: the operands it could hoist are
the as-of keys, which are evaluated anyway.
**Recommended change:** Build the residual ON condition as a generated
predicate with subexpression elimination turned off, keeping the interpreted
fallback and the codegen factory-mode handling that Predicate.create applies.
To reuse that handling rather than copy it, one option is a small
CodeGeneratorWithInterpretedFallback[Expression, BasePredicate] next to the
scanner whose generated object comes from GeneratePredicate.generate(expr,
useSubexprElimination = false) and whose interpreted object is
InterpretedPredicate, applied to the residual bound to the joined output. Keep
Predicate.create for the as-of condition. Update the comment above the two
conditions to say why the residual skips subexpression elimination. Add ASOF
join tests whose residual repeats an expression that can fail behind an AND
guard and behind an OR guard, run in ANSI mode with code generation and with
interpreted evaluation, plus a scanner-level check that an expression repeated
behind a FALSE guard is no
t evaluated.
**Why this works:** Without subexpression elimination, Spark's generated
code for AND and OR evaluates the right operand only inside the branch taken
when the left operand does not decide the result. The compiled residual
therefore evaluates exactly the operands the base branch's interpreted check
reached. It still reads values through primitive generated code, so the tree
walking and boxing that the PR removes stay removed. The interpreted fallback
already short-circuits, because it evaluates its shared subexpressions lazily.
**Scope:** How the sort-merge ASOF join scanner compiles its residual ON
condition, plus regression coverage for residuals guarded by AND and OR in the
ASOF join tests.
**Compatibility:** The as-of condition stays compiled with Predicate.create,
residual checks still run as generated code without per-row boxing, output rows
stay uncopied, and a NULL condition result still counts as no match.
**Risks:** A residual that repeats an expression in positions that are
always evaluated loses the reuse subexpression elimination gave it in this PR,
so the expression runs once per occurrence, as on the base branch. The change
can be rolled back by compiling the residual with Predicate.create again.
**Constraints:** Keep Predicate.create's fallback behavior for the residual:
a failed compilation still falls back to interpreted evaluation, and the
interpreted-only factory mode still selects it. Inherit this from
CodeGeneratorWithInterpretedFallback.createObject instead of copying its
factory-mode switch into the scanner. Leave the as-of condition on
Predicate.create, since every operand it can hoist is evaluated anyway.
**Exclusions:**
- Changing Spark's default for
spark.sql.subexpressionElimination.skipForShortcutExpr, or the same eager
hoisting in other operators that compile conditions with Predicate.create, such
as non-codegen filters and Cartesian or nested-loop join conditions, is out of
scope: that behavior predates this PR and affects every operator.
**Success:** A sort-merge ASOF join whose residual repeats an expression
behind an AND guard, such as l.y <> r.y AND l.x / (l.y - r.y) > 1 AND l.x /
(l.y - r.y) < 10, returns the same rows as on the base branch in ANSI mode
instead of failing with DIVIDE_BY_ZERO. The same holds for an OR guard and for
a left-only guard in a LEFT ASOF JOIN, such as a repeated cast of a string
column behind l.kind = 'num', with code generation and with interpreted
evaluation. An expression repeated behind a guard that is FALSE for a candidate
pair is not evaluated for that pair. Residual checks still run as generated
code, give the same TRUE, FALSE, and NULL outcomes, and the as-of condition is
evaluated as in this PR.
**Verification:**
- **Regression:** An inner sort-merge ASOF join in ANSI mode whose residual
repeats a division behind a guard that references both sides, over data where
some candidate pair passes the as-of condition, fails the guard, and has a zero
divisor, returns the expected rows; before the repair it fails with
DIVIDE_BY_ZERO.
- **Regression:** A LEFT ASOF JOIN in ANSI mode whose residual repeats a
string-to-integer cast behind a left-only guard, over non-numeric right values
paired with left rows that fail the guard, returns the expected rows, including
null-padded left rows, instead of failing with CAST_INVALID_INPUT.
- **Behavior:** A residual guarded by OR, where the guard is TRUE on pairs
for which the repeated expression would fail, returns the expected rows in ANSI
mode.
- **Behavior:** A scanner-level check with a counting expression that occurs
twice behind a guard shows the expression is evaluated only for candidate pairs
that pass the guard.
- **Compatibility:** The guarded-residual queries return the same rows with
the default codegen factory mode and with interpreted-only evaluation, and with
subexpression elimination enabled and disabled.
- **Compatibility:** Existing sort-merge ASOF results stay unchanged,
including NULL condition results, tolerance bounds, nearest matching, and left
outer null padding.
- **Inspection:** The residual is compiled to generated code with
subexpression elimination off and the interpreted fallback kept, while the
as-of condition still goes through Predicate.create.
--
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]
