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. The
doc of `spark.sql.legacy.inferFailableFiltersFromConstraints` says "only a
filter that cannot fail is inferred", which isn't true for these either.
--
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]
