This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5f39cc278f34 [SPARK-57437][SQL] Infer additional constraints by
substituting attribute-to-literal bindings
5f39cc278f34 is described below
commit 5f39cc278f34e84ccadfeb8b069130dfe261753f
Author: James Xu <[email protected]>
AuthorDate: Thu Jul 2 14:28:41 2026 +0800
[SPARK-57437][SQL] Infer additional constraints by substituting
attribute-to-literal bindings
### What changes were proposed in this pull request?
When a predicate binds an attribute to a literal (e.g. a.pt = '20260610')
and another predicate references that attribute (e.g. b.pt >= f(a.pt)),
Catalyst previously did not exploit the literal binding to derive a simpler,
pushable predicate.
```sql
SELECT *
FROM a
LEFT JOIN b
ON a.key = b.key
AND b.pt >= f(a.pt)
WHERE a.pt = '20260610';
```
Here table b is full scaned, thus very bad performance.
This change adds a new inferConstraintsFromLiteralBindings that:
1. Collects Attribute = Literal bindings from the constraint set.
2. Substitutes the literal into predicates that reference those
attributes.
3. Adds the resulting deterministic expressions as new inferred
constraints.
After constant folding, the inferred predicates can be pushed into scans as
partition filters, avoiding full-table scans in cases where only a small subset
of partitions can match.
### Why are the changes needed?
Currently query of the following pattern causes full table scan for table b:
```sql
SELECT *
FROM a
LEFT JOIN b
ON a.key = b.key
AND b.pt >= f(a.pt)
WHERE a.pt = '20260610';
```
With this optimization table b can get very good partition pruning.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added unit tests.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #56499 from
xumingming/catalyst-infer-constraints-from-literal-bindings.
Authored-by: James Xu <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 24550454644568cb31bfab28768debcd259da1aa)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/catalyst/optimizer/Optimizer.scala | 4 +-
.../plans/logical/QueryPlanConstraints.scala | 46 +++++++
.../InferFiltersFromConstraintsSuite.scala | 52 +++++++
.../plans/ConstraintPropagationSuite.scala | 149 +++++++++++++++++++++
4 files changed, 250 insertions(+), 1 deletion(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 6753dd9a723c..87bf4c7e1111 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -1834,7 +1834,9 @@ object InferFiltersFromConstraints extends
Rule[LogicalPlan]
conditionOpt: Option[Expression]): ExpressionSet = {
val baseConstraints = left.constraints.union(right.constraints)
.union(ExpressionSet(conditionOpt.map(splitConjunctivePredicates).getOrElse(Nil)))
- baseConstraints.union(inferAdditionalConstraints(baseConstraints))
+ baseConstraints
+ .union(inferAdditionalConstraints(baseConstraints))
+ .union(inferConstraintsFromLiteralBindings(baseConstraints))
}
private def inferNewFilter(plan: LogicalPlan, constraints: ExpressionSet):
LogicalPlan = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
index ef035eba5922..6a1aa54a54d0 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.plans.logical
import scala.annotation.tailrec
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.UnsafeRowUtils.isBinaryStable
trait QueryPlanConstraints extends ConstraintHelper { self: LogicalPlan =>
@@ -33,6 +34,7 @@ trait QueryPlanConstraints extends ConstraintHelper { self:
LogicalPlan =>
if (conf.constraintPropagationEnabled) {
validConstraints
.union(inferAdditionalConstraints(validConstraints))
+ .union(inferConstraintsFromLiteralBindings(validConstraints))
.union(constructIsNotNullConstraints(validConstraints, output))
.filter { c =>
c.references.nonEmpty && c.references.subsetOf(outputSet) &&
c.deterministic
@@ -77,9 +79,53 @@ trait ConstraintHelper {
inferredConstraints ++= replaceConstraints(predicates - eq -
EqualNullSafe(l, r), l, r)
case _ => // No inference
}
+
inferredConstraints -- constraints
}
+ /**
+ * Infers additional constraints by substituting known attribute-to-literal
bindings into
+ * non-equality predicates. For example, given `a = 5` and `b >= a`, infers
`b >= 5`.
+ *
+ * Attribute-to-attribute (and cast-form) [[EqualTo]] and all
[[EqualNullSafe]] predicates are
+ * excluded from substitution targets.
+ * Substituting into attribute-attribute [[EqualTo]] would duplicate work
already done by the
+ * transitivity case in [[inferAdditionalConstraints]] (e.g. `a = 5` and `b
= a` imply `b = 5`,
+ * already derived there). Substituting into [[EqualNullSafe]] would produce
a structurally
+ * distinct form (e.g. `b <=> 5`) that downstream rules (such as
subquery-reuse matching)
+ * treat differently from the [[EqualTo]] form, leading to duplicate
subqueries being generated.
+ */
+ def inferConstraintsFromLiteralBindings(constraints: ExpressionSet):
ExpressionSet = {
+ // Collect attr -> literal bindings, guarded by binary-stable collation so
that
+ // the substitution is semantically safe (non-binary-stable collations may
equate
+ // strings that are binary-distinct, so substituting the literal would
change results).
+ val bindings: Map[Attribute, Literal] = constraints.collect {
+ case EqualTo(a: Attribute, l: Literal) if isBinaryStable(a.dataType) =>
a -> l
+ case EqualTo(l: Literal, a: Attribute) if isBinaryStable(a.dataType) =>
a -> l
+ }.toMap
+
+ if (bindings.isEmpty) return ExpressionSet()
+
+ val targets = constraints.filterNot {
+ // attr=attr EqualTo: already covered by inferAdditionalConstraints
(transitivity).
+ // cast-equality EqualTo: would produce redundant cast literals
derivable from that path.
+ // EqualNullSafe: substituting a literal produces a structurally
distinct b <=> lit form
+ // that subquery-reuse matching treats differently from b = lit, causing
duplicate subqueries.
+ // IsNotNull: handled separately by constructIsNotNullConstraints.
+ case EqualTo(_: Attribute, _: Attribute) => true
+ case EqualTo(Cast(_: Attribute, _, _, _), _: Attribute) => true
+ case EqualTo(_: Attribute, Cast(_: Attribute, _, _, _)) => true
+ case _: EqualNullSafe | _: IsNotNull => true
+ case _ => false
+ }
+
+ var inferred = ExpressionSet()
+ bindings.foreach { case (attr, lit) =>
+ inferred ++= replaceConstraints(targets, attr, lit)
+ }
+ inferred -- constraints
+ }
+
private def replaceConstraints(
constraints: ExpressionSet,
source: Expression,
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala
index d8d8a2b333bc..bfed9427c863 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala
@@ -399,4 +399,56 @@ class InferFiltersFromConstraintsSuite extends PlanTest {
comparePlans(optimizedQuery, correctAnswer)
comparePlans(InferFiltersFromConstraints(optimizedQuery), correctAnswer)
}
+
+ test("correlated range predicate in left-outer join ON clause is pushed to
right side " +
+ "when left attr is bound to a literal") {
+ testRangePredicatePushDown(LeftOuter)
+ }
+
+ test("correlated range predicate in inner join ON clause is pushed to right
side " +
+ "when left attr is bound to a literal") {
+ testRangePredicatePushDown(Inner)
+ }
+
+ private def testRangePredicatePushDown(joinType: JoinType): Unit = {
+ // Simulates: SELECT * FROM a JOIN b ON a.key = b.key AND b.v >= a.k AND
b.v <= a.k + 10
+ // WHERE a.k = 5
+ // The constant binding a.k = 5 should be substituted into the range
predicates, producing
+ // b.v >= 5 and b.v <= (5 + 10) that get pushed into the right scan.
+ 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) &&
+ ("b.v".attr <= ("a.k".attr + 10))
+
+ val originalQuery = left
+ .where("a.k".attr === 5)
+ .join(right, joinType, Some(joinCond))
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // The optimized right side must carry the inferred range filters.
+ var joinFound = false
+ optimized.foreach {
+ case Join(_, rightChild, `joinType`, _, _) =>
+ joinFound = true
+ val rightFilters = rightChild.collect { case Filter(cond, _) => cond }
+ val allConds = rightFilters.flatMap(splitConjunctivePredicates)
+ // b.v >= 5 and b.v <= (5 + 10) must appear (arithmetic not folded
without ConstantFolding)
+ assert(allConds.exists {
+ case GreaterThanOrEqual(_: Attribute, Literal(v, IntegerType)) => v
== 5
+ case _ => false
+ }, s"Expected b.v >= 5 in right filter; found: $allConds")
+ assert(allConds.exists {
+ case LessThanOrEqual(_: Attribute, add: Add)
+ if add.left == Literal(5, IntegerType) && add.right == Literal(10,
IntegerType) => true
+ case _ => false
+ }, s"Expected b.v <= (5 + 10) in right filter; found: $allConds")
+ case _ =>
+ }
+ assert(joinFound, "Expected a Join node in the optimized plan")
+ }
+
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
index 5f6084287815..61ee06d4aaf1 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
@@ -24,6 +24,7 @@ import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.util.CollationFactory
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataType, DoubleType, IntegerType,
LongType, StringType}
@@ -427,4 +428,152 @@ class ConstraintPropagationSuite extends PlanTest {
assert(aliasedRelation.analyze.constraints.isEmpty)
}
}
+
+ test("infer range constraints by substituting attr=literal bindings") {
+ // When a.pt = '20260610' (modeled as a.k = 5) and b.v >= a.k are both
present,
+ // inferConstraintsFromLiteralBindings should emit b.v >= 5.
+ val tr = LocalRelation($"k".int, $"v".int)
+ val a = resolveColumn(tr, "k")
+ val b = resolveColumn(tr, "v")
+
+ val constraints = ExpressionSet(Seq(
+ EqualTo(a, Literal(5)),
+ GreaterThanOrEqual(b, a),
+ LessThanOrEqual(b, Add(a, Literal(10)))))
+
+ val helper = new ConstraintHelper {}
+ val inferred = helper.inferConstraintsFromLiteralBindings(constraints)
+
+ // b >= 5 must be inferred
+ assert(inferred.exists {
+ case GreaterThanOrEqual(attr: Attribute, Literal(v, IntegerType)) =>
+ attr.semanticEquals(b) && v == 5
+ case _ => false
+ }, s"Expected b >= 5 in inferred constraints; got: $inferred")
+
+ // b <= 15 must be inferred (a + 10 with a=5, i.e. Add(Literal(5),
Literal(10)))
+ assert(inferred.exists {
+ case LessThanOrEqual(attr: Attribute, rhs) if attr.semanticEquals(b) =>
+ // rhs is either Literal(15) or Add(Literal(5), Literal(10)) - both
are foldable
+ rhs.foldable && rhs.eval(null) == 15
+ case _ => false
+ }, s"Expected b <= 15 in inferred constraints; got: $inferred")
+ }
+
+ test("infer range constraints: literal on left side of EqualTo") {
+ val tr = LocalRelation($"k".int, $"v".int)
+ val a = resolveColumn(tr, "k")
+ val b = resolveColumn(tr, "v")
+
+ // Literal appears on the left: 5 = a.k
+ val constraints = ExpressionSet(Seq(
+ EqualTo(Literal(5), a),
+ GreaterThanOrEqual(b, a)))
+
+ val helper = new ConstraintHelper {}
+ val inferred = helper.inferConstraintsFromLiteralBindings(constraints)
+
+ assert(inferred.exists {
+ case GreaterThanOrEqual(attr: Attribute, Literal(v, IntegerType)) =>
+ attr.semanticEquals(b) && v == 5
+ case _ => false
+ }, s"Expected b >= 5 in inferred constraints; got: $inferred")
+ }
+
+ test("infer constraints by substituting attr=literal bindings into EqualTo
expressions") {
+ val tr = LocalRelation($"k".int, $"v".int)
+ val a = resolveColumn(tr, "k")
+ val b = resolveColumn(tr, "v")
+
+ val constraints = ExpressionSet(Seq(
+ EqualTo(a, Literal(5)),
+ EqualTo(b, Add(a, Literal(1)))))
+
+ val helper = new ConstraintHelper {}
+ val inferred = helper.inferConstraintsFromLiteralBindings(constraints)
+
+ // b = a + 1 with a = 5 should infer b = 5 + 1 (foldable to 6)
+ assert(inferred.exists {
+ case EqualTo(attr: Attribute, rhs) if attr.semanticEquals(b) =>
+ rhs.foldable && rhs.eval(null) == 6
+ case _ => false
+ }, s"Expected b = 6 in inferred constraints; got: $inferred")
+ }
+
+ test("non-deterministic constraints inferred from literal substitution are
filtered out") {
+ val tr = LocalRelation($"k".int, $"v".int)
+ val a = resolveColumn(tr, "k")
+ val b = resolveColumn(tr, "v")
+
+ // b >= a.k + Rand(); after substituting a.k=5 the result is
non-deterministic.
+ // inferConstraintsFromLiteralBindings may emit it, but the plan-level
constraints
+ // filter must drop it.
+ val randExpr = Add(a, Cast(Rand(0L), IntegerType))
+ val condition = EqualTo(a, Literal(5)) && GreaterThanOrEqual(b, randExpr)
+ val plan = Filter(condition, tr.analyze)
+
+ assert(!plan.analyze.constraints.exists {
+ case GreaterThanOrEqual(attr: Attribute, _) => attr.semanticEquals(b)
+ case _ => false
+ }, s"Should not keep non-deterministic constraint in plan constraints; " +
+ s"got: ${plan.analyze.constraints}")
+ }
+
+ test("do not infer constraints by substituting non-binary-stable collated
attributes") {
+ // a is UTF8_LCASE (case-insensitive). a = 'hello' only tells us a is
case-insensitively
+ // equal to 'hello'; it could still be 'HELLO'. Substituting a with
'hello' into a
+ // UTF8_BINARY comparison like b >= a would infer b >= 'hello', which is
wrong.
+ val a = AttributeReference("a",
StringType(CollationFactory.UTF8_LCASE_COLLATION_ID))()
+ val b = AttributeReference("b", StringType)()
+ val hello = Literal("hello")
+
+ val constraints = ExpressionSet(Seq(
+ EqualTo(a, hello),
+ GreaterThanOrEqual(b, a)))
+
+ val helper = new ConstraintHelper {}
+ val inferred = helper.inferConstraintsFromLiteralBindings(constraints)
+
+ assert(!inferred.exists {
+ case GreaterThanOrEqual(attr: Attribute, lit: Literal) =>
+ attr.semanticEquals(b) && lit.semanticEquals(hello)
+ case _ => false
+ }, s"Should not infer b >= 'hello' for non-binary-stable collation; got:
$inferred")
+ }
+
+ test("do not substitute attr=literal bindings into EqualNullSafe or
attr-attr EqualTo targets") {
+ // EqualNullSafe and attribute-attribute EqualTo are excluded from
substitution targets:
+ // - Substituting into EqualNullSafe(b, a) would yield b <=> 5, a
structurally distinct form
+ // that subquery-reuse matching treats differently from b = 5, causing
duplicate subqueries
+ // (the q78 regression root).
+ // - Substituting into EqualTo(c, a) would duplicate the transitivity
already handled by
+ // inferAdditionalConstraints, which derives c = 5 from c = a and a = 5.
+ val tr = LocalRelation($"k".int, $"v".int, $"w".int)
+ val a = resolveColumn(tr, "k")
+ val b = resolveColumn(tr, "v")
+ val c = resolveColumn(tr, "w")
+
+ val constraints = ExpressionSet(Seq(
+ EqualTo(a, Literal(5)),
+ EqualNullSafe(b, a),
+ EqualTo(c, a)))
+
+ val helper = new ConstraintHelper {}
+ val inferred = helper.inferConstraintsFromLiteralBindings(constraints)
+
+ // No b <=> 5 should be synthesized from the EqualNullSafe target.
+ assert(!inferred.exists {
+ case EqualNullSafe(attr: Attribute, lit: Literal) =>
+ attr.semanticEquals(b) && lit.semanticEquals(Literal(5))
+ case _ => false
+ }, s"Should not infer b <=> 5 from EqualNullSafe target; got: $inferred")
+
+ // No c = 5 should be synthesized from the attr-attr EqualTo target; that
derivation
+ // belongs to inferAdditionalConstraints, not literal substitution.
+ assert(!inferred.exists {
+ case EqualTo(attr: Attribute, lit: Literal) =>
+ attr.semanticEquals(c) && lit.semanticEquals(Literal(5))
+ case _ => false
+ }, s"Should not infer c = 5 from attr-attr EqualTo target; got: $inferred")
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
