HyukjinKwon commented on code in PR #59258:
URL: https://github.com/apache/spark/pull/59258#discussion_r4263387021
##########
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)
Review Comment:
**Nit (P3):** This line lets chains start from edges or bindings that only
the equality passes create, but no added test depends on that. In one
`constraints` evaluation, `a = c AND c < b AND b < 5` gives `a < 5` only
because `inferAdditionalConstraints` rewrites `c < b` into `a < b`. Likewise `a
= 5 AND a = b AND c < b` gives `c < 5` only through the derived `b = 5`. If
this line became `inferInequalityConstraints(constraints)`, both bounds would
disappear and every added test would still pass.
InferFiltersFromConstraintsSuite can't catch it, because its FixedPoint(100)
batch re-derives the bound on a later iteration, while production runs 'Infer
Filters' Once. A ConstraintPropagationSuite case asserting those two bounds on
`plan.constraints` would pin it.
See **Shared repair plan 1** in the review body.
##########
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:
**Nit (P3):** `toBound` accepts any foldable operand as the bound's `lit`.
The worklist then copies that `lit` into the inferred predicate on the
relocated site, and `isTighter` calls `eval()` on it during optimization.
`isSafeToRelocate` checks only the edge endpoint, and `!b.throwable` does not
flag ANSI casts or arithmetic. `inferConstraintsFromLiteralBindings` runs in
the same call and can produce exactly such an operand. With ANSI on:
```sql
SELECT * FROM (SELECT * FROM x WHERE s = 'abc') x
LEFT JOIN y ON x.k = y.k AND y.c < x.b AND x.b < CAST(x.s AS INT)
```
Substitution turns the preserved-side ON conjunct into `x.b < CAST('abc' AS
INT)`. The `y.c < x.b` edge then relocates it to `y.c < CAST('abc' AS INT)` on
the right side, and ConstantFolding raises CAST_INVALID_INPUT during
optimization, even when no key-matched pair exists. Before this PR, that
left-only conjunct was never pushed for a LeftOuter join. If x's subquery also
has `b < 5`, `isTighter` throws inside InferFiltersFromConstraints itself. This
is the error-on-unevaluated-rows behavior the relocation guard was added to
rule out, reached through the bound operand instead of the edge endpoint. With
ANSI off the cast yields NULL and the query is unaffected.
Evaluating the foldable operand once in `toBound`, inside a `NonFatal`
guard, would close this: drop the bound if evaluation fails, and keep the
evaluated value on `Bound` so `isTighter` compares stored values. The emitted
predicate can keep its current operand expression, so existing plan
expectations stay the same.
**Verification:**
- **Regression:** Under ANSI, a left-outer join whose preserved-side ON
conjunct substitutes into a failing cast plans and returns the same rows with
constraint propagation on and off.
- **Regression:** A site bounded by both a valid literal and a failing
foldable is optimized without an exception and still infers the chained bound
from the valid literal.
- **Compatibility:** Existing inequality-inference, constraint-propagation,
and q12 plan-stability expectations remain unchanged.
##########
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:
**Nit (P3):** This says the equality-based passes never produce a new
literal binding. `inferAdditionalConstraints` documents exactly that a few
lines above: (`a = 5`, `a = b`) gives `b = 5`.
`inferConstraintsFromLiteralBindings` can also produce `b = 5 + 1` from `a = 5`
and `b = a + 1`. The code is correct only because line 143 passes `constraints
++ equalityInferred` to the inequality pass. A maintainer who trusts this
comment could call `inferInequalityConstraints(constraints)` directly and
silently lose chains such as (`a = 5`, `a = b`, `c < b`) => `c < 5`.
Could the comment give the actual reason? The inequality pass has to see the
equality passes' output, and because it emits only inequalities, the equality
passes gain nothing from re-running on its result.
See **Shared repair plan 1** in the review body.
##########
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.
Review Comment:
**Nit (P3):** This method doesn't infer the full transitive set. It emits
only `site op literal` bounds, keeps the single tightest one per direction and
site, and bounds only relocatable sites, so `a < c` from `a < b AND b < c` is
never produced. Separately, the `isSafeToRelocate` comment at line 208 calls
the admitted cast lossless, but `UpCastRule.canUpCast` also admits Long to
Float/Double and Timestamp to Long, which lose precision. Relocation is sound
because those casts can't fail and the same Cast expression is reused, not
because they are lossless. Could both comments state what the code actually
guarantees?
**Verification:**
- **Inspection:** The headline matches the literal-bound, tightest-per-site,
relocatable-site behavior, and the relocation comment matches what
UpCastRule.canUpCast admits.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala:
##########
@@ -465,4 +465,350 @@ class InferFiltersFromConstraintsSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}
+ test("infer inequality constraints: chained inequalities within a single
relation") {
+ Seq(($"a" < $"b" && $"b" < 3, $"a" < $"b" && $"b" < 3 && $"a" < 3),
+ ($"a" < $"b" && $"b" <= 3, $"a" < $"b" && $"b" <= 3 && $"a" < 3),
+ ($"a" < $"b" && $"b" === 3, $"a" < $"b" && $"b" === 3 && $"a" < 3),
+ ($"a" <= $"b" && $"b" < 3, $"a" <= $"b" && $"b" < 3 && $"a" < 3),
+ ($"a" <= $"b" && $"b" <= 3, $"a" <= $"b" && $"b" <= 3 && $"a" <= 3),
+ ($"a" <= $"b" && $"b" === 3, $"a" <= $"b" && $"b" === 3 && $"a" <= 3),
+ ($"a" > $"b" && $"b" > 3, $"a" > $"b" && $"b" > 3 && $"a" > 3),
+ ($"a" > $"b" && $"b" >= 3, $"a" > $"b" && $"b" >= 3 && $"a" > 3),
+ ($"a" > $"b" && $"b" === 3, $"a" > $"b" && $"b" === 3 && $"a" > 3),
+ ($"a" >= $"b" && $"b" > 3, $"a" >= $"b" && $"b" > 3 && $"a" > 3),
+ ($"a" >= $"b" && $"b" >= 3, $"a" >= $"b" && $"b" >= 3 && $"a" >= 3),
+ ($"a" >= $"b" && $"b" === 3, $"a" >= $"b" && $"b" === 3 && $"a" >= 3)
+ ).foreach {
+ case (filter, inferred) =>
+ val original = testRelation.where(filter)
+ val optimized = testRelation.where(IsNotNull($"a") && IsNotNull($"b")
&& inferred)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+ }
+
+ test("infer inequality constraints: chained inequality across an equi-join")
{
+ Seq(("left.b".attr < "right.b".attr, $"b" < 1, $"b" < 1),
+ ("left.b".attr < "right.b".attr, $"b" === 1, $"b" < 1),
+ ("left.b".attr < "right.b".attr, $"b" <= 1, $"b" < 1),
+ ("left.b".attr <= "right.b".attr, $"b" <= 1, $"b" <= 1),
+ ("left.b".attr <= "right.b".attr, $"b" === 1, $"b" <= 1),
+ ("left.b".attr > "right.b".attr, $"b" > 1, $"b" > 1),
+ ("left.b".attr > "right.b".attr, $"b" === 1, $"b" > 1),
+ ("left.b".attr > "right.b".attr, $"b" >= 1, $"b" > 1),
+ ("left.b".attr >= "right.b".attr, $"b" >= 1, $"b" >= 1),
+ ("left.b".attr >= "right.b".attr, $"b" === 1, $"b" >= 1)
+ ).foreach {
+ case (cond, filter, inferred) =>
+ val originalLeft = testRelation.subquery("left")
+ val originalRight = testRelation.where(filter).subquery("right")
+
+ val left =
+ testRelation.where(IsNotNull($"a") && IsNotNull($"b") &&
inferred).subquery("left")
+ val right =
+ testRelation.where(IsNotNull($"a") && IsNotNull($"b") &&
filter).subquery("right")
+ val condition = Some("left.a".attr === "right.a".attr && cond)
+ testConstraintsAfterJoin(originalLeft, originalRight, left, right,
Inner, condition)
+ }
+ }
+
+ test("infer inequality constraints: chained inequalities through an implicit
cast") {
+ Seq(($"a" < $"b" && $"b" < 3L, $"a".cast(LongType) < $"b" && $"b" < 3L &&
+ $"a".cast(LongType) < 3L),
+ ($"a" < $"b" && $"b" <= 3L, $"a".cast(LongType) < $"b" && $"b" <= 3L &&
+ $"a".cast(LongType) < 3L),
+ ($"a" < $"b" && $"b" === 3L, $"a".cast(LongType) < $"b" && $"b" === 3L &&
+ $"a".cast(LongType) < 3L),
+ ($"a" <= $"b" && $"b" < 3L, $"a".cast(LongType) <= $"b" && $"b" < 3L &&
+ $"a".cast(LongType) < 3L),
+ ($"a" <= $"b" && $"b" <= 3L, $"a".cast(LongType) <= $"b" && $"b" <= 3L &&
+ $"a".cast(LongType) <= 3L),
+ ($"a" <= $"b" && $"b" === 3L, $"a".cast(LongType) <= $"b" && $"b" === 3L
&&
+ $"a".cast(LongType) <= 3L),
+ ($"a" < $"b" && $"b" < 3, $"a".cast(LongType) < $"b" && $"b" <
Literal(3).cast(LongType)
+ && $"a".cast(LongType) < Literal(3).cast(LongType)),
+ ($"a" > $"b" && $"b" > 3L, $"a".cast(LongType) > $"b" && $"b" > 3L &&
+ $"a".cast(LongType) > 3L),
+ ($"a" > $"b" && $"b" >= 3L, $"a".cast(LongType) > $"b" && $"b" >= 3L &&
+ $"a".cast(LongType) > 3L),
+ ($"a" > $"b" && $"b" === 3L, $"a".cast(LongType) > $"b" && $"b" === 3L &&
+ $"a".cast(LongType) > 3L),
+ ($"a" >= $"b" && $"b" > 3L, $"a".cast(LongType) >= $"b" && $"b" > 3L &&
+ $"a".cast(LongType) > 3L),
+ ($"a" >= $"b" && $"b" >= 3L, $"a".cast(LongType) >= $"b" && $"b" >= 3L &&
+ $"a".cast(LongType) >= 3L),
+ ($"a" >= $"b" && $"b" === 3L, $"a".cast(LongType) >= $"b" && $"b" === 3L
&&
+ $"a".cast(LongType) >= 3L),
+ ($"a" > $"b" && $"b" > 3, $"a".cast(LongType) > $"b" && $"b" >
Literal(3).cast(LongType)
+ && $"a".cast(LongType) > Literal(3).cast(LongType))
+ ).foreach {
+ case (filter, inferred) =>
+ val testRelation = LocalRelation($"a".int, $"b".long)
+ val original = testRelation.where(filter)
+ val optimized = testRelation.where(IsNotNull($"a") && IsNotNull($"b")
&& inferred)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+ }
+
+ test("infer inequality constraints: an attr=literal binding on one join side
bounds " +
+ "an inequality join key on the other side") {
+ val condition = Some("x.a".attr > "y.a".attr)
+ val optimizedLeft = testRelation.where(IsNotNull($"a") && $"a" ===
1).as("x")
+ val optimizedRight = testRelation.where($"a" < 1 &&
IsNotNull($"a")).as("y")
+ val correct = optimizedLeft.join(optimizedRight, Inner, condition)
+
+ Seq(Literal(1) === $"a", $"a" === Literal(1)).foreach { filter =>
+ val original =
+ testRelation.where(filter).as("x").join(testRelation.as("y"), Inner,
condition)
+ comparePlans(Optimize.execute(original.analyze), correct.analyze)
+ }
+ }
+
+ test("infer inequality constraints: a three-attribute chain bounded by a
literal") {
+ val original = testRelation.where($"a" < $"b" && $"b" < $"c" && $"c" < 5)
+ val optimized = testRelation.where(IsNotNull($"a") && IsNotNull($"b") &&
IsNotNull($"c")
+ && $"a" < $"b" && $"b" < $"c" && $"a" < 5 && $"b" < 5 && $"c" < 5)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+
+ test("infer inequality constraints: a range join condition combined with a
range filter") {
+ val left = testRelation.where($"b" >= 3 && $"b" <= 13).as("x")
+ val right = testRelation.as("y")
+
+ val optimizedLeft = testRelation.where(IsNotNull($"a") && IsNotNull($"b")
+ && $"b" >= 3 && $"b" <= 13).as("x")
+ val optimizedRight = testRelation.where(IsNotNull($"a") && IsNotNull($"b")
&& IsNotNull($"c")
+ && $"c" > 3 && $"b" <= 13).as("y")
+ val condition = Some("x.a".attr === "y.a".attr
+ && "x.b".attr >= "y.b".attr && "x.b".attr < "y.c".attr)
+ val original = left.join(right, Inner, condition)
+ val optimized = optimizedLeft.join(optimizedRight, Inner, condition)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+
+ test("infer inequality constraints: chain across a join through an implicit
cast") {
+ val testRelation1 = LocalRelation($"a".long, $"b".long, $"c".long).as("x")
+ val testRelation2 = LocalRelation($"a".int, $"b".int, $"c".int).as("y")
+
+ // y.b < 13 inferred from y.b < x.b && x.b <= 13
+ val left = testRelation1.where($"b" <= 13L).as("x")
+ val right = testRelation2.as("y")
+
+ val optimizedLeft =
+ testRelation1.where(IsNotNull($"a") && IsNotNull($"b") && $"b" <=
13L).as("x")
+ val optimizedRight = testRelation2.where(IsNotNull($"a") && IsNotNull($"b")
+ && $"b".cast(LongType) < 13L).as("y")
+
+ val condition = Some("x.a".attr === "y.a".attr && "y.b".attr < "x.b".attr)
+ val original = left.join(right, Inner, condition)
+ val optimized = optimizedLeft.join(optimizedRight, Inner, condition)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+
+ test("infer inequality constraints: range predicate in left-outer join ON
clause " +
+ "is pushed to right side when left attr is bounded by inequality") {
+ // Left side: a.k >= 5
+ // Join condition: b.v >= a.k
+ // Transitive inference should deduce b.v >= 5 and push it down to the
right side of
+ // LeftOuter join.
+ val left = LocalRelation($"k".int, $"key".int).subquery("a")
+ val right = LocalRelation($"v".int, $"key".int).subquery("b")
+
+ val joinCond = ("a.key".attr === "b.key".attr) && ("b.v".attr >=
"a.k".attr)
+ val originalQuery = left
+ .where("a.k".attr >= 5)
+ .join(right, LeftOuter, Some(joinCond))
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+ var pushedToRight = false
+ optimized.foreach {
+ case Join(_, Filter(cond, _), LeftOuter, _, _) =>
+ pushedToRight = splitConjunctivePredicates(cond).exists {
+ case GreaterThanOrEqual(attr: Attribute, Literal(v: Int,
IntegerType)) =>
+ attr.name == "v" && v == 5
+ case _ => false
+ }
+ case _ =>
+ }
+ assert(pushedToRight, s"Expected b.v >= 5 pushed to right side; plan
was:\n$optimized")
+ }
+
+ test("SPARK-30768: do not infer filters from throwable expressions") {
+ val x = testRelation.where($"b" < 10).subquery("x")
+ val y = testRelation.subquery("y")
+ val raiseErr = new RaiseError(Literal("neg")).cast(IntegerType)
+ val caseExpr = CaseWhen(Seq(("y.b".attr < 0, raiseErr)), elseValue =
Some("y.b".attr))
+ val cond = Some("x.a".attr === "y.a".attr && caseExpr < "x.b".attr)
+ val query = x.join(y, Inner, cond).analyze
+ val optimized = Optimize.execute(query)
+ var inferredThrowable = false
+ optimized.foreach {
+ case Join(_, Filter(cond, _), _, _, _) =>
+ inferredThrowable = cond.throwable
+ case _ =>
+ }
+ assert(!inferredThrowable, s"Unexpected throwable filter condition
inferred: $optimized")
+ }
+
+ test("SPARK-30768: equality orientation does not affect inequality
chaining") {
+ Seq($"a" + 1 === 5, Literal(5) === ($"a" + 1)).foreach { eq =>
+ val original = testRelation.where(eq && ($"a" + 1) < $"b")
+ val optimized = testRelation.where(
+ IsNotNull($"a") && IsNotNull($"b") && eq && ($"a" + 1) < $"b" && $"b"
> 5)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+ }
+
+ test("SPARK-30768: equality orientation does not affect join-side inequality
pushdown") {
+ def hasPushedFilter(plan: LogicalPlan)(pred: Expression => Boolean):
Boolean = {
+ plan.collectFirst {
+ case Join(_, Filter(cond, _), _, _, _) =>
splitConjunctivePredicates(cond).exists(pred)
+ }.getOrElse(false)
+ }
+
+ Seq($"a" + 1 === 5, Literal(5) === ($"a" + 1)).foreach { eq =>
+ val x = testRelation.where(eq).subquery("x")
+ val y = testRelation.subquery("y")
+
+ // Upper bound on a + 1 propagates to y.b > 5
Review Comment:
**Nit (P3):** These two labels are reversed. For `(x.a + 1) < y.b`, `x.a +
1` is not relocatable, so the only edge is the lower edge keyed by `a + 1`, and
`y.b > 5` comes from the lower bound `5 <= a + 1`. For `y.b < (x.a + 1)` at
line 680, `y.b < 5` comes from the upper bound `a + 1 <= 5`. The assertions are
right. Swapping the two comments, and the `optUpper`/`optLower` names, would
make the test describe the direction it checks.
**Verification:**
- **Inspection:** The labels attribute `y.b > 5` to the lower bound and `y.b
< 5` to the upper bound on `a + 1`, and the assertions are unchanged.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala:
##########
@@ -465,4 +465,350 @@ class InferFiltersFromConstraintsSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}
+ test("infer inequality constraints: chained inequalities within a single
relation") {
+ Seq(($"a" < $"b" && $"b" < 3, $"a" < $"b" && $"b" < 3 && $"a" < 3),
+ ($"a" < $"b" && $"b" <= 3, $"a" < $"b" && $"b" <= 3 && $"a" < 3),
+ ($"a" < $"b" && $"b" === 3, $"a" < $"b" && $"b" === 3 && $"a" < 3),
+ ($"a" <= $"b" && $"b" < 3, $"a" <= $"b" && $"b" < 3 && $"a" < 3),
+ ($"a" <= $"b" && $"b" <= 3, $"a" <= $"b" && $"b" <= 3 && $"a" <= 3),
+ ($"a" <= $"b" && $"b" === 3, $"a" <= $"b" && $"b" === 3 && $"a" <= 3),
+ ($"a" > $"b" && $"b" > 3, $"a" > $"b" && $"b" > 3 && $"a" > 3),
+ ($"a" > $"b" && $"b" >= 3, $"a" > $"b" && $"b" >= 3 && $"a" > 3),
+ ($"a" > $"b" && $"b" === 3, $"a" > $"b" && $"b" === 3 && $"a" > 3),
+ ($"a" >= $"b" && $"b" > 3, $"a" >= $"b" && $"b" > 3 && $"a" > 3),
+ ($"a" >= $"b" && $"b" >= 3, $"a" >= $"b" && $"b" >= 3 && $"a" >= 3),
+ ($"a" >= $"b" && $"b" === 3, $"a" >= $"b" && $"b" === 3 && $"a" >= 3)
+ ).foreach {
+ case (filter, inferred) =>
+ val original = testRelation.where(filter)
+ val optimized = testRelation.where(IsNotNull($"a") && IsNotNull($"b")
&& inferred)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+ }
+
+ test("infer inequality constraints: chained inequality across an equi-join")
{
+ Seq(("left.b".attr < "right.b".attr, $"b" < 1, $"b" < 1),
+ ("left.b".attr < "right.b".attr, $"b" === 1, $"b" < 1),
+ ("left.b".attr < "right.b".attr, $"b" <= 1, $"b" < 1),
+ ("left.b".attr <= "right.b".attr, $"b" <= 1, $"b" <= 1),
+ ("left.b".attr <= "right.b".attr, $"b" === 1, $"b" <= 1),
+ ("left.b".attr > "right.b".attr, $"b" > 1, $"b" > 1),
+ ("left.b".attr > "right.b".attr, $"b" === 1, $"b" > 1),
+ ("left.b".attr > "right.b".attr, $"b" >= 1, $"b" > 1),
+ ("left.b".attr >= "right.b".attr, $"b" >= 1, $"b" >= 1),
+ ("left.b".attr >= "right.b".attr, $"b" === 1, $"b" >= 1)
+ ).foreach {
+ case (cond, filter, inferred) =>
+ val originalLeft = testRelation.subquery("left")
+ val originalRight = testRelation.where(filter).subquery("right")
+
+ val left =
+ testRelation.where(IsNotNull($"a") && IsNotNull($"b") &&
inferred).subquery("left")
+ val right =
+ testRelation.where(IsNotNull($"a") && IsNotNull($"b") &&
filter).subquery("right")
+ val condition = Some("left.a".attr === "right.a".attr && cond)
+ testConstraintsAfterJoin(originalLeft, originalRight, left, right,
Inner, condition)
+ }
+ }
+
+ test("infer inequality constraints: chained inequalities through an implicit
cast") {
+ Seq(($"a" < $"b" && $"b" < 3L, $"a".cast(LongType) < $"b" && $"b" < 3L &&
+ $"a".cast(LongType) < 3L),
+ ($"a" < $"b" && $"b" <= 3L, $"a".cast(LongType) < $"b" && $"b" <= 3L &&
+ $"a".cast(LongType) < 3L),
+ ($"a" < $"b" && $"b" === 3L, $"a".cast(LongType) < $"b" && $"b" === 3L &&
+ $"a".cast(LongType) < 3L),
+ ($"a" <= $"b" && $"b" < 3L, $"a".cast(LongType) <= $"b" && $"b" < 3L &&
+ $"a".cast(LongType) < 3L),
+ ($"a" <= $"b" && $"b" <= 3L, $"a".cast(LongType) <= $"b" && $"b" <= 3L &&
+ $"a".cast(LongType) <= 3L),
+ ($"a" <= $"b" && $"b" === 3L, $"a".cast(LongType) <= $"b" && $"b" === 3L
&&
+ $"a".cast(LongType) <= 3L),
+ ($"a" < $"b" && $"b" < 3, $"a".cast(LongType) < $"b" && $"b" <
Literal(3).cast(LongType)
+ && $"a".cast(LongType) < Literal(3).cast(LongType)),
+ ($"a" > $"b" && $"b" > 3L, $"a".cast(LongType) > $"b" && $"b" > 3L &&
+ $"a".cast(LongType) > 3L),
+ ($"a" > $"b" && $"b" >= 3L, $"a".cast(LongType) > $"b" && $"b" >= 3L &&
+ $"a".cast(LongType) > 3L),
+ ($"a" > $"b" && $"b" === 3L, $"a".cast(LongType) > $"b" && $"b" === 3L &&
+ $"a".cast(LongType) > 3L),
+ ($"a" >= $"b" && $"b" > 3L, $"a".cast(LongType) >= $"b" && $"b" > 3L &&
+ $"a".cast(LongType) > 3L),
+ ($"a" >= $"b" && $"b" >= 3L, $"a".cast(LongType) >= $"b" && $"b" >= 3L &&
+ $"a".cast(LongType) >= 3L),
+ ($"a" >= $"b" && $"b" === 3L, $"a".cast(LongType) >= $"b" && $"b" === 3L
&&
+ $"a".cast(LongType) >= 3L),
+ ($"a" > $"b" && $"b" > 3, $"a".cast(LongType) > $"b" && $"b" >
Literal(3).cast(LongType)
+ && $"a".cast(LongType) > Literal(3).cast(LongType))
+ ).foreach {
+ case (filter, inferred) =>
+ val testRelation = LocalRelation($"a".int, $"b".long)
+ val original = testRelation.where(filter)
+ val optimized = testRelation.where(IsNotNull($"a") && IsNotNull($"b")
&& inferred)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+ }
+
+ test("infer inequality constraints: an attr=literal binding on one join side
bounds " +
+ "an inequality join key on the other side") {
+ val condition = Some("x.a".attr > "y.a".attr)
+ val optimizedLeft = testRelation.where(IsNotNull($"a") && $"a" ===
1).as("x")
+ val optimizedRight = testRelation.where($"a" < 1 &&
IsNotNull($"a")).as("y")
+ val correct = optimizedLeft.join(optimizedRight, Inner, condition)
+
+ Seq(Literal(1) === $"a", $"a" === Literal(1)).foreach { filter =>
+ val original =
+ testRelation.where(filter).as("x").join(testRelation.as("y"), Inner,
condition)
+ comparePlans(Optimize.execute(original.analyze), correct.analyze)
+ }
+ }
+
+ test("infer inequality constraints: a three-attribute chain bounded by a
literal") {
+ val original = testRelation.where($"a" < $"b" && $"b" < $"c" && $"c" < 5)
+ val optimized = testRelation.where(IsNotNull($"a") && IsNotNull($"b") &&
IsNotNull($"c")
+ && $"a" < $"b" && $"b" < $"c" && $"a" < 5 && $"b" < 5 && $"c" < 5)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+
+ test("infer inequality constraints: a range join condition combined with a
range filter") {
+ val left = testRelation.where($"b" >= 3 && $"b" <= 13).as("x")
+ val right = testRelation.as("y")
+
+ val optimizedLeft = testRelation.where(IsNotNull($"a") && IsNotNull($"b")
+ && $"b" >= 3 && $"b" <= 13).as("x")
+ val optimizedRight = testRelation.where(IsNotNull($"a") && IsNotNull($"b")
&& IsNotNull($"c")
+ && $"c" > 3 && $"b" <= 13).as("y")
+ val condition = Some("x.a".attr === "y.a".attr
+ && "x.b".attr >= "y.b".attr && "x.b".attr < "y.c".attr)
+ val original = left.join(right, Inner, condition)
+ val optimized = optimizedLeft.join(optimizedRight, Inner, condition)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+
+ test("infer inequality constraints: chain across a join through an implicit
cast") {
+ val testRelation1 = LocalRelation($"a".long, $"b".long, $"c".long).as("x")
+ val testRelation2 = LocalRelation($"a".int, $"b".int, $"c".int).as("y")
+
+ // y.b < 13 inferred from y.b < x.b && x.b <= 13
+ val left = testRelation1.where($"b" <= 13L).as("x")
+ val right = testRelation2.as("y")
+
+ val optimizedLeft =
+ testRelation1.where(IsNotNull($"a") && IsNotNull($"b") && $"b" <=
13L).as("x")
+ val optimizedRight = testRelation2.where(IsNotNull($"a") && IsNotNull($"b")
+ && $"b".cast(LongType) < 13L).as("y")
+
+ val condition = Some("x.a".attr === "y.a".attr && "y.b".attr < "x.b".attr)
+ val original = left.join(right, Inner, condition)
+ val optimized = optimizedLeft.join(optimizedRight, Inner, condition)
+ comparePlans(Optimize.execute(original.analyze), optimized.analyze)
+ }
+
+ test("infer inequality constraints: range predicate in left-outer join ON
clause " +
+ "is pushed to right side when left attr is bounded by inequality") {
+ // Left side: a.k >= 5
+ // Join condition: b.v >= a.k
+ // Transitive inference should deduce b.v >= 5 and push it down to the
right side of
+ // LeftOuter join.
+ val left = LocalRelation($"k".int, $"key".int).subquery("a")
+ val right = LocalRelation($"v".int, $"key".int).subquery("b")
+
+ val joinCond = ("a.key".attr === "b.key".attr) && ("b.v".attr >=
"a.k".attr)
+ val originalQuery = left
+ .where("a.k".attr >= 5)
+ .join(right, LeftOuter, Some(joinCond))
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+ var pushedToRight = false
+ optimized.foreach {
+ case Join(_, Filter(cond, _), LeftOuter, _, _) =>
+ pushedToRight = splitConjunctivePredicates(cond).exists {
+ case GreaterThanOrEqual(attr: Attribute, Literal(v: Int,
IntegerType)) =>
+ attr.name == "v" && v == 5
+ case _ => false
+ }
+ case _ =>
+ }
+ assert(pushedToRight, s"Expected b.v >= 5 pushed to right side; plan
was:\n$optimized")
+ }
+
+ test("SPARK-30768: do not infer filters from throwable expressions") {
Review Comment:
**Nit (P3):** Since `isSafeToRelocate` was added, this test no longer
exercises `!b.throwable`. Without the guard, `caseExpr < x.b` becomes a
candidate edge, but `caseExpr` fails `isSafeToRelocate`, so no upper edge on
`x.b` is recorded. There is also no bound on `caseExpr` to propagate, so
nothing is inferred and the assertion passes either way.
The guard still matters within one relation. In `Filter(x < caseExpr AND
caseExpr < 5)`, `caseExpr` is the bounded site and `x` is relocatable, so
without the guard `x < 5` is inferred and prepended. That short-circuits the
`raise_error` the original predicate evaluates for rows with `x >= 5`. Using
that shape and asserting that no `x < 5` is inferred would give the guard a
real regression signal.
**Verification:**
- **Regression:** A single-relation filter whose throwable expression is
bounded by a literal and chained to a relocatable attribute infers no new bound
on that attribute.
--
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]
