wangyum commented on code in PR #59258:
URL: https://github.com/apache/spark/pull/59258#discussion_r4272953577
##########
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:
Updated the doc
##########
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:
add the test
--
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]
