Yang Jie created SPARK-58902:
--------------------------------
Summary: Evaluate a multi-referenced common expression lazily
instead of pre-evaluating it in a Project
Key: SPARK-58902
URL: https://issues.apache.org/jira/browse/SPARK-58902
Project: Spark
Issue Type: Improvement
Components: SQL
Affects Versions: 5.0.0
Reporter: Yang Jie
`With` promises a common expression is evaluated only once even when referenced
more
than once (see the scaladoc on `With`). `RewriteWithExpression` keeps that
promise by
hoisting the definition into a `Project`, which works because a `Project` is
evaluated
for every row. Inside a conditional branch that is not true: the branch may not
be
evaluated at all, so the rule inlines instead, and inlining hands each
reference its own
evaluation.
For a nondeterministic definition that is wrong, not just wasteful. SPARK-58818
fixed the
case that is reachable through the generators which cannot raise, by
pre-evaluating them
in the child `Project` anyway. That fix is bounded by an allowlist
(`canPreEvaluateInBranch`): pre-evaluation happens on every row, so an
expression that can
raise would turn a wrong result into a spurious error. Everything the allowlist
turns down
keeps the old inlining and the old wrong result, which the code says in as many
words.
h3. What is still wrong
{code:sql}
-- randstr is referenced twice by BETWEEN and inlined, so the two comparisons
see
-- two different strings
SELECT CASE WHEN a < 0 THEN false
ELSE randstr(3, 0) BETWEEN 'a' AND 'b' END
FROM t
{code}
The same shape with {{nullif}}, and the same for {{uniform(lo, hi)}},
{{shuffle(arr)}},
{{reflect(...)}}, {{rand() / col}}, {{cast(rand() as int)}} -- anything
nondeterministic
that is not one of the six leaf generators on the allowlist. Note the allowlist
approximates "cannot raise" by syntactic shape, so {{randstr(3, 0)}} is turned
down for
being a {{BinaryLike}} even though only {{randstr(-1, 0)}} raises.
Outside a conditional branch these are correct, because the main rewrite hoists
any
multi-referenced non-cheap definition without consulting the allowlist. The
wrong results
are specific to a definition that is both inside a branch and turned down by
the allowlist.
There is a second, older instance of the same root cause. When a join
condition's common
expression references columns from both sides, no single child plan can hold
the column,
so `RewriteWithExpression` force-inlines it. The rule carries a TODO admitting
this goes
wrong for a nondeterministic definition and is kept only to match the old buggy
behavior.
h3. What lazy memoization would give
An evaluable expression that caches its value per row, so a reference reads the
value the
first time it is reached and reuses it after. Then:
* The allowlist can go. Evaluation happens only where the original expression
would have
evaluated, so an expression that can raise raises on exactly the rows it did
before, and
the wrong results above are fixed as one class rather than one generator at a
time.
* The evaluation domain is per reference reached, not per branch reached. A
reference
behind a short-circuiting operator (`a > 0 AND rand() BETWEEN 0.4 AND 0.6`)
or inside a
nested conditional is not read on every row of the branch; pre-evaluation
cannot express
that, memoization does.
* The join TODO goes away, since the definition no longer has to be placed in a
child
plan.
* The supporting machinery in `RewriteWithExpression` -- child projects, column
naming,
the per-child registry, projecting the extra columns away, the interaction
with
`CollapseProject` -- is no longer needed for these cases.
h3. Why the existing machinery does not cover it
Subexpression elimination cannot be reused.
`EquivalentExpressions.updateExprInMap` is
gated on `expr.deterministic`, so `rand`/`randn` never become common
subexpressions, and
`updateExprTree` additionally skips every `LeafExpression`, which excludes
`uuid`,
`monotonically_increasing_id`, `spark_partition_id` and `input_file_name` twice
over.
Where it does apply it is eager rather than lazy: codegen emits one `subExpr`
function per
common expression and calls them all up front, and
`SubExprEliminationState.children` only
orders dependencies. Interpreted evaluation does have per-row memoization in
`SubExprEvaluationRuntime`, but its proxies come from the same gated map.
So this needs a new expression with per-row caching plus codegen support.
`With`, `CommonExpressionDef` and `CommonExpressionRef` are all `Unevaluable`
today, which
is why `RewriteWithExpression` has to eliminate them before execution.
h3. A negative result worth recording
SPARK-58818 also tried the closest thing reachable in the plan: keep the eager
pre-evaluation but wrap the column in `If(branch is reached, definition,
default)`, so a
stateful generator is only advanced on the rows that reach the branch. It does
not hold up.
A guard in a plain `Project` is outside conditional evaluation, so subexpression
elimination hoists a part repeated across guards to the top of the projection
and evaluates
it eagerly (`spark.sql.subexpressionElimination.skipForShortcutExpr` is false
by default),
and a condition that can raise then raises on rows the branch never reached:
{code:sql}
SELECT CASE WHEN a = 0 THEN false
WHEN 6 / a > 2 THEN rand(1) BETWEEN 0 AND 1
WHEN 6 / a < -2 THEN rand(2) BETWEEN 0 AND 1
ELSE false END
FROM t -- t contains a = 0
{code}
gives DIVIDE_BY_ZERO with the frame in `project_subExpr_0$`. Restricting guards
to nodes
that cannot raise fixes that but leaves only plain column comparisons: `a + b >
0`,
`abs(a) > 0`, `upper(s) = 'X'`, `cast(a as long) > 0` all lose the guard,
silently. The
guard is a dead end; memoization is not affected by any of this, since it does
not put the
condition anywhere.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
