peter-toth commented on code in PR #60162:
URL: https://github.com/apache/spark/pull/60162#discussion_r4491015259
##########
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(
Review Comment:
**Finding 5.** This breaks an existing test outside the diff, "SPARK-58792:
inferred literal-binding conjunct must not corrupt a copied Hive UDF"
(`sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala:1236`).
Its query is:
```sql
SELECT id FROM gt_table WHERE pt = DATE '2024-09-10' AND hive_gt(CAST(pt AS
TIMESTAMP), created_at)
```
The test asserts that the optimized `Filter` holds two copies of the Hive
UDF. One is the original, and the other is the copy that
`inferConstraintsFromLiteralBindings` makes with `pt` bound to the literal.
That copy is a `HiveGenericUDF`, which `canEvaluateUnconditionally` rejects. On
this head:
```
build/sbt -Phive 'hive/testOnly *HiveUDFSuite -- -z SPARK-58792'
- SPARK-58792: inferred literal-binding conjunct must not corrupt a copied
Hive UDF *** FAILED ***
List(HiveGenericUDF#org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPGreaterThan(cast(pt#17
as timestamp),created_at#18)) had size 1 instead of expected size 2
(HiveUDFSuite.scala:1270)
```
The test guards against two diverging copies of one Hive UDF reaching
execution, and only the legacy behavior still makes them. So it should keep
running with the legacy conf. With its body wrapped in
`withSQLConf(SQLConf.LEGACY_INFER_FAILABLE_FILTERS_FROM_CONSTRAINTS.key ->
"true") { ... }`, both SPARK-58792 tests pass. Its leading comment should then
say that this needs the legacy conf.
##########
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
+ case i: InSubquery =>
i.values.map(IsNotNull(_)).reduce[Expression](And)
Review Comment:
**Finding 7.** The doc comment says that only an uncorrelated subquery
counts as a constant, but the code does not check that. It also drops the
subquery's correlation condition (`joinCond`). In a `WHERE` clause or an inner
join, a correlated subquery never gets here. `Filter.validConstraints` drops
it, and an inner join pushes a one-sided condition down first. An outer join
keeps it in `ON`, though, and `getAllConstraints` reads that condition
directly. With ANSI mode on, `l.x` = {6}, `r.y` = {0, 6} and `s(a, b)` = {(0,
1), (6, 1)}:
```sql
SELECT * FROM l LEFT JOIN r ON l.x = r.y AND l.x IN (SELECT a FROM s WHERE
s.b = 10 / l.x)
```
With constraint propagation off, it returns `[6, null]`. On the base and on
the head it fails with `DIVIDE_BY_ZERO`, with AQE on and off. The copy `r.y IN
(SELECT a FROM s WHERE s.b = 10 / r.y)` passes the check as `IsNotNull(r.y)`,
and it divides by zero on the row with `r.y = 0`.
`EXISTS` has the opposite problem. There is no case for it here, so it keeps
its plan expression and is rejected. For `l LEFT JOIN r ON l.x = r.y AND EXISTS
(SELECT 1 FROM s WHERE s.a = l.x)`, the base adds a semi-join on `r`, and the
head adds none, although `a = y` can't fail. Both return the same rows.
Judging the correlation condition with the rest of the constraint fixes
both. Together with the top-down rewrite of finding 8:
```scala
private def canBeFilter(constraint: Expression): Boolean = {
conf.legacyInferFailableFiltersFromConstraints ||
ExprUtils.canEvaluateUnconditionally(
stripCannotFail(ConstantFolding.constantFolding(constraint,
isConditionalBranch = true)))
}
// Top-down, so that each cast is judged by the type of its original input.
private def stripCannotFail(e: Expression): Expression = e match {
case c: Cast if isNumericUpCast(c.child.dataType, c.dataType) =>
stripCannotFail(c.child)
case i: InSubquery =>
(i.values.map(IsNotNull(_)) ++ i.query.joinCond).map(stripCannotFail)
.reduce[Expression](And)
case x: Exists =>
x.joinCond.map(stripCannotFail).foldLeft[Expression](Literal.TrueLiteral)(And)
case s: ScalarSubquery if !s.isCorrelated => Literal.create(null,
s.dataType)
case other => other.mapChildren(stripCannotFail)
}
```
With it:
- the `IN` query returns `[6, null]`,
- the `EXISTS` query gets its semi-join back,
- `InferFiltersFromConstraintsSuite`, the new `DataFrameSuite` test and the
`*PlanStability*Suite`s (322 tests, ANSI mode on) pass.
The `!s.isCorrelated` guard changes nothing today, because
`RewriteCorrelatedScalarSubquery` runs before this batch. It makes the code
match the comment, though. The comment should then say that a subquery counts
by its correlation condition. "it runs once whichever rows use it" no longer
explains why that is safe.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1851,6 +1851,18 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val LEGACY_INFER_FAILABLE_FILTERS_FROM_CONSTRAINTS =
+ buildConf("spark.sql.legacy.inferFailableFiltersFromConstraints")
+ .internal()
+ .doc("When true, the query optimizer also infers a filter from data
constraints when " +
+ "evaluating it can fail or is otherwise observable, such as `10 / x >
0` from " +
+ "`x = y AND 10 / y > 0`, or a UDF. Such a filter runs on rows that the
query does not " +
+ "evaluate the expression on, so the query can raise an error that it
does not raise " +
+ "without the filter. When false, only a filter that cannot fail is
inferred.")
+ .version("4.4.0")
Review Comment:
**Finding 6.** `SparkConfigBindingPolicySuite` requires every new config to
declare a binding policy, and its exceptions file is frozen. On this head:
```
build/sbt -Phive 'hive/testOnly *SparkConfigBindingPolicySuite'
- Config enforcement for bindingPolicy *** FAILED ***
spark.sql.legacy.inferFailableFiltersFromConstraints
(SparkConfigBindingPolicySuite.scala:66)
```
`NOT_APPLICABLE` matches the neighboring
`spark.sql.legacy.filterShortCircuitOnNullConjunct`. With it, the suite passes:
```suggestion
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
```
##########
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:
**Finding 8.** `transformUp` replaces the inner cast with its child before
it looks at the outer one, so the outer cast is judged by the stripped child's
type. Take `CAST(CAST(y AS FLOAT) AS INT)`. The inner `INT` to `FLOAT` cast is
stripped, so the outer cast reads as `INT` to `INT`. `Cast.canUpCast` accepts
that, and the constraint becomes `x > 0`. But `FLOAT` to `INT` overflows in
ANSI mode. With ANSI mode on and `c(x, y)` = {(2147483647, 1), (1, 1)}:
```sql
SELECT * FROM c WHERE x = y AND CAST(CAST(y AS FLOAT) AS INT) > 0
```
With constraint propagation off, it returns `[1, 1]`. On the base and on the
head it fails with `CAST_OVERFLOW`, with AQE on and off, because the inferred
`CAST(CAST(x AS FLOAT) AS INT) > 0` runs on the row `(2147483647, 1)`.
The top-down `stripCannotFail` in finding 7 judges each cast by its original
input, and with it the query returns `[1, 1]`. If this PR lands after #60119
and drops the cast stripping (finding 1), this goes away as well. Then
`canEvaluateUnconditionally` sees each cast with its real input.
##########
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:
**Finding 9.** The description says that a constant that fails to evaluate
does not become a filter. That rejection fixes a literal binding that fails.
With ANSI mode on and `lb(x, y)` = {(1, 1)}:
```sql
SELECT * FROM lb a JOIN lb b ON a.x = b.x AND b.y = 10 / a.y WHERE a.y = 0
```
The binding `a.y = 0` turns the join condition into `b.y = 10 / 0` for `b`.
The base infers that filter and fails with `DIVIDE_BY_ZERO`, although no row
has `a.y = 0`. The head returns no rows. No test pins this, though. With
`isConditionalBranch = false` in `canBeFilter`,
`InferFiltersFromConstraintsSuite` and this test still pass, while the query
above fails again. This assertion in this test passes on the head, and fails
with `DIVIDE_BY_ZERO` under that change:
```scala
checkAnswer(
df.as("a").join(df.as("b"), $"a.x" === $"b.x" && $"b.y" ===
lit(10L) / $"a.y")
.where($"a.y" === 0),
Seq.empty)
```
--
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]
