dongjoon-hyun commented on code in PR #60162:
URL: https://github.com/apache/spark/pull/60162#discussion_r4495938486
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2317,17 +2317,40 @@ object InferFiltersFromConstraints extends
Rule[LogicalPlan]
}
private def inferNewFilter(plan: LogicalPlan, constraints: ExpressionSet):
LogicalPlan = {
- val newPredicates = constraints
+ val newPredicates = (constraints
.union(constructIsNotNullConstraints(constraints, plan.output))
.filter { c =>
c.references.nonEmpty && c.references.subsetOf(plan.outputSet) &&
c.deterministic
- } -- plan.constraints
+ } -- plan.constraints).toSeq.filter(canBeFilter)
if (newPredicates.isEmpty) {
plan
} else {
Filter(newPredicates.reduce(And), plan)
}
}
+
+ /**
+ * A constraint only says what holds on the rows an operator outputs. As a
filter, it also runs
+ * on rows and values the query may never evaluate it on, e.g. `10 / x > 0`
from `x = y AND
+ * 10 / y > 0` on a row where `x` is 0 and `y` is not, so only a constraint
that cannot fail
+ * becomes a filter. Constant parts count as what they evaluate to, as they
are folded right
+ * after this rule, and an uncorrelated subquery as a constant, as it runs
once whichever rows
+ * use it. Callers pass the original expressions, because
[[ExpressionSet.filter]] tests the
+ * canonicalized ones.
+ */
+ private def canBeFilter(constraint: Expression): Boolean = {
+ conf.legacyInferFailableFiltersFromConstraints ||
ExprUtils.canEvaluateUnconditionally(
+ ConstantFolding.constantFolding(constraint, isConditionalBranch =
true).transformUp {
+ case c: Cast if isNumericUpCast(c.child.dataType, c.dataType) =>
c.child
Review Comment:
Thanks, reproduced: the query fails with `CAST_OVERFLOW` on the base and on
the head, and returns `[1, 1]` with constraint propagation off.
Fixed in 9759a107a1a. `stripCannotFail` now walks the expression top-down,
so each cast is judged by the type of its original input. The `DataFrameSuite`
test has a row `(3000000000, 1)` and an assertion for `x = y AND CAST(CAST(y AS
FLOAT) AS INT) > 0`.
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala:
##########
@@ -3022,6 +3022,21 @@ class DataFrameSuite extends SharedSparkSession
}
}
+ test("SPARK-60985: an inferred filter does not raise an error the query does
not raise") {
Review Comment:
Good catch. Added in 9759a107a1a, using the assertion from your comment. I
checked that it fails with `DIVIDE_BY_ZERO` when `canBeFilter` folds with
`isConditionalBranch = false`, and passes with the current code.
--
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]
