HyukjinKwon commented on code in PR #59258:
URL: https://github.com/apache/spark/pull/59258#discussion_r4270130044
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala:
##########
@@ -126,6 +128,229 @@ trait ConstraintHelper {
inferred -- constraints
}
+ /**
+ * Infers the full additional constraint set for a given set of constraints
by combining the
+ * equality-based inferences above ([[inferAdditionalConstraints]],
+ * [[inferConstraintsFromLiteralBindings]]) with transitive inequality
chaining
+ * ([[inferInequalityConstraints]]). The inequality pass must see the
equality passes' output
+ * to chain through equality-derived edges and bounds. Because the
inequality pass emits only
+ * inequalities, the equality-based passes gain nothing from re-running on
its result, so a
Review Comment:
**Nit (P3):** This says the equality passes gain nothing from re-running on
the inequality output, but that doesn't hold for multi-hop equality chains.
`inferAdditionalConstraints` does one substitution round, and the inequality
pass drops every `EqualTo` without a safe literal side. Take `a = c AND c = d
AND d < b AND b < 5`. One round adds only `a = d` and `c < b`, so the
inequality pass emits `d < 5` and `c < 5` but never `a < 5`. Re-running the
equality pass on `c < 5` with `a = c` would derive it. Production runs Infer
Filters once, so `a < 5` is never added as a filter. Results stay correct; only
that pushdown is missed.
The `validComparisons` comment at lines 240-241 has the same issue from the
other side. It says a non-literal `EqualTo` is excluded "since no inference
path below handles a non-literal equality operand". But the canonical step
already expands `EqualTo` into both orientations, and the edge loop matches any
`BinaryComparison`. An admitted `a + 1 = b` would therefore become non-strict
edges that soundly carry a bound across. So the exclusion is a choice, not a
limit of the code below.
The "gain nothing" wording came from my suggestion in the previous round,
and it was wrong. Could both comments describe the actual boundary instead? One
round of the equality passes is followed by one inequality pass, and
non-literal equalities are deliberately not used as chaining edges, so
multi-hop equality chains can leave a bound uninferred.
**Verification:**
- **Inspection:** Both comments match the code. One equality substitution
round precedes one inequality pass, a two-hop chain such as a = c, c = d, d <
b, b < 5 does not infer a < 5, and excluding non-literal EqualTo is described
as a design choice consistent with the literal-bounds-only scope.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala:
##########
@@ -126,6 +128,229 @@ trait ConstraintHelper {
inferred -- constraints
}
+ /**
+ * Infers the full additional constraint set for a given set of constraints
by combining the
+ * equality-based inferences above ([[inferAdditionalConstraints]],
+ * [[inferConstraintsFromLiteralBindings]]) with transitive inequality
chaining
+ * ([[inferInequalityConstraints]]). The inequality pass must see the
equality passes' output
+ * to chain through equality-derived edges and bounds. Because the
inequality pass emits only
+ * inequalities, the equality-based passes gain nothing from re-running on
its result, so a
+ * single downstream call to [[inferInequalityConstraints]] suffices without
interleaving.
+ */
+ def inferAllAdditionalConstraints(constraints: ExpressionSet): ExpressionSet
= {
+ val equalityInferred =
+ inferAdditionalConstraints(constraints) ++
inferConstraintsFromLiteralBindings(constraints)
+ equalityInferred ++ inferInequalityConstraints(constraints ++
equalityInferred)
+ }
+
+ /**
+ * True when `e` cannot raise an error during evaluation: a non-null
literal, or an upcast of
+ * one admitted by [[UpCastRule.canUpCast]]. Used by
[[inferInequalityConstraints]] to decide
+ * which literals are safe to treat as chain anchors without risking an ANSI
eval failure.
+ * Arbitrary expressions (such as a parsing cast `CAST('abc' AS INT)` or
arithmetic that can
+ * fail under ANSI) are excluded.
+ */
+ @tailrec
+ private def isSafeBoundLiteral(e: Expression): Boolean = e match {
+ case l: Literal => l.value != null
+ case Cast(child, toType, _, _) =>
+ UpCastRule.canUpCast(child.dataType, toType) && isSafeBoundLiteral(child)
+ case _ => false
+ }
+
+ private def hasBoundOperand(e: Expression): Boolean = e match {
+ case b: BinaryComparison => isSafeBoundLiteral(b.left) ||
isSafeBoundLiteral(b.right)
+ case _ => false
+ }
+
+ /**
+ * True when `e` is safe to copy into a newly inferred, single-side filter
produced by
+ * [[inferInequalityConstraints]] without risking an error on rows that
never evaluated it
+ * before: an attribute reference, or an upcast of one admitted by
[[UpCastRule.canUpCast]].
+ * While some upcasts may lose precision (e.g. Long to Float/Double),
relocation is sound
+ * because they cannot fail and reuse the existing Cast expression.
Arbitrary expressions -- a
+ * parsing/narrowing cast, arithmetic that can overflow under ANSI, etc. --
are excluded, since
+ * the comparison they came from may have been a cross-side join condition
that only ever ran
+ * on matched rows.
+ */
+ private def isSafeToRelocate(e: Expression): Boolean = e match {
+ case _: Attribute => true
+ case Cast(a: Attribute, toType, _, _) => UpCastRule.canUpCast(a.dataType,
toType)
+ case _ => false
+ }
+
+ /**
+ * A known literal bound discovered while chaining inequality constraints in
+ * [[inferInequalityConstraints]]: `site {<,<=} lit` when `upper`, or `lit
{<,<=} site`
+ * otherwise. `value` stores the evaluated literal value to avoid repeatedly
evaluating `lit`
+ * during optimization.
+ */
+ private case class InequalityBound(
+ upper: Boolean,
+ site: Expression,
+ lit: Expression,
+ strict: Boolean,
+ value: Any)
+
+ /**
+ * Converts a canonicalized comparison into an [[InequalityBound]] when
exactly one side is a
+ * safe literal bound ([[isSafeBoundLiteral]]); `None` for
attribute-to-attribute comparisons.
+ */
+ private def toBound(e: Expression): Option[InequalityBound] = e match {
+ case b: BinaryComparison =>
+ val strict = b.isInstanceOf[LessThan]
+ if (isSafeBoundLiteral(b.right) && !isSafeBoundLiteral(b.left)) {
+ Some(InequalityBound(upper = true, b.left, b.right, strict,
b.right.eval()))
+ } else if (isSafeBoundLiteral(b.left) && !isSafeBoundLiteral(b.right)) {
+ Some(InequalityBound(upper = false, b.right, b.left, strict,
b.left.eval()))
+ } else {
+ None
+ }
+ case _ => None
+ }
+
+ /**
+ * Infers tightest literal inequality bounds on relocatable sites by
transitively chaining
+ * inequality constraints with literal bounds. For e.g., if an operator has
constraints of the
+ * form (`a > b`, `b > 5`), this returns an additional constraint of the
form `a > 5`; a longer
+ * chain such as `a < b && b < c && c < 5` derives both `b < 5` and `a < 5`
in this same call.
+ * Note that attribute-to-attribute inequalities without literal bounds
(e.g., `a < c` from
+ * `a < b && b < c`) are not inferred.
+ *
+ * This is computed as a worklist over a directed bound graph rather than a
repeated fixed point
+ * over the full candidate set: each candidate comparison becomes a directed
edge once, and each
+ * newly discovered bound is propagated only along its adjacent edges,
instead of rescanning the
+ * full candidates x bounds product after every newly inferred bound (which
costs O(n^3) on a
+ * sparse n-element chain such as `x1 < x2 < ... < xn < lit`). A site only
ever keeps its single
+ * tightest bound per direction, so reaching it via multiple literals (e.g.
`b < 5` and `b < 10`
+ * both reaching `a` through `a < b`) converges to one inferred predicate
instead of one per
+ * literal; a relaxation only re-enters the worklist when it strictly
tightens that site's bound,
+ * which bounds the number of relaxations per site by the number of distinct
literals that can
+ * reach it and guarantees termination.
+ */
+ private def inferInequalityConstraints(constraints: ExpressionSet):
ExpressionSet = {
+ // Admits only comparisons between binary-stable types. This is a
conservative guard, not a
+ // strict requirement: chaining never substitutes a value the way
+ // inferConstraintsFromLiteralBindings does, it only reuses an operand
under the same
+ // collation-aware comparison, so a non-binary-stable collation could in
principle still order
+ // consistently with its binary form here. EqualTo is admitted only when
at least one side is
+ // a safe literal bound; every other EqualTo -- attribute-to-attribute
(handled by
+ // inferAdditionalConstraints's transitivity), as well as any other shape
without a safe
+ // literal bound such as `a + 1 = b` -- is excluded, since no inference
path below handles a
+ // non-literal equality operand.
+ val validComparisons = constraints.toSeq.filter {
+ case b: BinaryComparison =>
+ b.deterministic && !b.throwable && !b.isInstanceOf[EqualNullSafe] &&
Review Comment:
**Nit (P3):** `b.deterministic` is the only thing that keeps the pass from
chaining through a non-deterministic operand. `Filter.validConstraints` keeps
non-deterministic conjuncts, and the later `c.deterministic` filters only drop
non-deterministic outputs. But no test covers this check. Without it, `WHERE a
< rand(0) AND rand(0) < 0.5` on an int column would infer the deterministic
`cast(a as double) < 0.5`. Both `rand(0)` instances canonicalize to the same
edge key, so the bound on one is chained through the other, even though they
are evaluated separately and the original filter doesn't imply it. Every added
and existing test still passes with the check removed, because the existing
rand tests use `a = rand(0)` or arithmetic sites that the new pass never
chains. A small test next to the rewritten throwable one, asserting that this
filter infers no bound on `a`, would pin it.
**Verification:**
- **Regression:** A literal-bounded chain through a non-deterministic
operand infers no deterministic bound on the relocatable attribute, and
removing the deterministic admission check makes the assertion fail.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala:
##########
@@ -126,6 +128,189 @@ trait ConstraintHelper {
inferred -- constraints
}
+ /**
+ * Infers the full additional constraint set for a given set of constraints
by combining the
+ * equality-based inferences above ([[inferAdditionalConstraints]],
+ * [[inferConstraintsFromLiteralBindings]]) with transitive inequality
chaining
+ * ([[inferInequalityConstraints]]). The two equality-based passes never
produce a new literal
Review Comment:
Thanks for updating this. The wording I suggested doesn't fully hold,
though. With `a = c AND c = d AND d < b AND b < 5`, one equality round only
adds `a = d` and `c < b`, so the inequality pass emits `d < 5` and `c < 5` but
never `a < 5`, which re-running the equality pass on its output would derive. I
left a separate comment with a suggested rewording.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4263387004","thread_id":"inline:4263387004","verdict_sha256":"504ace95b97cc57b766e13a0b73e622f081c281a8659d4632255b64ca45c5d35"}
-->
--
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]
