wangyum commented on code in PR #59258:
URL: https://github.com/apache/spark/pull/59258#discussion_r4268719887
##########
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
+ * binding (an `attr = literal` equality), only new inequalities or
substitutions into existing
+ * predicates, so a single downstream call to [[inferInequalityConstraints]]
sees every literal
+ * bound they can ever contribute; it does not need to be interleaved with
them.
+ */
+ def inferAllAdditionalConstraints(constraints: ExpressionSet): ExpressionSet
= {
+ val equalityInferred =
+ inferAdditionalConstraints(constraints) ++
inferConstraintsFromLiteralBindings(constraints)
+ equalityInferred ++ inferInequalityConstraints(constraints ++
equalityInferred)
+ }
+
+ /**
+ * Infers the full set of constraints derivable by transitively chaining
inequality constraints.
+ * 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.
+ *
+ * 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
+ // foldable (a literal bound); every other EqualTo --
attribute-to-attribute (handled by
+ // inferAdditionalConstraints's transitivity), as well as any other shape
without a foldable
+ // side such as `a + 1 = b` -- is excluded, since no inference path below
handles a
+ // non-foldable equality operand.
+ val validComparisons = constraints.toSeq.filter {
+ case b: BinaryComparison =>
+ b.deterministic && !b.throwable && !b.isInstanceOf[EqualNullSafe] &&
+ (!b.isInstanceOf[EqualTo] || b.left.foldable || b.right.foldable) &&
+ isBinaryStable(b.left.dataType) && isBinaryStable(b.right.dataType)
+ case _ => false
+ }
+
+ def hasFoldableOperand(e: Expression): Boolean = e match {
+ case b: BinaryComparison => b.left.foldable || b.right.foldable
+ case _ => false
+ }
+
+ if (!validComparisons.exists(hasFoldableOperand)) {
+ return ExpressionSet()
+ }
+
+ // Canonicalize comparisons into LessThan / LessThanOrEqual direction.
EqualTo is expanded
+ // into both directions, unlike GreaterThan/GreaterThanOrEqual, because an
equality acts as
+ // both an upper and a lower bound and the matching below is
direction-sensitive on which
+ // side carries the literal. Kept as a Seq, not an ExpressionSet:
ExpressionSet canonicalizes
+ // EqualTo(l, r) and EqualTo(r, l) to the same entry and would silently
keep only one of the
+ // two directions just expanded.
+ val canonical = validComparisons.flatMap {
+ case GreaterThan(l, r) => LessThan(r, l) :: Nil
+ case GreaterThanOrEqual(l, r) => LessThanOrEqual(r, l) :: Nil
+ case EqualTo(l, r) => EqualTo(l, r) :: EqualTo(r, l) :: Nil
+ case other => other :: Nil
+ }
+
+ val (initialBounds, candidates) = canonical.partition(hasFoldableOperand)
+ if (initialBounds.isEmpty || candidates.isEmpty) return ExpressionSet()
+
+ // A candidate operand is only safe to copy into a newly inferred,
single-side filter when
+ // doing so cannot raise an error on rows that never evaluated it before:
an attribute
+ // reference, or a lossless (non-truncating, non-failing) widening cast of
one. 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.
+ def isSafeToRelocate(e: Expression): Boolean = e match {
+ case _: Attribute => true
+ case Cast(a: Attribute, toType, _, _) =>
UpCastRule.canUpCast(a.dataType, toType)
+ case _ => false
+ }
+
+ // Bound represents a known literal bound on `site`: `site {<,<=} lit`
when `upper`, or
+ // `lit {<,<=} site` otherwise.
+ case class Bound(upper: Boolean, site: Expression, lit: Expression,
strict: Boolean)
+
+ def toBound(e: Expression): Option[Bound] = e match {
+ case b: BinaryComparison =>
+ val strict = b.isInstanceOf[LessThan]
+ if (b.right.foldable) Some(Bound(upper = true, b.left, b.right,
strict))
Review Comment:
Addressed. Instead of swallowing runtime exceptions with try-catch NonFatal,
we statically enforce bound safety via isSafeBoundLiteral, allowing only
non-null Literals or valid recursive UpCasts. Evaluated literal values are
stored on InequalityBound.value (line 188-210), and isTighter (line 302-312)
compares them directly.
--
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]
