spark-test commented on code in PR #60163:
URL: https://github.com/apache/spark/pull/60163#discussion_r4490299455
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -230,21 +230,51 @@ object ConstantPropagation extends Rule[LogicalPlan] {
private def safeToReplace(ar: AttributeReference, nullIsFalse: Boolean) =
!ar.nullable || nullIsFalse
+ /**
+ * Replaces attributes in the comparisons of `condition` with their constant
values.
+ *
+ * The constant is the value of an attribute only on the rows that pass the
whole condition.
+ * The rewritten condition also runs on other rows, e.g. a row with `y = 1`
for
+ * `10 / (id - y) > 0 AND y = 0`, and a part of it that becomes constant is
folded with no row
+ * at all, e.g. `10 / y` for `y = 0 AND id > 10 / y`. So that this raises no
error the original
+ * condition cannot raise, an expression that can fail keeps its attributes,
unless all of them
+ * are replaced and the result evaluates without an error.
+ */
private def replaceConstants(
condition: Expression,
equalityPredicates: AttributeMap[(Literal, BinaryComparison)]):
Expression = {
val constantsMap = AttributeMap(equalityPredicates.map { case (attr, (lit,
_)) => attr -> lit })
val predicates = equalityPredicates.values.map(_._2).toSet
def replaceInComparison(b: BinaryComparison): Expression = {
lazy val collationSafeReplacement = isSameCollationAttrRefComparison(b)
- b transform {
- case a: AttributeReference
- if isBinaryStable(a.dataType) || collationSafeReplacement =>
- constantsMap.getOrElse(a, a)
+ def canReplace(a: Attribute): Boolean =
+ constantsMap.contains(a) && (isBinaryStable(a.dataType) ||
collationSafeReplacement)
+ def replace(e: Expression): Expression = e match {
+ case a: AttributeReference if canReplace(a) => constantsMap(a)
+ case _ if ExprUtils.isTotalOperator(e) => e.mapChildren(replace)
+ case _ if e.references.nonEmpty && e.references.forall(canReplace) =>
+ val replaced = e.transform { case a: AttributeReference =>
constantsMap(a) }
+ if (canFold(replaced)) replaced else e
+ case _ => e
Review Comment:
**Blocking (P1):** Keeping the attribute here means the expression now runs
with the row's own value on rows that fail the equality, while guards written
over the same column are still substituted and folded away. Under ANSI, `SELECT
* FROM t WHERE d <> 0 AND n / d > 1 AND d = 5` on rows (10, 0) and (10, 5)
becomes `n / d > 1 AND d = 5` and fails with DIVIDE_BY_ZERO on (10, 0), where
the base branch produced `n / 5.0 > 1.0 AND d = 5` and returned (10, 5). The
PR's own `10 / (id - y) > 0 AND y = 0` fails the same way on a row (5, 5),
where the base branch returned no rows. Any conjunct that keeps a propagated
column and comes before the equality, as written or after CombineFilters puts a
view's filter first, is affected: casts, array indexes, overflow, Scala UDFs,
assert_true, and CASE branches that can fail. Python UDF conjuncts are not,
because the equality is pushed below the node that evaluates them. So the
description's statement that these comparisons give the same results do
es not hold for such data.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -230,21 +230,51 @@ object ConstantPropagation extends Rule[LogicalPlan] {
private def safeToReplace(ar: AttributeReference, nullIsFalse: Boolean) =
!ar.nullable || nullIsFalse
+ /**
+ * Replaces attributes in the comparisons of `condition` with their constant
values.
+ *
+ * The constant is the value of an attribute only on the rows that pass the
whole condition.
+ * The rewritten condition also runs on other rows, e.g. a row with `y = 1`
for
+ * `10 / (id - y) > 0 AND y = 0`, and a part of it that becomes constant is
folded with no row
+ * at all, e.g. `10 / y` for `y = 0 AND id > 10 / y`. So that this raises no
error the original
+ * condition cannot raise, an expression that can fail keeps its attributes,
unless all of them
+ * are replaced and the result evaluates without an error.
+ */
private def replaceConstants(
condition: Expression,
equalityPredicates: AttributeMap[(Literal, BinaryComparison)]):
Expression = {
val constantsMap = AttributeMap(equalityPredicates.map { case (attr, (lit,
_)) => attr -> lit })
val predicates = equalityPredicates.values.map(_._2).toSet
def replaceInComparison(b: BinaryComparison): Expression = {
lazy val collationSafeReplacement = isSameCollationAttrRefComparison(b)
- b transform {
- case a: AttributeReference
- if isBinaryStable(a.dataType) || collationSafeReplacement =>
- constantsMap.getOrElse(a, a)
+ def canReplace(a: Attribute): Boolean =
+ constantsMap.contains(a) && (isBinaryStable(a.dataType) ||
collationSafeReplacement)
+ def replace(e: Expression): Expression = e match {
+ case a: AttributeReference if canReplace(a) => constantsMap(a)
+ case _ if ExprUtils.isTotalOperator(e) => e.mapChildren(replace)
+ case _ if e.references.nonEmpty && e.references.forall(canReplace) =>
+ val replaced = e.transform { case a: AttributeReference =>
constantsMap(a) }
Review Comment:
**Blocking (P1):** `e.references.forall(canReplace)` checks only the columns
in `references`, but this transform looks up every `AttributeReference` it
visits with `constantsMap(a)`, outside canFold's try/catch. For a correlated
subquery, `references` covers only the outer columns, while its children also
include the join condition that PullupCorrelatedPredicates pulled up, over the
subquery's own columns. `AttributeMap` does not override `apply`, so the
inherited `Map.apply` throws for those. As a result, `SELECT * FROM t1 WHERE
t1.a = 1 AND t1.b = (SELECT max(t2.b) FROM t2 WHERE t2.a = t1.a)`, the common
latest-row-for-a-key pattern, now fails during optimization with
`java.util.NoSuchElementException: key not found`, where the base branch ran
it. The same applies under COALESCE, under other operators whose remaining
columns are also equated to literals, and to EXISTS or IN nested in a
comparison. No test puts a subquery operand in a rewritten comparison.
**Verification:**
- **Regression:** A rule-level case whose comparison operand is a correlated
scalar subquery, with its outer attribute propagated and an inner attribute in
its join condition, optimizes without error and keeps the subquery unchanged.
- **Behavior:** An end-to-end query of the form `t1.a = 1 AND t1.b = (SELECT
max(t2.b) FROM t2 WHERE t2.a = t1.a)` returns the expected rows.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -230,21 +230,51 @@ object ConstantPropagation extends Rule[LogicalPlan] {
private def safeToReplace(ar: AttributeReference, nullIsFalse: Boolean) =
!ar.nullable || nullIsFalse
+ /**
+ * Replaces attributes in the comparisons of `condition` with their constant
values.
+ *
+ * The constant is the value of an attribute only on the rows that pass the
whole condition.
+ * The rewritten condition also runs on other rows, e.g. a row with `y = 1`
for
+ * `10 / (id - y) > 0 AND y = 0`, and a part of it that becomes constant is
folded with no row
+ * at all, e.g. `10 / y` for `y = 0 AND id > 10 / y`. So that this raises no
error the original
+ * condition cannot raise, an expression that can fail keeps its attributes,
unless all of them
+ * are replaced and the result evaluates without an error.
+ */
private def replaceConstants(
condition: Expression,
equalityPredicates: AttributeMap[(Literal, BinaryComparison)]):
Expression = {
val constantsMap = AttributeMap(equalityPredicates.map { case (attr, (lit,
_)) => attr -> lit })
val predicates = equalityPredicates.values.map(_._2).toSet
def replaceInComparison(b: BinaryComparison): Expression = {
lazy val collationSafeReplacement = isSameCollationAttrRefComparison(b)
- b transform {
- case a: AttributeReference
- if isBinaryStable(a.dataType) || collationSafeReplacement =>
- constantsMap.getOrElse(a, a)
+ def canReplace(a: Attribute): Boolean =
+ constantsMap.contains(a) && (isBinaryStable(a.dataType) ||
collationSafeReplacement)
+ def replace(e: Expression): Expression = e match {
+ case a: AttributeReference if canReplace(a) => constantsMap(a)
+ case _ if ExprUtils.isTotalOperator(e) => e.mapChildren(replace)
+ case _ if e.references.nonEmpty && e.references.forall(canReplace) =>
+ val replaced = e.transform { case a: AttributeReference =>
constantsMap(a) }
+ if (canFold(replaced)) replaced else e
Review Comment:
**Nit (P3):** `canFold` already computes the folded literal and then
discards it, so the unfolded `replaced` reaches ConstantFolding, which
evaluates the same constant again a few rules later in the same rule set. For
`y = 5 AND x = y + 1` that is `5 + 1` twice instead of once. Plans and results
are unchanged, but an expensive constant, such as `size(sequence(1, y))` with a
large propagated y, is built twice during planning.
**Verification:**
- **Regression:** Running ConstantPropagation alone on `y = 5 AND x = y + 1`
yields `y = 5 AND x = 6`, so no unfolded constant is left for ConstantFolding
to evaluate again.
- **Compatibility:** The existing ConstantPropagation cases that fold a
fully substituted subexpression, such as `b = 2 AND a > 10.0 / CAST(b AS
DOUBLE)` becoming `b = 2 AND a > 5.0`, keep their optimized plans.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -230,21 +230,51 @@ object ConstantPropagation extends Rule[LogicalPlan] {
private def safeToReplace(ar: AttributeReference, nullIsFalse: Boolean) =
!ar.nullable || nullIsFalse
+ /**
+ * Replaces attributes in the comparisons of `condition` with their constant
values.
+ *
+ * The constant is the value of an attribute only on the rows that pass the
whole condition.
+ * The rewritten condition also runs on other rows, e.g. a row with `y = 1`
for
+ * `10 / (id - y) > 0 AND y = 0`, and a part of it that becomes constant is
folded with no row
+ * at all, e.g. `10 / y` for `y = 0 AND id > 10 / y`. So that this raises no
error the original
+ * condition cannot raise, an expression that can fail keeps its attributes,
unless all of them
+ * are replaced and the result evaluates without an error.
+ */
private def replaceConstants(
condition: Expression,
equalityPredicates: AttributeMap[(Literal, BinaryComparison)]):
Expression = {
val constantsMap = AttributeMap(equalityPredicates.map { case (attr, (lit,
_)) => attr -> lit })
val predicates = equalityPredicates.values.map(_._2).toSet
def replaceInComparison(b: BinaryComparison): Expression = {
lazy val collationSafeReplacement = isSameCollationAttrRefComparison(b)
- b transform {
- case a: AttributeReference
- if isBinaryStable(a.dataType) || collationSafeReplacement =>
- constantsMap.getOrElse(a, a)
+ def canReplace(a: Attribute): Boolean =
+ constantsMap.contains(a) && (isBinaryStable(a.dataType) ||
collationSafeReplacement)
+ def replace(e: Expression): Expression = e match {
+ case a: AttributeReference if canReplace(a) => constantsMap(a)
+ case _ if ExprUtils.isTotalOperator(e) => e.mapChildren(replace)
+ case _ if e.references.nonEmpty && e.references.forall(canReplace) =>
+ val replaced = e.transform { case a: AttributeReference =>
constantsMap(a) }
+ if (canFold(replaced)) replaced else e
+ case _ => e
}
+ b.mapChildren(replace)
+ }
+ def replaceInCondition(e: Expression): Expression = e match {
+ case b: BinaryComparison if predicates.contains(b) => b
+ case b: BinaryComparison => replaceInComparison(b)
+ case _ if ExprUtils.isTotalOperator(e) =>
e.mapChildren(replaceInCondition)
Review Comment:
**Non-blocking (P2):** This recursion rewrites each operand of OR, AND,
COALESCE and IN on its own, but those operators short-circuit, so a rewritten
operand decides whether its siblings run. Under ANSI, `SELECT * FROM t WHERE (y
< 3 OR 10 / x > 1) AND y = 5` on a row with y = 1 and x = 0 cannot raise in
either conjunct order, yet `y < 3` becomes `5 < 3`, folds away, and leaves `10
/ x > 1` to fail with DIVIDE_BY_ZERO, as it already did on the base branch.
`coalesce(m[k], 10 / x) > 0 AND k = 'a'` and an IN value or list item inside a
comparison behave the same way. Once the rewrite drops the last reference to
the propagated column, pushdown can also move the conjunct below an Aggregate
or Window: `SELECT g, sum(x) AS s FROM t GROUP BY g HAVING s = 5 AND (s < 3 OR
10 / g > 1)` fails on a table whose only group has g = 0 and s = 1. So the new
doc comment's guarantee, no error the original condition cannot raise, does not
hold for these inputs.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -230,21 +230,51 @@ object ConstantPropagation extends Rule[LogicalPlan] {
private def safeToReplace(ar: AttributeReference, nullIsFalse: Boolean) =
!ar.nullable || nullIsFalse
+ /**
+ * Replaces attributes in the comparisons of `condition` with their constant
values.
+ *
+ * The constant is the value of an attribute only on the rows that pass the
whole condition.
+ * The rewritten condition also runs on other rows, e.g. a row with `y = 1`
for
+ * `10 / (id - y) > 0 AND y = 0`, and a part of it that becomes constant is
folded with no row
+ * at all, e.g. `10 / y` for `y = 0 AND id > 10 / y`. So that this raises no
error the original
+ * condition cannot raise, an expression that can fail keeps its attributes,
unless all of them
+ * are replaced and the result evaluates without an error.
+ */
private def replaceConstants(
condition: Expression,
equalityPredicates: AttributeMap[(Literal, BinaryComparison)]):
Expression = {
val constantsMap = AttributeMap(equalityPredicates.map { case (attr, (lit,
_)) => attr -> lit })
val predicates = equalityPredicates.values.map(_._2).toSet
def replaceInComparison(b: BinaryComparison): Expression = {
lazy val collationSafeReplacement = isSameCollationAttrRefComparison(b)
- b transform {
- case a: AttributeReference
- if isBinaryStable(a.dataType) || collationSafeReplacement =>
- constantsMap.getOrElse(a, a)
+ def canReplace(a: Attribute): Boolean =
+ constantsMap.contains(a) && (isBinaryStable(a.dataType) ||
collationSafeReplacement)
+ def replace(e: Expression): Expression = e match {
+ case a: AttributeReference if canReplace(a) => constantsMap(a)
+ case _ if ExprUtils.isTotalOperator(e) => e.mapChildren(replace)
+ case _ if e.references.nonEmpty && e.references.forall(canReplace) =>
+ val replaced = e.transform { case a: AttributeReference =>
constantsMap(a) }
+ if (canFold(replaced)) replaced else e
+ case _ => e
}
+ b.mapChildren(replace)
+ }
+ def replaceInCondition(e: Expression): Expression = e match {
+ case b: BinaryComparison if predicates.contains(b) => b
+ case b: BinaryComparison => replaceInComparison(b)
+ 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
Review Comment:
**Nit (P3):** Returning other containers unchanged is what keeps a
substituted branch condition, as in `IF(c > b, true, 10 / c > 1) AND b = 0`,
from sending a row with b = -1 and c = 0 into the failing branch under ANSI. No
test exercises it: every new SPARK-60983 case puts the division directly under
the rewritten comparison, and ConstantPropagationSuite has no IF or CASE WHEN
case. If this went back to descending into every comparison, the failure would
return with every assertion still passing.
**Verification:**
- **Regression:** Under ANSI mode, `IF(c > b, true, 10.0 / CAST(c AS DOUBLE)
> 0.0) AND b = 0` keeps its IF predicate and raises no error on a row with b =
-1 and c = 0, and the case fails against the base branch's rewriting.
--
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]
