spark-test commented on code in PR #59984:
URL: https://github.com/apache/spark/pull/59984#discussion_r4465530255
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -2330,10 +2332,64 @@ class CodegenContext extends Logging {
// We need to make sure that we do not reuse stateful expressions. This is
needed for codegen
// as well because some expressions may implement `CodegenFallback`.
val cleanedExpressions =
expressions.map(_.freshCopyIfContainsStatefulExpression())
- if (doSubexpressionElimination)
subexpressionElimination(cleanedExpressions)
+ if (doSubexpressionElimination) {
+ subexprFunctions ++=
subexpressionElimination(cleanedExpressions).map(_._2)
+ }
cleanedExpressions.map(e => e.genCode(this))
}
+ /**
+ * Returns the index of the first of `expressions` that references the
common subexpression
+ * `subexpr`, or -1. When the expressions are evaluated in order and one of
them can stop the
+ * evaluation, such as a FALSE conjunct of a predicate, the subexpression is
evaluated right
+ * before that expression instead of before all of them (SPARK-56032,
SPARK-60332). An
+ * occurrence in a conditionally evaluated part of an expression, such as a
branch of CASE WHEN,
+ * counts as a reference, so the subexpression is still evaluated before
such an expression.
+ */
+ def firstReferencingExpression(expressions: Seq[Expression], subexpr:
ExpressionEquals): Int = {
+ expressions.indexWhere(_.exists(e => ExpressionEquals(e) == subexpr))
+ }
+
+ /**
+ * Generates code for expressions with subexpression elimination, as
`generateExpressions`
+ * does, but instead of adding the evaluation of the common subexpressions to
+ * `subexprFunctionsCode`, to be put before all the expressions, returns it
per expression: the
+ * code that evaluates the subexpressions the expression is the first to
reference, to be put
+ * right before the code of the expression. The expressions are assumed to
be evaluated in the
+ * `expressions` order, so a subexpression is still evaluated before the
expressions that
+ * reference it, but not before an expression that stops the evaluation,
such as a FALSE
+ * conjunct of a predicate before the ones that share the subexpression
(SPARK-60332).
+ *
+ * Returns None, and generates nothing, when the expressions have no common
subexpression. The
+ * caller can then generate them without subexpression elimination. This
must be the only
+ * subexpression elimination on this context, so that the common
subexpressions are all found
+ * in `expressions`.
+ */
+ def generateExpressionsWithSubexprCodes(
+ expressions: Seq[Expression]): Option[Seq[(String, ExprCode)]] = {
+ val cleanedExpressions =
expressions.map(_.freshCopyIfContainsStatefulExpression())
+ val subexprCodes = subexpressionElimination(cleanedExpressions)
+ if (subexprCodes.isEmpty) {
+ None
+ } else {
+ val codes = cleanedExpressions.map(e => e.genCode(this))
+ val firstUses = subexprCodes.map { case (subexpr, code) =>
+ val first = firstReferencingExpression(cleanedExpressions,
ExpressionEquals(subexpr))
Review Comment:
**Blocking (P1):** This places each common subexpression's call before the
first conjunct that semantically references it, independently of the other
subexpressions, but the function of a subexpression that contains another one
reads that one's state instead of computing it. Under commutative
canonicalization an earlier conjunct can hold an equal copy of the outer
subexpression that contains no copy of the inner one, in a position the
analysis does not count. For example, with default configs the optimizer
filters `SELECT * FROM VALUES (3, 3, 1, 1, 0) AS t(a, b, c, d, x) WHERE IF(x >
0, a + c + b, d) > 0 AND a + b + c > 1 AND a + b + c < 10 AND a + b > 5` on the
driver through this generated predicate. `(a + c) + b` in the IF branch matches
`(a + b) + c`, so that call is emitted before conjunct 0, while `a + b` is
emitted only before conjunct 1. `(a + b) + c` is then computed from the
initial, or the previous row's, `a + b`, evaluates to 1, and `1 > 1` drops the
row although every
conjunct is TRUE. Multiply, And, Or and the bitwise operators canonicalize
the same way, and filters outside whole-stage codegen, non-codegen join
conditions and the other Predicate.create callers are affected too; the
whole-stage FilterExec branch emits a state's children first and is not.
Emitting every not-yet-emitted dependency of a call before the call, as
evaluateSubExprEliminationState does for whole-stage states, keeps the
per-conjunct placement and fixes the order.
**Verification:**
- **Regression:** A generated predicate whose earlier conjunct holds, only
in an IF or CASE WHEN branch, a commutatively reordered copy of a nested common
subexpression returns the same result as interpreted evaluation, both on a
first row and on rows that follow a row with different values.
- **Behavior:** Nested common subexpressions, whether first referenced by
the same conjunct or by different ones, still evaluate inner before outer, and
a FALSE earlier conjunct still skips both.
- **Compatibility:** A query that filters inline VALUES data with such a
predicate returns the same rows with and without the optimizer's local-relation
folding.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GeneratePredicate.scala:
##########
@@ -38,8 +38,49 @@ object GeneratePredicate extends CodeGenerator[Expression,
BasePredicate] {
val ctx = newCodeGenContext()
// Do sub-expression elimination for predicates.
- val eval = ctx.generateExpressions(Seq(predicate),
useSubexprElimination).head
- val evalSubexpr = ctx.subexprFunctionsCode
+ val conjuncts = splitConjunctivePredicates(predicate)
+ // With a subexpression common to several top-level conjuncts, or within
one of them, the
+ // conjuncts are generated one by one, and the subexpression is evaluated
just before the
+ // first conjunct that references it, not before the whole condition, so
that a FALSE
+ // conjunct before them stops the evaluation, as `And.eval` and the
whole-stage codegen of a
+ // filter do (SPARK-60332). The conjuncts are evaluated like `And.eval`: a
FALSE one stops
+ // the evaluation, a NULL one does not, and the condition is TRUE only if
all of them are.
+ // Otherwise the condition is generated as a whole, as before.
+ val conjunctCodes = if (useSubexprElimination && conjuncts.length > 1) {
+ ctx.generateExpressionsWithSubexprCodes(conjuncts)
Review Comment:
**Blocking (P1):** Analyzing each conjunct as its own root changes which
subexpressions are eliminated, not only where they are evaluated. Inside a
task, EquivalentExpressions rejects any input that contains a plan expression,
so on the base branch a filter such as `WHERE a > (SELECT min(k) FROM u) AND
IF(q, b > 0, y * 2 + 1 > 0) AND y * 2 + 1 > 5 AND y * 2 + 1 < 10` got no
elimination at all: on a row with q = true, b = 0 and y = 2147483647 the IF
conjunct is FALSE and the row is dropped. Now only the subquery conjunct is
skipped, `y * 2 + 1` becomes common, and because its first occurrence is in the
IF's ELSE branch, its call runs before the IF conjunct and raises
ARITHMETIC_OVERFLOW in ANSI mode. That needs a FilterExec outside whole-stage
codegen, for example `SELECT *` over more than spark.sql.codegen.maxFields
columns. With spark.sql.subexpressionElimination.skipForShortcutExpr=true the
base branch analyzed only the first conjunct, so filtering inline VALUES with
`(b = 1 OR
y + 1 > 0) AND a < 0 AND y + 1 > -1 AND y + 1 > -2` dropped a row with b = 1,
a = 0 and y = 2147483647, while now `y + 1` is evaluated before the first
conjunct and overflows. Both queries returned rows on the base branch through
this generated predicate. Where whole-stage codegen applies, FilterExec already
raises for them, so for those filters the change removes the difference by
making the generated-predicate path fail too.
**Recommended change:** Decide the eliminated set from the whole cleaned
predicate, as the base branch did, and only place it per conjunct. Add the
cleaned predicate to the subexpression analysis as one tree, and split that
same cleaned predicate into its conjuncts. Keep only the common subexpressions
that some conjunct references, dropping top-level AND nodes that no conjunct
contains, and generate and place only those. When none remain, generate the
predicate as a whole without elimination, as the change already does.
**Why this works:** The analysis's whole-tree rules then apply unchanged. A
predicate that holds a plan expression inside a task, or an
objects.LambdaVariable, gets no elimination. With
spark.sql.subexpressionElimination.skipForShortcutExpr, only the first
conjunct's subexpressions are eliminated. So subexpressions those rules kept
lazy are again evaluated only where the AND and conditional code reaches them.
For every other predicate, with skipForShortcutExpr off, the whole-tree and
per-conjunct analyses find the same subexpressions apart from AND groups of the
top-level chain, so the placement fix for the queries the PR targets is
unaffected.
**Scope:** Move the subexpression analysis back to the whole predicate while
keeping the per-conjunct placement.
**Compatibility:** With default configs, each shared subexpression is still
evaluated just before the first conjunct that references it, which is the PR's
fix.
**Risks:** A predicate in which an AND group of the top-level chain occurs
again, either repeated in a non-adjacent position or nested in another conjunct
such as under NOT, loses elimination of that group's conjuncts, because the
analysis counts the AND node instead of its conjuncts. Results are unchanged,
and the optimizer already folds adjacent duplicates. Inside a task, a predicate
with a plan-expression conjunct goes back to the base branch's generation
without elimination, giving up the new elimination of its other conjuncts.
**Constraints:** The conjuncts must be split from the same cleaned predicate
the analysis saw, so that stateful expressions keep a single fresh copy. The
dependency ordering of the precompute calls must be computed over the retained
subexpressions and their representatives.
**Exclusions:**
- The whole-stage FilterExec subexpression branch already analyzed each
conjunct separately on the base branch, and it still raises for these shapes
with whole-stage codegen on.
- Placing a precompute only before always-evaluated references, which the PR
tracks as SPARK-60774.
**Success:** With spark.sql.subexpressionElimination.skipForShortcutExpr set
to true, a subexpression shared by later conjuncts is not eliminated, and a row
that an earlier conjunct drops without reaching it is dropped without
evaluating it. A predicate generated inside a task that has a scalar or IN
subquery conjunct gets no subexpression elimination, as on the base branch.
With default configs, the reproducer, the `b < 0` variant and the join case
still drop the row before the shared expression overflows.
**Verification:**
- **Regression:** With whole-stage codegen off and ANSI on, consider a
filter whose earlier conjunct is an IF or CASE WHEN that references a
subexpression shared by later conjuncts only in an untaken branch. It returns
its rows instead of raising, both with skipForShortcutExpr set to true and with
a scalar-subquery conjunct evaluated inside the task.
- **Behavior:** With default configs, the filter reproducer and the
cartesian-product join case still drop the row before the shared expression
overflows.
- **Compatibility:** Generated predicates with no plan-expression conjunct,
and with skipForShortcutExpr off, still eliminate the same subexpressions as
before, apart from top-level AND nodes.
##########
docs/sql-migration-guide.md:
##########
@@ -28,6 +28,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
+- Since Spark 4.4, a predicate evaluated through generated code outside of
whole-stage codegen, such as a filter condition with whole-stage codegen
disabled or a join condition, evaluates a subexpression shared by several of
its top-level conjuncts just before the first conjunct that references it,
instead of before the whole condition. So a `FALSE` conjunct before them stops
the evaluation, as `AND` does when evaluated directly, and an error the
subexpression raises on the row, such as `ARITHMETIC_OVERFLOW` in ANSI mode, is
no longer raised. For example, `SELECT a FROM t WHERE b < 0 AND y + 1 > 0 AND y
+ 1 > -1` no longer fails on a row where `b` is `1` and `y` is `2147483647`. A
subexpression whose first occurrence is in a conditional branch of a conjunct,
such as a branch of `CASE WHEN`, is still evaluated before that conjunct.
Review Comment:
**Nit (P3):** This limits the change to a subexpression shared by several
top-level conjuncts, but a subexpression repeated inside one conjunct moves
too: GeneratePredicate takes the conjunct-wise path whenever the conjuncts have
any common subexpression. For example, with whole-stage codegen off and ANSI
mode on, `SELECT a FROM t WHERE b < 0 AND (y + 1) * (y + 1) > 0` failed with
ARITHMETIC_OVERFLOW on a row with `b = 1` and `y = 2147483647`, and now drops
the row at `b < 0`. Consider describing it as a subexpression common to the
top-level conjuncts, whether several of them share it or one of them repeats it.
**Verification:**
- **Inspection:** The 4.4 migration-guide entry names subexpressions
repeated within one conjunct alongside those shared by several conjuncts, and
its other statements still match the generator's behavior once the other
repairs are applied.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GeneratePredicate.scala:
##########
@@ -38,8 +38,49 @@ object GeneratePredicate extends CodeGenerator[Expression,
BasePredicate] {
val ctx = newCodeGenContext()
// Do sub-expression elimination for predicates.
- val eval = ctx.generateExpressions(Seq(predicate),
useSubexprElimination).head
- val evalSubexpr = ctx.subexprFunctionsCode
+ val conjuncts = splitConjunctivePredicates(predicate)
+ // With a subexpression common to several top-level conjuncts, or within
one of them, the
+ // conjuncts are generated one by one, and the subexpression is evaluated
just before the
+ // first conjunct that references it, not before the whole condition, so
that a FALSE
+ // conjunct before them stops the evaluation, as `And.eval` and the
whole-stage codegen of a
+ // filter do (SPARK-60332). The conjuncts are evaluated like `And.eval`: a
FALSE one stops
+ // the evaluation, a NULL one does not, and the condition is TRUE only if
all of them are.
+ // Otherwise the condition is generated as a whole, as before.
+ val conjunctCodes = if (useSubexprElimination && conjuncts.length > 1) {
+ ctx.generateExpressionsWithSubexprCodes(conjuncts)
+ } else {
+ None
+ }
+ val evalCode = conjunctCodes match {
+ case Some(codes) =>
+ val nullConjunct = ctx.freshName("nullConjunct")
+ val checks = codes.map { case (subexprs, ev) =>
+ s"""
+ |$subexprs
+ |${ev.code}
+ |if (${ev.isNull}) {
+ | $nullConjunct = true;
+ |} else if (!${ev.value}) {
+ | return false;
+ |}
+ """.stripMargin
+ }
+ s"""
+ |boolean $nullConjunct = false;
+ |${checks.mkString("\n")}
Review Comment:
**Non-blocking (P2):** Every conjunct's precompute calls, code and
null/FALSE check are concatenated into eval(). Each conjunct's own code is
still bounded by reduceCodeSize, and each conjunct's calls are split
separately, but nothing bounds the total, whereas on the base branch the
condition was one And tree whose large subtrees reduceCodeSize moved into their
own methods. So a predicate with many top-level conjuncts and any shared
subexpression, even a cheap one such as a struct field read twice, gets an
eval() that grows linearly with the conjunct count when it is evaluated outside
whole-stage codegen: a filter over a wide schema, a non-codegen join, or a
filter over local data. Past HotSpot's 8000-byte limit the method is not
JIT-compiled, and the predicate runs interpreted on every row; past 64KB the
class does not compile, and every Predicate.create logs the error and falls
back to InterpretedPredicate. Results stay correct, and the conjunct counts at
which these limits are
crossed were not measured.
**Recommended change:** Group consecutive conjunct blocks (precompute calls,
conjunct code and check) into private boolean methods once their code exceeds
spark.sql.codegen.methodSplitThreshold. Each method returns false at a FALSE
conjunct and true otherwise. The NULL-conjunct flag becomes a mutable field
reset at the start of eval. eval calls the methods in order and returns false
as soon as one does. A predicate that fits within the threshold keeps the
single inline block.
**Why this works:** The existing split helper already groups code into
methods by the split threshold and nests them through inner classes. Feeding it
boolean-returning blocks, folded so that a false result returns from eval,
bounds every generated method as on the base branch, while keeping the conjunct
order, the per-conjunct placement and AND's three-valued result.
**Scope:** Split the conjunct-wise eval body into threshold-sized methods
without changing evaluation order or results.
**Risks:** The NULL-conjunct flag becomes per-instance mutable state of the
generated predicate, so it must be reset at the start of every evaluation.
**Constraints:** Split methods must receive only the input row as an
argument, never mutable state.
**Success:** For a predicate with hundreds of conjuncts and a shared
subexpression, every generated method stays within the split threshold and the
class compiles. Results and the per-conjunct placement of shared subexpressions
are unchanged.
**Verification:**
- **Regression:** A generated predicate with several hundred conjuncts plus
one shared non-leaf subexpression compiles, keeps every generated method below
the HotSpot huge-method limit, and matches interpreted evaluation on rows where
some conjunct is FALSE, rows where some conjunct is NULL, and rows where all
are TRUE.
- **Behavior:** Small predicates still generate a single inline block, and a
FALSE conjunct inside a split method still stops evaluation before any later
conjunct's precompute.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodegenSubexpressionEliminationSuite.scala:
##########
@@ -56,18 +56,81 @@ class CodegenSubexpressionEliminationSuite extends
SparkFunSuite {
}
}
+ test("SPARK-60332: GeneratePredicate evaluates a common subexpression of
later conjuncts " +
+ "after the conjuncts before it") {
+ // `Not(leaf)` is common to the last two conjuncts, and evaluated just
before the first of
+ // them, so the first conjunct decides whether `leaf` is evaluated at all:
a FALSE one stops
+ // the evaluation, while a TRUE or a NULL one does not, as `And.eval` does.
+ Seq(true, false).foreach { useSubexprElimination =>
+ Seq((false, false, false), (true, true, true), (null, true,
false)).foreach {
+ case (first, leafEvaluated, result) =>
+ // Distinct `id`s, so that the first conjunct and the leaf are not
the same expression.
+ val leaf = ExprWithEvaluatedState(id = 1)
+ val cond = And(And(ExprWithEvaluatedState(first, isNullable = first
== null),
+ Not(leaf)), Not(leaf))
+ val instance = GeneratePredicate.generate(cond,
useSubexprElimination)
+ instance.initialize(0)
+ withClue(s"useSubexprElimination = $useSubexprElimination, first =
$first: ") {
+ assert(instance.eval(null) === result)
+ assert(leaf.evaluated === leafEvaluated)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60332: GeneratePredicate evaluates nested common subexpressions
in the order of " +
+ "the conjuncts that first reference them") {
+ // `Not(leaf)` is first referenced by the second conjunct and
`Or(Not(leaf), leaf2)`, which
+ // contains it, by the third. Each is evaluated just before its conjunct,
the inner one first,
+ // so the outer one reads the evaluated inner one, not its default. A
FALSE first conjunct
+ // evaluates neither.
+ Seq((false, false, false), (true, true, true)).foreach { case (first,
leafEvaluated, result) =>
+ val leaf = ExprWithEvaluatedState(id = 1)
+ val leaf2 = ExprWithEvaluatedState(id = 2)
+ val inner = Not(leaf)
+ val outer = Or(inner, leaf2)
+ val cond = And(And(And(ExprWithEvaluatedState(first), inner), And(inner,
outer)), outer)
+ val instance = GeneratePredicate.generate(cond, useSubexprElimination =
true)
+ instance.initialize(0)
+ withClue(s"first = $first: ") {
+ assert(instance.eval(null) === result)
+ assert(leaf.evaluated === leafEvaluated)
+ // `Or(TRUE, leaf2)` never evaluates `leaf2`.
+ assert(!leaf2.evaluated)
+ }
+ }
+ }
+
+ test("SPARK-60332: GeneratePredicate splits the evaluation of many common
subexpressions " +
+ "into functions") {
+ // The calls of the functions of 100 common subexpressions before the
conjunct that first
+ // references them exceed the length at which they are split into
functions of their own.
+ Seq((false, false, false), (true, true, true)).foreach {
+ case (first, leavesEvaluated, result) =>
+ val leaves = (1 to 100).map(i => ExprWithEvaluatedState(id = i))
+ val common = leaves.map(leaf => Not(leaf): Expression).reduce(And)
Review Comment:
**Nit (P3):** GeneratePredicate.create flattens every nested AND, so
`And(And(first, common), common)` becomes 201 conjuncts and each `Not(leaf_i)`
is first referenced by its own conjunct. Every conjunct then gets a single
call, splitExpressions inlines a single block, and the multi-block split this
test is named for never runs: the assertions would still pass if that split
were broken or removed. Reaching it needs many common subexpressions first
referenced by one conjunct, for example `And(And(first, Not(Not(common))),
common)`, where the second conjunct holds every `Not(leaf_i)` and the flattened
conjuncts after it share them. The same flattening affects the nested test's
comment above: its conjuncts are `[first, inner, inner, outer, outer]`, so
`outer` is first referenced by the fourth conjunct, not the third.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodegenSubexpressionEliminationSuite.scala:
##########
@@ -56,18 +56,81 @@ class CodegenSubexpressionEliminationSuite extends
SparkFunSuite {
}
}
+ test("SPARK-60332: GeneratePredicate evaluates a common subexpression of
later conjuncts " +
+ "after the conjuncts before it") {
+ // `Not(leaf)` is common to the last two conjuncts, and evaluated just
before the first of
+ // them, so the first conjunct decides whether `leaf` is evaluated at all:
a FALSE one stops
+ // the evaluation, while a TRUE or a NULL one does not, as `And.eval` does.
+ Seq(true, false).foreach { useSubexprElimination =>
+ Seq((false, false, false), (true, true, true), (null, true,
false)).foreach {
+ case (first, leafEvaluated, result) =>
+ // Distinct `id`s, so that the first conjunct and the leaf are not
the same expression.
+ val leaf = ExprWithEvaluatedState(id = 1)
+ val cond = And(And(ExprWithEvaluatedState(first, isNullable = first
== null),
+ Not(leaf)), Not(leaf))
+ val instance = GeneratePredicate.generate(cond,
useSubexprElimination)
+ instance.initialize(0)
+ withClue(s"useSubexprElimination = $useSubexprElimination, first =
$first: ") {
+ assert(instance.eval(null) === result)
+ assert(leaf.evaluated === leafEvaluated)
Review Comment:
**Nit (P3):** This flag records only whether `leaf` ran, and the
expectations are the same with elimination on and off, so the test passes
whether the conjuncts reuse the shared state or recompute `Not(leaf)` at each
occurrence. None of the new tests has a common subexpression whose first
occurrence is in an IF or CASE WHEN branch of an earlier conjunct, or a
semantically equal but structurally different copy, so a call placed after a
branch that reads its state, which then sees the previous row's value, would
also go unnoticed. Counting evaluations per row, and adding such a branch-first
case evaluated over rows with different values, would give both properties a
failure signal.
See **Shared repair plan 1** in the review body.
--
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]
