peter-toth commented on code in PR #60162:
URL: https://github.com/apache/spark/pull/60162#discussion_r4490459278
##########
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:
**Finding 1.** `canEvaluateUnconditionally` is a short allowlist, so
`canBeFilter` also rejects expressions that can't fail. I ran these joins on
the base and the head as sort-merge joins, with AQE on and off. `l` and `ld`
are partitioned by the join key:
| Query | Base | Head |
|---|---|---|
| `l JOIN r ON l.s = r.s WHERE r.s LIKE 'a%'` | `l` partition filters:
`StartsWith(s, a)`, `isnotnull(s)` | `isnotnull(s)` only |
| `ld JOIN rd ON ld.d = rd.d WHERE year(rd.d) = 2024` | `ld` partition
filters: `year(d) = 2024`, `isnotnull(d)` | `isnotnull(d)` only |
| `li JOIN ri ON li.i = ri.i WHERE ri.i + 1 > 5`, ANSI off | `li` data
filters: `(i + 1) > 5`, `isnotnull(i)` | `isnotnull(i)` only |
`StartsWith` and `year` can't fail in any mode, and the addition only wraps
with ANSI mode off. So the head scans every partition of `l` and `ld` with no
correctness gain. DPP does not recover it here. The plans have
`dynamicpruningexpression(true)`, because a sort-merge join has no broadcast to
reuse.
#60119 adds these expressions, and the casts that can't fail, to
`canEvaluateUnconditionally`. With its `ExprUtils.scala` on top of this PR, all
three rows match the base again, and `InferFiltersFromConstraintsSuite` still
passes with ANSI mode on (finding 2 covers ANSI mode off).
Could this PR land after #60119? Then `isNumericUpCast` is not needed
either, because #60119 accepts the numeric widening casts itself. If this PR
lands first, the description and the migration guide should not list `LIKE
'a%'` among the filters that can fail. The guide also says "arithmetic ... in
ANSI mode", but arithmetic is no longer inferred with ANSI mode off either.
##########
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:
**Finding 2.** `Literal(10.0) / e` takes its eval mode from the session, so
it can fail only with ANSI mode on. The same holds for the narrowing cast in
"filter lower data type". Today the tests pass in both modes only because
`canEvaluateUnconditionally` rejects every division and cast. With #60119's
`ExprUtils.scala` on top of this PR, `SPARK_ANSI_SQL_MODE=false` fails three of
them:
- `Constraints shouldn't be inferred from cast equality constraint(filter
lower data type)`
- `SPARK-60985: a constraint that can fail is not inferred as a filter`
- `SPARK-60985: a constraint that can fail is not inferred on the other side
of a join`
With #60119, pinning ANSI mode makes the suite pass in both modes (32 of 32):
```suggestion
private def failable(e: Expression): Expression =
Divide(Literal(10.0), e, EvalMode.ANSI) > 0.0
```
And in "filter lower data type":
```scala
// A narrowing cast can fail in ANSI mode, so it is not inferred.
withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
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, right,
Inner, condition)
}
}
```
##########
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:
**Finding 3.** `PushExtraPredicateThroughJoin`
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushExtraPredicateThroughJoin.scala`)
copies the part of a disjunctive join condition that reads one side to that
side. So it has the same problem. With ANSI mode on, `l` has the row `(1, 2)`
and `r` has the rows `(1, 6)` and `(2, 0)`:
```sql
SELECT * FROM l JOIN r ON l.k = r.k AND ((l.a = 1 AND 10 / r.b > 0) OR (l.a
= 2 AND r.b > 5));
```
On the head it fails with `DIVIDE_BY_ZERO`. `r` gets `(10.0 / b > 0.0) OR (b
> 5)`, and `(2, 0)` matches no row of `l`. With the rule excluded, the query
returns `(1, 2, 1, 6)`. This PR doesn't change that rule, and I found no ticket
for it. Could you file one?
--
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]
