LuciferYang commented on PR #58045:
URL: https://github.com/apache/spark/pull/58045#issuecomment-5324037877
Thanks for the review. I fixed the `defs.exists` problem you flagged, and I
looked into the lazy-memoization suggestion. Reporting what I found on the
second one, because I do not think I should decide its scope alone.
### The `defs.exists` hoisting is real, fixed in 01f629aa0d3
You were right, and my first attempt to reproduce it was wrong: putting the
`With` *outside* the `CaseWhen` goes down the main rewrite, which hoists
everything anyway. With the `With` inside the ELSE branch (the shape `Between`
actually produces) and two definitions, `rand(1)` plus `randstr(-1, 0)`, the
unsafe sibling did get pre-evaluated. On bb3296963f2:
```
Project [CASE WHEN (a#0 > 0) THEN true ELSE (((_common_expr_0#3 >= 0.4) AND
(_common_expr_0#3 <= 0.6)) AND ((_common_expr_1#4 >= a) AND (_common_expr_1#4
<= z))) END AS col#2]
+- Project [a#0, b#1, rand(1) AS _common_expr_0#3, randstr(-1, 0, false) AS
_common_expr_1#4]
+- LocalRelation <empty>, [a#0, b#1]
```
`randstr(-1, 0)` then raises `INVALID_PARAMETER_VALUE.LENGTH` on rows whose
branch is never taken. The follow-up moves the decision to each definition, so
`rand()` is pre-evaluated while the `randstr` sibling keeps its inlining:
```
Project [CASE WHEN (a#0 > 0) THEN true ELSE (((_common_expr_0#3 >= 0.4) AND
(_common_expr_0#3 <= 0.6)) AND ((randstr(-1, 0, false) >= a) AND (randstr(-1,
0, false) <= z))) END AS col#2]
+- Project [a#0, b#1, rand(1) AS _common_expr_0#3]
+- LocalRelation <empty>, [a#0, b#1]
```
A test covers this shape. It also let me delete the recursion back into the
main branch, so the two paths now read more alike. (The alias names above are
index-based; with `spark.sql.useCommonExprIdForAlias` at its default of true
you would see the common-expression id instead.)
### On branch-local lazy memoization: no mechanism exists today, and I would
rather you decide the scope
I checked whether this could be built on the existing
subexpression-elimination machinery. It cannot, in either execution mode, and
the blocker is not that CSE avoids branches — it does look inside them — but
that it never admits a nondeterministic expression and never evaluates lazily.
- **CSE does reach into conditional branches.** `childrenToRecurse` takes
only `alwaysEvaluatedInputs` for a `ConditionalExpression`
(EquivalentExpressions.scala:160), but `commonChildrenToRecurse` additionally
recurses into `branchGroups` (line 169), and `updateCommonExprs` (lines
103-130) hoists whatever is common to *every* branch of a group —
`If.branchGroups` is `Seq(Seq(trueValue, falseValue))`
(conditionalExpressions.scala:72). So the machinery already reasons about
branches; it admits expressions that are guaranteed to be evaluated whichever
branch runs.
- **But it excludes nondeterministic expressions entirely.** Every insertion
goes through `updateExprInMap`, which is gated on `if (expr.deterministic)`
(EquivalentExpressions.scala:65). `rand`/`randn` never become common
subexpressions. The nondeterministic leaves — `uuid`,
`monotonically_increasing_id`, `spark_partition_id`, `input_file_name` — are
excluded twice over, because `updateExprTree` also skips every `LeafExpression`
regardless of determinism (line 198).
- **And where it does apply, it is eager, not lazy.** Interpreted evaluation
has genuine per-row memoization: `SubExprEvaluationRuntime` invalidates its
cache in `setInput` (SubExprEvaluationRuntime.scala:67-70), which every caller
invokes once per row (`InterpretedPredicate.eval`,
`InterpretedUnsafeProjection.apply`, `InterpretedMutableProjection.apply`,
`InterpretedSafeProjection.apply`), and `ExpressionProxy.eval` loads through it
(line 135). But proxies come only from that same gated map (`proxyExpressions`,
line 88). Codegen emits one `subExpr` function per common expression and calls
them all up front (CodeGenerator.scala:1322-1367, `subexprFunctionsCode` at
GeneratePredicate.scala:42-63). `SubExprEliminationState.children` (line 81)
orders dependencies, it does not defer them. Whole-stage codegen has its own
path, `subexpressionEliminationForWholeStageCodegen` (line 1193), which lets
the operator choose where to place the evaluation instead of emitting it all at
once — bu
t it builds the same `EquivalentExpressions` (lines 1195-1196), so the same
gate applies, and the placement is still a fixed point rather than
per-reference memoization.
A runtime check, hand-duplicating the expression inside the branch so no
`With` is involved at all:
```sql
SELECT CASE WHEN id < 0 THEN false
ELSE monotonically_increasing_id() >= 3
AND monotonically_increasing_id() <= 5 END
FROM range(0, 10, 1, 1)
```
returns 6 true rows instead of 3, with
`spark.sql.subexpressionElimination.enabled` both true and false. To be precise
about what this does and does not show: it demonstrates that no existing
machinery memoizes a repeated nondeterministic expression in a branch, but
because `monotonically_increasing_id` is a leaf it cannot isolate the
determinism gate from the leaf skip.
Repeated evaluation is of course not a bug when the user writes the
duplicates: `SELECT rand(), rand()` is meant to give two draws. The bug is
specific to `With`, which promises the opposite — "The common expressions are
guaranteed to be evaluated only once even if it's referenced more than once"
(With.scala:26-27) — and inlining silently downgraded it to the repeated-call
behavior in a place the user never wrote a duplicate, because `Between` put it
there.
So lazy memoization needs a new evaluable expression with per-row caching
plus codegen support, and `With`/`CommonExpressionDef`/`CommonExpressionRef`
are all `Unevaluable` today (With.scala:30,154,165), which is why the rewrite
has to eliminate them before execution.
That is a bigger change than this bug fix, and it would subsume the eager
projection here rather than sit beside it. But I do not want to declare it out
of scope unilaterally: if you would rather this PR wait for the real mechanism
than ship a partial fix that trades RNG advancement on untaken rows, I am happy
to do it that way. Which would you prefer?
For the record, the RNG advancement you named is real and the PR description
calls it out: the pre-evaluated generator now advances on untaken rows, so
`monotonically_increasing_id()` values and `rand()` draws shift relative to the
current behavior. My argument for accepting it is that the current behavior is
already wrong, and that outside a conditional branch the main rewrite already
hoists a multi-referenced, non-cheap definition into a project without any
determinism check. Where it cannot — e.g. the `childPlanIndex == -1`
force-inline — the file already carries a TODO saying exactly this: "things can
go wrong if the common expression is nondeterministic. We don't fix it for now
to match the old buggy behavior when certain `RuntimeReplaceable` did not use
the `With` expression" (RewriteWithExpression.scala:198-200). So
inline-plus-nondeterministic is already documented upstream as known-buggy
legacy behavior; this PR removes one class of it rather than introducing a new
po
licy.
--
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]
