dongjoon-hyun commented on code in PR #60162:
URL: https://github.com/apache/spark/pull/60162#discussion_r4495936381
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala:
##########
@@ -313,17 +313,61 @@ class InferFiltersFromConstraintsSuite extends PlanTest {
Seq(Some("left.a".attr === "right.b".attr.cast(IntegerType)),
Some("right.b".attr.cast(IntegerType) === "left.a".attr)).foreach {
condition =>
- testConstraintsAfterJoin(
- originalLeft,
- originalRight,
- left,
- testRelation2.where(IsNotNull($"b") && $"b".attr.cast(IntegerType) ===
1)
- .subquery("right"),
- Inner,
- condition)
+ // A narrowing cast can fail, so it is not inferred.
+ testConstraintsAfterJoin(originalLeft, originalRight, left, right,
Inner, condition)
}
}
+ private def failable(e: Expression): Expression = Literal(10.0) / e > 0.0
Review Comment:
Done in 9759a107a1a. `failable` now builds `Divide(Literal(10.0), e,
EvalMode.ANSI)`, and the narrowing-cast case runs with ANSI mode on, so the
suite no longer depends on the session mode.
##########
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)
Review Comment:
Thanks for the repro. I ran it and get the same result: `DIVIDE_BY_ZERO` on
the head, and `(1, 2, 1, 6)` with `PushExtraPredicateThroughJoin` excluded.
This PR does not touch that rule, so I will file a separate JIRA for it rather
than widen this change.
##########
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 = {
Review Comment:
Thanks for working this out in detail. I would leave it out of this PR. It
only gates the two places that turn a constraint into a filter, and a TRY copy
with `IS NOT FALSE` adds a second kind of inferred filter to a subtle rule. The
counter-arguments you list also hold for me: data source pushdown does not come
back, `IntegralDivide` and `Pmod` still raise in TRY mode, and I have not
prototyped or measured it either. Once #60119 is in, the remaining cost is
limited to the constraints that can fail only in ANSI mode, and I would revisit
it with a measurement on those.
##########
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:
Thanks, reproduced. Fixed in 9759a107a1a: the test body runs with
`spark.sql.legacy.inferFailableFiltersFromConstraints` set to `true`, and its
leading comment says that a Hive UDF is only copied with the legacy
configuration. Both SPARK-58792 tests pass.
##########
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:
Thanks, fixed in 9759a107a1a with `NOT_APPLICABLE`, as on the neighboring
legacy config. I had not run the `hive` module tests, and
`SparkConfigBindingPolicySuite` passes now.
##########
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:
Thanks, reproduced: the correlated `IN` query fails with `DIVIDE_BY_ZERO` on
the base and on the head, and returns `[6, null]` with constraint propagation
off.
Fixed in 9759a107a1a along the lines of your suggestion. `stripCannotFail`
judges an `InSubquery` by its values and its `joinCond`, an `Exists` by its
`joinCond`, and treats a `ScalarSubquery` as a constant only when it is not
correlated. The comment now says that a subquery counts by its correlation
condition. A new `DataFrameSuite` test runs your `LEFT JOIN` query and expects
`(6, null)`.
I did not add a test for the `EXISTS` semi-join. It only changes the plan,
not the rows, and I could not observe the difference in a query result.
--
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]
