HyukjinKwon commented on code in PR #59258:
URL: https://github.com/apache/spark/pull/59258#discussion_r4225050761
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala:
##########
@@ -126,6 +125,92 @@ 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 step is iterated to a
fixed point because a
+ * single pass only chains pairs present in the input set: e.g. `a < b && b
< c && c < 5`
+ * requires one pass to derive `b < 5` and a second to derive `a < 5` from
it.
+ */
+ def inferAllAdditionalConstraints(constraints: ExpressionSet): ExpressionSet
= {
+ @tailrec
+ def chainInequalities(inferred: ExpressionSet): ExpressionSet = {
+ val newlyInferred = inferInequalityConstraints(constraints ++ inferred)
+ if (newlyInferred.isEmpty) inferred else chainInequalities(inferred ++
newlyInferred)
+ }
+
+ chainInequalities(
+ inferAdditionalConstraints(constraints) ++
+ inferConstraintsFromLiteralBindings(constraints))
+ }
+
+ /**
+ * Infers an additional set of constraints from a given set of 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`.
+ */
+ private def inferInequalityConstraints(constraints: ExpressionSet):
ExpressionSet = {
+ // Substituting through a comparison is only safe under binary-stable
collations, for the
+ // same reason as in inferConstraintsFromLiteralBindings:
non-binary-stable collations may
+ // order values differently than their binary representation would
suggest. Attribute-to-
+ // attribute equality is excluded here too: its transitivity is already
handled by
+ // inferAdditionalConstraints via substitution, so chaining it here would
only re-derive the
+ // same fact as a redundant <= / >= pair instead of the clean substituted
equality that path
+ // already produces.
+ val validComparisons = constraints.filter {
+ case b: BinaryComparison =>
+ b.deterministic && !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 BinaryComparison(l, r) => l.foldable || r.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.
+ val canonical = validComparisons.toSet[Expression].flatMap {
Review Comment:
**Nit (P3):** `toSet[Expression]`, and the `partition` and for-comprehension
that follow, switch from the insertion-ordered `ExpressionSet` to Scala `Set`s.
Those become hash-ordered `HashSet`s once there are 5 or more elements. Since
`AttributeReference.hashCode` includes the exprId, the order of the inferred
conjuncts can change between runs of the same query whenever one pass infers
two or more bounds. That order carries into EXPLAIN, `PushedFilters`, and
generated code. An example is `a < c AND b < c AND c < 10 AND a > 0 AND b > 0`.
Plan-stability goldens normalize exprIds but not conjunct order, so such a
query would make them flaky. Keeping these as `Seq`s (e.g.
`validComparisons.toSeq.flatMap`) would preserve the input order, as the other
inference helpers do.
**Verification:**
- **Regression:** A Filter with at least five chainable comparisons that
infers two or more bounds in one pass yields the same inferred-conjunct order
when its attributes are re-created with different exprIds, checked without
order-normalizing the plan.
- **Inspection:** inferInequalityConstraints iterates only insertion-ordered
collections between validComparisons and the returned ExpressionSet.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala:
##########
@@ -126,6 +125,92 @@ 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 step is iterated to a
fixed point because a
+ * single pass only chains pairs present in the input set: e.g. `a < b && b
< c && c < 5`
+ * requires one pass to derive `b < 5` and a second to derive `a < 5` from
it.
+ */
+ def inferAllAdditionalConstraints(constraints: ExpressionSet): ExpressionSet
= {
+ @tailrec
+ def chainInequalities(inferred: ExpressionSet): ExpressionSet = {
+ val newlyInferred = inferInequalityConstraints(constraints ++ inferred)
+ if (newlyInferred.isEmpty) inferred else chainInequalities(inferred ++
newlyInferred)
+ }
+
+ chainInequalities(
+ inferAdditionalConstraints(constraints) ++
+ inferConstraintsFromLiteralBindings(constraints))
+ }
+
+ /**
+ * Infers an additional set of constraints from a given set of 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`.
+ */
+ private def inferInequalityConstraints(constraints: ExpressionSet):
ExpressionSet = {
+ // Substituting through a comparison is only safe under binary-stable
collations, for the
+ // same reason as in inferConstraintsFromLiteralBindings:
non-binary-stable collations may
+ // order values differently than their binary representation would
suggest. Attribute-to-
+ // attribute equality is excluded here too: its transitivity is already
handled by
+ // inferAdditionalConstraints via substitution, so chaining it here would
only re-derive the
+ // same fact as a redundant <= / >= pair instead of the clean substituted
equality that path
+ // already produces.
+ val validComparisons = constraints.filter {
+ case b: BinaryComparison =>
+ b.deterministic && !b.isInstanceOf[EqualNullSafe] &&
Review Comment:
**Nit (P3):** `validComparisons` checks `deterministic` but not `throwable`,
so a comparison whose operand can throw is still chained.
`InferFiltersFromConstraints` then materializes the derived bound where the
original predicate was never evaluated. For example, in `SELECT * FROM x JOIN y
ON x.k = y.k AND CASE WHEN y.v < 0 THEN raise_error('neg') ELSE y.v END < x.b
WHERE x.b < 10`, the inferred `CASE ... END < 10` is pushed as a Filter on `y`,
so an unmatched `y` row with `v < 0` now fails the query.
`PushPredicateThroughJoin` and `CombineFilters` keep throwable predicates in
place for this reason. Adding `!b.throwable` here, with a small test, would
keep the new chaining consistent with them. The literal-binding path already
has the same gap, so guarding where `inferNewFilter` materializes inferred
constraints would cover both.
**Verification:**
- **Regression:** An inner equi-join whose residual compares a
raise_error-containing CASE expression with a join attribute that has a literal
bound on its own side optimizes with no inferred Filter on the throwable side.
- **Compatibility:** The existing inequality-inference optimizer,
constraint-propagation, and q12 plan-stability expectations remain unchanged.
##########
sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala:
##########
@@ -5369,6 +5369,55 @@ class SQLQuerySuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
Seq(Row(0), Row(1), Row(2)))
}
}
+
+ test("SPARK-30768: inferred inequality chaining does not drop rows with NaN
operands") {
Review Comment:
**Nit (P3):** This test cannot fail for the case its name describes. Under
Spark's NaN-is-largest ordering, every NaN-bearing row already fails `a < b AND
b < 3.0`, so the expected `Row(1.0, 2.0)` does not depend on how the inferred
`a < 3.0` handles NaN. To pin NaN behavior, add a case where the original
predicate keeps a NaN row: for example `a > b AND b > 3.0` over `(Double.NaN,
5.0)`. Then assert that the row survives and that `a > 3.0` is inferred.
**Verification:**
- **Behavior:** With constraint propagation on and off, `a > b AND b > 3.0`
over data containing (NaN, 5.0) returns that row, and the inferred `a > 3.0` is
present only when propagation is enabled.
--
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]
