spark-test commented on code in PR #60190:
URL: https://github.com/apache/spark/pull/60190#discussion_r4500547845
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -264,12 +267,31 @@ object ConstantPropagation extends Rule[LogicalPlan] with
PredicateHelper {
def replaceInCondition(e: Expression): Expression = e match {
case b: BinaryComparison if predicates.contains(b) => b
case b: BinaryComparison => replaceInComparison(b)
+ case _: And | _: Or | _: Coalesce | _: In => replaceOperands(e,
replaceInCondition)
case _ if ExprUtils.isTotalOperator(e) =>
e.mapChildren(replaceInCondition)
// The children of these include the comparisons of a pulled up
correlated condition.
case _: InSubquery | _: ListQuery | _: Exists =>
e.mapChildren(replaceInCondition)
case _ => e
}
- replaceInCondition(condition)
+ def replaceInConjuncts(e: Expression): Expression = e match {
+ case a: And => a.mapChildren(replaceInConjuncts)
+ case _ => replaceInCondition(e)
+ }
+ replaceInConjuncts(condition)
+ }
+
+ /**
+ * Rewrites the operands of `e`, which short-circuits, with `f`. An operand
decides whether the
+ * ones after it run, so it is rewritten only if none of those, as
rewritten, can fail.
+ */
+ private def replaceOperands(e: Expression, f: Expression => Expression):
Expression = {
+ val (newChildren, _) = e.children.foldRight((List.empty[Expression],
false)) {
+ case (child, (laterChildren, laterCanFail)) =>
+ val newChild = if (laterCanFail) child else f(child)
Review Comment:
**Blocking (P1):** This keeps the original operand whenever a later one can
fail, even when the rewritten operand folds to its operator's absorbing value:
TRUE in `OR`, FALSE in `AND`, or a non-null literal at the head of `COALESCE`.
Such a rewrite makes every later operand unreachable, so it can never run
anything the condition as written skips, and the base branch applied it in
every ANSI mode. Blocking it regresses plans that were already correct.
`country = 'DE' AND ((country = 'US' AND amount * 1.08 > 100) OR (country =
'CA' AND amount * 1.13 > 100))` used to fold to FALSE, and PruneFilters
replaced the scan with an empty relation. Now it scans every DE row and
evaluates both dead branches. The description's own `y = 5 AND (y > 3 OR 10 / x
> 1)` and `y = 5 AND coalesce(y, 10 / x) > 3` keep operands that can never run,
and the columns they read. In `y = 5 AND ((y < 3 AND 10 / x > 1) OR z = 1)`, `z
= 1` no longer drives partition pruning or source pushdown. On a row where `y`
is NULL inside an `OR` branch, the kept operand can also raise the error that
the condition as written raises, where the base branch returned the row.
Keeping the rewrite when it folds to the absorbing value, and treating the
operands after it as unreachable, restores these plans. The cases this PR fixes
fold to values that are not absorbing, so they stay blocked.
**Verification:**
- **Regression:** With ANSI mode on, OR, AND and COALESCE operands whose
rewrite folds to TRUE, FALSE or a non-null literal before an operand that can
fail are rewritten and folded, including a contradiction that prunes to an
empty relation.
- **Compatibility:** Operands placed before an operand that can fail stay
unrewritten when their rewrite does not fold to the absorbing value, such as
FALSE in OR, TRUE in AND, or a lookup that can be NULL in COALESCE.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -303,7 +328,11 @@ object ConstantPropagation extends Rule[LogicalPlan] with
PredicateHelper {
val keepsFailable = others.exists { c =>
c.references.intersect(propagated).nonEmpty &&
!ExprUtils.canEvaluateUnconditionally(c)
}
- if (keepsFailable && reordered != conjuncts) {
+ val firstRewritten = splitConjunctivePredicates(original).zip(conjuncts)
+ .indexWhere { case (before, after) => !before.fastEquals(after) }
+ val unmasksFailable = firstRewritten >= 0 &&
Review Comment:
**Non-blocking (P2):** Moving the equalities first relies on a failing
equality to stop the `AND`, but a propagated column can be nullable:
`safeToReplace` admits it wherever NULL counts as FALSE, which includes every
`OR` branch. On a row where it is NULL, `y = 1` is NULL. `And.eval` keeps
evaluating past it, and so does FilterExec with the default
`spark.sql.legacy.filterShortCircuitOnNullConjunct=false`. A guard that is
FALSE on that row as written, such as `coalesce(y, 0) > 0`, `nvl(y, 0) > 0` or
`y <=> z`, is rewritten to TRUE and folded away, so the later conjunct still
runs.
With ANSI on and default configs, `SELECT * FROM t WHERE (coalesce(y, 0) > 0
AND 10 / x > 0 AND y = 1) OR w` fails with DIVIDE_BY_ZERO on the row (NULL, 0,
true), where the condition as written is TRUE without dividing. No
`isnotnull(y)` is inferred from a disjunction. At the filter top level, the
inferred `isnotnull(y)` masks the same shape unless
`spark.sql.constraintPropagation.enabled` is false. The rewrite of an `OR`
branch can also turn the branch from FALSE into NULL for an enclosing `AND`, so
`((y = 1 AND coalesce(y, 0) > 0) OR w) AND 10 / x > 0` fails on (NULL, 0,
false). The reorder the base branch already does has the same gap, for example
`coalesce(a, 0) > 0 AND 10 / coalesce(a, c) > 0 AND a = 1`. The new test's
guard `a > 0` is NULL rather than FALSE on a NULL row, so the test cannot tell
these cases apart.
**Recommended change:** When ConstantPropagation rewrites an AND whose
equalities include a nullable attribute, keep the rewritten AND FALSE on rows
where that attribute is NULL. Insert a not-null check of the attribute right
after the equalities, unless one already precedes the conjuncts it guards, in
two cases. The first is when the rule relies on the equalities running first to
guard a later conjunct, that is, whenever moveEqualitiesFirst finds a conjunct
that can fail after a rewritten one or still referring to a propagated
attribute, including when the equalities are already first and nothing moves.
The second is when an AND inside an OR branch has a rewritten conjunct over
that attribute. Add optimizer tests and query tests with NULL rows.
**Why this works:** In a filter, an equality between an attribute and a
literal already implies that the attribute is not NULL, so the check changes no
result. Unlike the equality, the check is FALSE rather than NULL on a NULL row.
AND therefore stops before the conjunct that can fail, and an AND inside an OR
stays FALSE instead of becoming NULL.
**Scope:** Make the protection from running the equalities first also hold
on rows where a propagated nullable attribute is NULL.
**Compatibility:** Rows where the propagated attribute is not NULL are
accepted or rejected exactly as before, and filters over non-nullable
attributes keep their current plans.
**Risks:** Some plans gain a not-null check: ANDs whose equalities guard a
later conjunct that can fail, and ANDs inside an OR whose conjunct was
rewritten. Results do not change and the check cannot fail, but plan-stability
outputs that contain such filters may change, and so does the expected plan of
the existing ConstantPropagationSuite test that combines AND and OR predicates,
whose equality on the nullable `b` guards an OR that can fail. No configuration
gates it, like the rest of the PR's change.
**Constraints:** Add the check only for a nullable attribute that was
replaced where NULL counts as FALSE, never under NOT. Keep the rule idempotent:
a later run must neither add a second check nor reorder the first one.
**Success:** With ANSI mode on, `(coalesce(y, 0) > 0 AND 10 / x > 0 AND y =
1) OR w` returns a row where y is NULL, x is 0 and w is true, without a
division-by-zero error. With constraint propagation disabled, `coalesce(y, 0) >
0 AND 10 / x > 0 AND y = 1` raises no error on a row where y is NULL and x is
0. With constraint propagation disabled, `y = 1 AND coalesce(y, 0) > 0 AND 10 /
x > 0`, whose equality is already first, raises no error on a row where y is
NULL and x is 0. `((y = 1 AND coalesce(y, 0) > 0) OR w) AND 10 / x > 0` raises
no error on a row where y is NULL, w is false and x is 0. `coalesce(a, 0) > 0
AND 10 / coalesce(a, c) > 0 AND a = 1`, which the base branch already reorders,
raises no error on a row where a is NULL and c is 0. Rows where the propagated
attribute is not NULL are filtered as before, and the PR's example queries
still return their results.
**Verification:**
- **Regression:** With ANSI mode on, the OR-branch condition, the top-level
condition with constraint propagation disabled, and the enclosing-AND condition
above raise no division-by-zero error on rows where the propagated attribute is
NULL and the divisor is 0, with whole-stage codegen enabled and disabled.
- **Regression:** At the optimizer level, a rewritten AND over a nullable
propagated attribute that guards a conjunct that can fail checks that the
attribute is not NULL before that conjunct, whether or not the equalities had
to move, and an AND inside an OR branch whose conjunct was rewritten stays
FALSE on rows where the attribute is NULL.
- **Compatibility:** Filters over non-nullable propagated attributes keep
their current optimized plans. Otherwise only an AND whose equalities guard a
later conjunct, and an AND inside an OR branch whose conjunct was rewritten,
gain the not-null check, and optimizing the plan again adds no second check.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ConstantPropagationSuite.scala:
##########
@@ -255,4 +255,35 @@ class ConstantPropagationSuite extends PlanTest {
comparePlans(Optimize.execute(conditional), conditional)
}
}
+
+ test("SPARK-61033: Do not propagate a constant into an operand that guards
an expression " +
+ "that can fail") {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ def divide(e: Expression): Expression = Divide(Literal(10.0),
e.cast(DoubleType))
+ def first(plan: LogicalPlan): String = plan match {
+ case Filter(condition, _) =>
splitConjunctivePredicates(condition).head.sql
+ }
+
+ // `a < 3` and `a > 3` decide whether the division runs. With `a`
replaced, it would run on a
+ // row with `a` = 1 and `c` = 0. The conjunct keeps `a`, so the equality
goes first.
+ val nested = testRelation.where(
+ (columnA < 3 || (columnA > 3 && divide(columnC) > Literal(1.0))) &&
columnA === 5).analyze
+ val optimizedNested = Optimize.execute(nested)
+ comparePlans(optimizedNested, nested)
+ assert(first(optimizedNested) == "(a = 5)")
+
+ // With `k` replaced, the map lookup can be NULL on a row with another
`k`.
+ val maps = LocalRelation($"k".string, Symbol("m").map(StringType,
DoubleType), $"x".int)
+ val inComparison = maps.where(Coalesce(Seq(GetMapValue($"m", $"k"),
divide($"x"))) >
+ Literal(0.0) && $"k" === "a").analyze
+ comparePlans(Optimize.execute(inComparison), inComparison)
+
+ // A conjunct guards the ones after it as well. It is rewritten, and the
equality goes first.
+ val conjunct = Optimize.execute(testRelation
+ .where(columnA > 0 && divide(columnC) > Literal(0.0) && columnA ===
1).analyze)
+ comparePlans(conjunct,
+ testRelation.where(columnA === 1 && divide(columnC) >
Literal(0.0)).analyze)
+ assert(first(conjunct) == "(a = 1)")
Review Comment:
**Nit (P3):** This assertion is the only check of the order the fix depends
on, and it runs on the output of a three-rule `RuleExecutor` over a
`LocalRelation`, while `comparePlans` normalizes conjunct order. No test runs
the description's t1, t2 or t3 queries through the full optimizer and
FilterExec. If a later rule moved `10 / x > 0` ahead of `y = 1`, or FilterExec
evaluated the conjuncts in another order, t2 would fail with DIVIDE_BY_ZERO
again while this suite stayed green. Running the three queries through
`checkAnswer` with ANSI on, with whole-stage codegen on and off, would protect
the user-visible fix, as SPARK-60985 did in `DataFrameSuite`.
**Verification:**
- **Behavior:** With ANSI mode on, the OR-guard query returns no rows, the
guarding-conjunct query returns (1, 5), and the HAVING query returns no rows,
with whole-stage codegen enabled and disabled.
- **Regression:** If the equalities were no longer evaluated before the
guarded conjunct, the guarding-conjunct query would fail the test.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ConstantPropagationSuite.scala:
##########
@@ -255,4 +255,35 @@ class ConstantPropagationSuite extends PlanTest {
comparePlans(Optimize.execute(conditional), conditional)
}
}
+
+ test("SPARK-61033: Do not propagate a constant into an operand that guards
an expression " +
Review Comment:
**Nit (P3):** This test covers `OR`, `AND` and `COALESCE`, but no case
builds an `IN`, so removing `_: In` from either dispatch site in
`expressions.scala` leaves the suite passing. The shape is reachable. `IS TRUE`
parses to `EqualNullSafe(e, true)`, so in `(a IN (1, 10 / c)) IS TRUE AND a =
5` the `IN` sits inside a comparison. Without the guard, its value folds to
`5.0` while `10 / c` stays, and on a row with a = 1 and c = 0 the list divides
by zero, although the original matches the first item. A case like that, plus a
condition-level one such as `flag IN (a > 3, 10 / c > 1) AND a = 5`, would
cover both arms.
**Verification:**
- **Regression:** An IN inside a comparison whose later list item can fail
keeps its value unrewritten, so the test fails if IN is handled as a plain
total operator there.
- **Regression:** A boolean IN list at the condition level whose later item
can fail keeps its earlier items unrewritten, so the test fails if IN is
handled as a plain total operator there.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -264,12 +267,31 @@ object ConstantPropagation extends Rule[LogicalPlan] with
PredicateHelper {
def replaceInCondition(e: Expression): Expression = e match {
case b: BinaryComparison if predicates.contains(b) => b
case b: BinaryComparison => replaceInComparison(b)
+ case _: And | _: Or | _: Coalesce | _: In => replaceOperands(e,
replaceInCondition)
case _ if ExprUtils.isTotalOperator(e) =>
e.mapChildren(replaceInCondition)
// The children of these include the comparisons of a pulled up
correlated condition.
case _: InSubquery | _: ListQuery | _: Exists =>
e.mapChildren(replaceInCondition)
case _ => e
}
- replaceInCondition(condition)
+ def replaceInConjuncts(e: Expression): Expression = e match {
+ case a: And => a.mapChildren(replaceInConjuncts)
+ case _ => replaceInCondition(e)
+ }
+ replaceInConjuncts(condition)
+ }
+
+ /**
+ * Rewrites the operands of `e`, which short-circuits, with `f`. An operand
decides whether the
+ * ones after it run, so it is rewritten only if none of those, as
rewritten, can fail.
+ */
+ private def replaceOperands(e: Expression, f: Expression => Expression):
Expression = {
+ val (newChildren, _) = e.children.foldRight((List.empty[Expression],
false)) {
+ case (child, (laterChildren, laterCanFail)) =>
+ val newChild = if (laterCanFail) child else f(child)
+ (newChild :: laterChildren,
+ laterCanFail || !ExprUtils.canEvaluateUnconditionally(newChild))
Review Comment:
**Non-blocking (P2):** This check runs for every operand at every nesting
level, including the first operand, whose result `val (newChildren, _)`
discards. `canEvaluateUnconditionally` walks the operand's whole subtree
without memoization, and the recursion through nested `AND`/`OR` re-walks each
subtree at every level above it. A DataFrame filter such as `col("dt") ===
"2024-01-01" && ids.map(i => col("id") === i).reduce(_ || _)` builds a
left-deep chain of N disjuncts, so one application of the rule visits O(N^2)
nodes although nothing is rewritten. The base branch made one linear pass, and
the cost repeats in every iteration of both fixed-point operator-optimization
batches. Right-deep chains re-walk each later operand the same way, and
SQL-parsed chains, which are balanced, cost O(N log N). `unmasksFailable` also
splits and zips the condition again even when `keepsFailable` already decided
the outcome. Returning failability from the recursive rewrite, skipping the
first operan
d and computing `unmasksFailable` lazily remove the extra work without
changing any plan.
**Verification:**
- **Inspection:** Whether each operand can fail is determined once per rule
application, and no subtree is walked again after its operands were checked.
- **Compatibility:** All existing constant-propagation plan expectations,
including the PR's new cases, are unchanged.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -253,6 +255,7 @@ object ConstantPropagation extends Rule[LogicalPlan] with
PredicateHelper {
constantsMap.contains(a) && (isBinaryStable(a.dataType) ||
collationSafeReplacement)
def replace(e: Expression): Expression = e match {
case a: AttributeReference if canReplace(a) => constantsMap(a)
+ case _: And | _: Or | _: Coalesce | _: In => replaceOperands(e,
replace)
Review Comment:
**Non-blocking (P2):** Only `AND`, `OR`, `COALESCE` and `IN` reach
`replaceOperands`. Comparisons and map lookups are still rewritten operand by
operand, through `replaceInComparison`'s `b.mapChildren(replace)` and the
total-operator case below this line. So the failure this PR fixes remains in
two forms.
First, the HAVING pushdown from the description, through a comparison
instead of an `OR`. With ANSI on, `SELECT g, sum(x) s FROM t3 GROUP BY g HAVING
s = 5 AND s > 10 / g` over the row (0, 1) rewrites the comparison to `5.0 > 10
/ g`. That no longer refers to `s`, and nothing reorders it.
`pushBelowAggregate` then pushes it past the unpushable `s = 5`, because a
division does not report `throwable`. The query fails with DIVIDE_BY_ZERO on
the input row, although `s = 5` is FALSE for the only group. The same happens
with `10 / g < s` and with `count(*)`, so no NULL is involved.
Second, a comparison other than `<=>`, or a map lookup, skips its right
operand while the left one is NULL. `m[k] > 10 / x AND k = 'a'` becomes `m['a']
> 10 / x`, which divides on a row with k = 'b' and x = 0, where `m['b']` is
NULL. `y > 10 / x AND y = 5` divides on a row where y is NULL: with whole-stage
codegen, because the inferred `isnotnull(y)` check is generated only before `y
= 5`, and inside an `OR` branch in either mode.
Over a plain relation, rewriting the right operand when the left one can
fail stays correct, as in `b = 2 AND 10 / c > b`.
**Recommended change:** Extend the PR's guard to comparisons and map lookups
in two ways. First, treat the left operand of every comparison except null-safe
equality, and the map of a map lookup, as guarding the right operand: rewrite
it only when the right operand, as rewritten, cannot fail, or when the left
operand cannot be NULL before the rewrite. Second, when the operator under the
filter produces a propagated attribute itself, as the Aggregate under a HAVING
filter does for its aggregates, leave a conjunct unrewritten if the rewrite
would replace its last reference to that attribute while the rewritten conjunct
can still fail. A conjunct that keeps the attribute is then run after the
equalities by the existing reorder. Update the existing optimizer tests that
propagate into a comparison with a nullable attribute on the left inside an OR,
and add optimizer and query tests for HAVING comparisons, a map lookup and a
nullable attribute on the left of a comparison.
**Why this works:** A kept left operand is still NULL on the rows where it
skipped the right operand before, so those rows still skip it. A conjunct that
still refers to an attribute that the operator under the filter produces cannot
be pushed below that operator, because the attribute does not exist below it,
and the reorder runs it after the equalities, as for the PR's own operators. On
rows where the equalities hold, the original and rewritten operands agree, so
results do not change.
**Scope:** Extend the PR's guard to comparisons and map lookups: a guarding
left operand is not rewritten, and a conjunct that can fail keeps an attribute
that the operator under the filter produces.
**Compatibility:** When the operator under the filter does not produce the
propagated attribute, comparisons whose right operand cannot fail, or whose
left operand cannot be NULL, keep their current rewrite, and no query result
changes.
**Risks:** With the current failure predicate, more comparisons keep their
propagated attribute. A comparison with a nullable propagated attribute on the
left and an arithmetic, cast or string operand on the right keeps the attribute
in every ANSI mode, for example `y > x + 1 AND y = 5`, and the existing
ConstantPropagationSuite expectations that turn `a = c + 3` and `a = b + 3`
inside an OR into `11 = c + 3` and `2 = b + 3` over the nullable `a` change
accordingly. In a HAVING clause, a conjunct that can fail keeps the aggregate
it compares, so it stays above the Aggregate instead of being pushed below it,
and the Aggregate processes more rows, also with ANSI mode off, for example
`HAVING s = 5 AND s > g + 1`. Plan-stability outputs that contain such filters
may change. No configuration gates it, like the rest of the PR's guard, so
restoring the old plans needs a code change, such as a finer failure predicate.
**Constraints:** Rewriting the right operand while the left one can fail
stays allowed when the operator under the filter does not produce the replaced
attribute, as the existing test expects: `b = 2 AND 10 / c > b` becomes `b = 2
AND 10 / c > 2.0` over a relation. Use the same failure predicate as the
existing operand guard, so that a later refinement of that predicate applies to
both.
**Success:** With ANSI mode on, `SELECT g, sum(x) s FROM t3 GROUP BY g
HAVING s = 5 AND s > 10 / g` and the same query with `10 / g < s` return no
rows from a table holding (0, 1), without a division-by-zero error, and so do
both with `count(*)` in place of `sum(x)`. With ANSI mode on, `m[k] > 10 / x
AND k = 'a'` returns its rows from a table that also holds ('b', map('a', 1.0),
0), and `y > 10 / x AND y = 5` raises no error on a row where y is NULL and x
is 0, both at the top level with whole-stage codegen and inside an OR branch.
Over a relation, `b = 2 AND 10 / c > b` still becomes `b = 2 AND 10 / c > 2.0`,
and a comparison whose left operand cannot be NULL is still rewritten.
**Verification:**
- **Regression:** With ANSI mode on, HAVING comparisons that can fail raise
no division-by-zero error on the input rows of groups where the equality does
not hold, with the propagated aggregate on either side of the comparison and
with an aggregate that cannot be NULL such as count.
- **Regression:** With ANSI mode on, a comparison with a map lookup or a
nullable attribute on the left and a right operand that can fail raises no
error on rows where the equality does not hold, in interpreted evaluation, with
whole-stage codegen, and inside an OR branch.
- **Regression:** At the optimizer level, a guarding left operand is kept
with the equality ahead of it, and over an Aggregate a conjunct whose rewrite
would replace its last reference to an aggregate while it can still fail keeps
that reference.
- **Compatibility:** Over a relation, a comparison whose operand that can
fail is on the left, and a comparison whose left operand is not nullable, keep
their current rewrite, such as `b = 2 AND 10 / c > b` into `b = 2 AND 10 / c >
2.0`.
--
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]
