You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/09/22 19:55:10 UTC
(spark) branch master updated: [SPARK-31760][SQL] Simplify contained integral range predicates
This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 9d84a6c1185d [SPARK-31760][SQL] Simplify contained integral range predicates
9d84a6c1185d is described below
commit 9d84a6c1185d6957d9e7ccb99ce25a92862cd436
Author: 小H <xi...@qq.com>
AuthorDate: Tue Sep 22 19:54:57 2026 +0000
[SPARK-31760][SQL] Simplify contained integral range predicates
### What changes were proposed in this pull request?
Extend `BooleanSimplification` to remove redundant integral range comparisons:
```sql
a > 5 AND a > 0 --> a > 5
a > 1 OR a > 2 --> a > 1
a > 1 OR (a > 2 AND a < 4) --> a > 1
```
Support `<`, `<=`, `>`, and `>=` between direct Byte/Short/Int/Long attributes and non-null literals of the same type, including reversed operands and one level of nested absorption. Bounds are compared without arithmetic to avoid overflow. Casts and arbitrary expressions are excluded.
### Why are the changes needed?
The existing rule retains these redundant comparisons. Removing them reduces per-row evaluation. This implements the integral range subset of [SPARK-31760](https://issues.apache.org/jira/browse/SPARK-31760).
### Does this PR introduce _any_ user-facing change?
Yes: eligible queries have simpler optimized plans and may execute faster. Query results, including SQL NULL semantics, are preserved. No public API or configuration changes.
### How was this patch tested?
Added Catalyst and SQL tests covering operand order, inclusive/exclusive bounds, nested absorption, NULLs, integer extremes, unsupported expressions, and generated evaluation. The new contained-range regression test fails with the original rule.
On Spark 5.0.0-SNAPSHOT with Scala 2.13.18 and JDK 17.0.18:
- 37 Catalyst tests passed across the new suite and the existing BooleanSimplification and BinaryComparisonSimplification suites.
- The SQL suite passed all 16 filter/projection and AQE on/off combinations.
- Catalyst source/test and SQL test scalastyle checks passed.
Local synthetic-data benchmarks used 30 million cached rows and `local[2]`, covering filter/projection queries with 0% and 10% nulls. AND placed the wider condition first, OR placed the narrower condition first, and nested absorption placed the complex branch first. Four JVMs ran in baseline/candidate/candidate/baseline order. Median query time decreased by 3–5% for AND, 30–34% for OR, and 35–43% for nested absorption. A separate rule microbenchmark measured up to 3.2 microseconds add [...]
### Was this patch authored or co-authored using generative AI tooling?
Yes.
Closes #58940 from xiaoh1024/fix/spark-31760-integral-ranges.
Authored-by: 小H <xi...@qq.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../spark/sql/catalyst/optimizer/expressions.scala | 72 +++++++++
.../IntegralRangeSimplificationSuite.scala | 175 +++++++++++++++++++++
.../apache/spark/sql/IntegralRangeQuerySuite.scala | 57 +++++++
3 files changed, 304 insertions(+)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala
index cdc0444d74cc..23eb8009e247 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala
@@ -391,6 +391,7 @@ object OptimizeIn extends Rule[LogicalPlan] {
* 2. Eliminates / extracts common factors.
* 3. Merge same expressions
* 4. Removes `Not` operator.
+ * 5. Simplifies contained integral ranges.
*/
object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
@@ -425,6 +426,11 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
case a And b if a.semanticEquals(b) => a
case a Or b if a.semanticEquals(b) => a
+ case a And b if integralRangeImplies(a, b) => a
+ case a And b if integralRangeImplies(b, a) => b
+ case a Or b if integralRangeImplies(a, b) => b
+ case a Or b if integralRangeImplies(b, a) => a
+
// The following optimizations are applicable only when the operands are not nullable,
// since the three-value logic of AND and OR are different in NULL handling.
// See the chart:
@@ -564,6 +570,72 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
case _ => not
}
+
+ private case class IntegralRange(
+ attribute: AttributeReference,
+ bound: Long,
+ isLowerBound: Boolean,
+ inclusive: Boolean)
+
+ private object IntegralComparison {
+ def unapply(expression: Expression): Option[IntegralRange] = {
+ def range(
+ attribute: AttributeReference,
+ literal: Literal,
+ isLowerBound: Boolean,
+ inclusive: Boolean): Option[IntegralRange] = {
+ if (attribute.dataType.isInstanceOf[IntegralType] &&
+ literal.dataType == attribute.dataType && literal.value != null) {
+ Some(IntegralRange(attribute, literal.value.asInstanceOf[Number].longValue(),
+ isLowerBound, inclusive))
+ } else {
+ None
+ }
+ }
+
+ // ConstantFolding precedes BooleanSimplification in the operator optimization batches,
+ // so foldable bounds such as 2 + 2 can be matched as literals.
+ expression match {
+ case GreaterThan(a: AttributeReference, l: Literal) => range(a, l, true, false)
+ case GreaterThanOrEqual(a: AttributeReference, l: Literal) => range(a, l, true, true)
+ case LessThan(a: AttributeReference, l: Literal) => range(a, l, false, false)
+ case LessThanOrEqual(a: AttributeReference, l: Literal) => range(a, l, false, true)
+ case GreaterThan(l: Literal, a: AttributeReference) => range(a, l, false, false)
+ case GreaterThanOrEqual(l: Literal, a: AttributeReference) => range(a, l, false, true)
+ case LessThan(l: Literal, a: AttributeReference) => range(a, l, true, false)
+ case LessThanOrEqual(l: Literal, a: AttributeReference) => range(a, l, true, true)
+ case _ => None
+ }
+ }
+ }
+
+ // For comparisons on the same attribute, containment also preserves UNKNOWN: both
+ // comparisons return NULL together. Keep arbitrary expressions out of these rules so
+ // removing a comparison cannot suppress an exception or a nondeterministic evaluation.
+ private def integralRangeImplies(left: Expression, right: Expression): Boolean = {
+ (left, right) match {
+ case (IntegralComparison(l), IntegralComparison(r))
+ if l.attribute.semanticEquals(r.attribute) && l.isLowerBound == r.isLowerBound =>
+ val order = java.lang.Long.compare(l.bound, r.bound)
+ if (order == 0) {
+ !l.inclusive || r.inclusive
+ } else if (l.isLowerBound) {
+ order > 0
+ } else {
+ order < 0
+ }
+
+ case (And(l @ IntegralComparison(_), r @ IntegralComparison(_)),
+ target @ IntegralComparison(_)) =>
+ integralRangeImplies(l, target) || integralRangeImplies(r, target)
+
+ case (source @ IntegralComparison(_),
+ Or(l @ IntegralComparison(_), r @ IntegralComparison(_))) =>
+ integralRangeImplies(source, l) || integralRangeImplies(source, r)
+
+ case _ => false
+ }
+ }
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/IntegralRangeSimplificationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/IntegralRangeSimplificationSuite.scala
new file mode 100644
index 000000000000..1cc2e4185018
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/IntegralRangeSimplificationSuite.scala
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.types._
+
+class IntegralRangeSimplificationSuite extends PlanTest with ExpressionEvalHelper {
+ private val a = AttributeReference("a", IntegerType)()
+
+ private def optimize(expression: Expression): Expression = {
+ val relation = LocalRelation(expression.references.toSeq)
+ BooleanSimplification(Project(Seq(Alias(expression, "result")()), relation))
+ .asInstanceOf[Project].projectList.head.asInstanceOf[Alias].child
+ }
+
+ private def checkSimplification(input: Expression, expected: Expression): Unit = {
+ val relation = LocalRelation(input.references.toSeq)
+ comparePlans(
+ BooleanSimplification(Project(Seq(Alias(input, "result")()), relation)),
+ Project(Seq(Alias(expected, "result")()), relation))
+ comparePlans(BooleanSimplification(Filter(input, relation)), Filter(expected, relation))
+ }
+
+ test("SPARK-31760: simplify contained integral ranges in either operand order") {
+ val cases = Seq(
+ (GreaterThan(a, Literal(5)), GreaterThan(a, Literal(0))),
+ (GreaterThanOrEqual(a, Literal(5)), GreaterThan(a, Literal(0))),
+ (GreaterThan(a, Literal(5)), GreaterThanOrEqual(a, Literal(5))),
+ (LessThan(a, Literal(0)), LessThan(a, Literal(5))),
+ (LessThanOrEqual(a, Literal(0)), LessThan(a, Literal(5))),
+ (LessThan(a, Literal(5)), LessThanOrEqual(a, Literal(5))),
+ (LessThan(Literal(5), a), GreaterThan(a, Literal(0))),
+ (GreaterThan(Literal(0), a), LessThan(a, Literal(5))),
+ (LessThanOrEqual(Literal(5), a), GreaterThan(a, Literal(0))),
+ (GreaterThanOrEqual(Literal(0), a), LessThan(a, Literal(5))))
+
+ for ((narrow, wide) <- cases) {
+ checkSimplification(And(narrow, wide), narrow)
+ checkSimplification(And(wide, narrow), narrow)
+ checkSimplification(Or(narrow, wide), wide)
+ checkSimplification(Or(wide, narrow), wide)
+ }
+ }
+
+ test("SPARK-31760: absorb a contained range in a conjunction or disjunction") {
+ val wide = GreaterThan(a, Literal(1))
+ val narrow = GreaterThan(a, Literal(2))
+ val other = LessThan(a, Literal(4))
+ for (conjunction <- Seq(And(narrow, other), And(other, narrow))) {
+ checkSimplification(Or(wide, conjunction), wide)
+ checkSimplification(Or(conjunction, wide), wide)
+ }
+ for (disjunction <- Seq(Or(wide, other), Or(other, wide))) {
+ checkSimplification(And(narrow, disjunction), narrow)
+ checkSimplification(And(disjunction, narrow), narrow)
+ }
+ }
+
+ test("SPARK-31760: leave unrelated or unsupported comparisons unchanged") {
+ val b = AttributeReference("b", IntegerType)()
+ val x = AttributeReference("x", DoubleType)()
+ val d = AttributeReference("d", DecimalType(10, 2))()
+ val expressions = Seq(
+ And(GreaterThan(a, Literal(5)), GreaterThan(b, Literal(0))),
+ And(GreaterThan(a, Literal(5)), LessThan(a, Literal(0))),
+ And(GreaterThan(a, Literal(5)), EqualTo(a, Literal(0))),
+ And(GreaterThan(a, Literal(5)), GreaterThan(a, Literal(null, IntegerType))),
+ And(GreaterThan(Add(a, Literal(1)), Literal(5)),
+ GreaterThan(Add(a, Literal(1)), Literal(0))),
+ And(GreaterThan(x, Literal(5.0)), GreaterThan(x, Literal(0.0))),
+ And(GreaterThan(x, Literal(Double.NaN)), GreaterThan(x, Literal(0.0))),
+ And(GreaterThan(d, Literal.create(Decimal(5), d.dataType)),
+ GreaterThan(d, Literal.create(Decimal(0), d.dataType))),
+ And(GreaterThan(Cast(x, IntegerType), Literal(5)),
+ GreaterThan(Cast(x, IntegerType), Literal(0))),
+ And(GreaterThan(Cast(Rand(0), IntegerType), Literal(5)),
+ GreaterThan(Cast(Rand(0), IntegerType), Literal(0))))
+ expressions.foreach(expression => checkSimplification(expression, expression))
+
+ val wide = GreaterThan(a, Literal(1))
+ val narrow = GreaterThan(a, Literal(2))
+ val other = GreaterThan(Cast(x, IntegerType), Literal(0))
+ checkSimplification(Or(wide, And(narrow, other)), Or(wide, And(narrow, other)))
+ checkSimplification(And(narrow, Or(wide, other)), And(narrow, Or(wide, other)))
+ }
+
+ test("SPARK-31760: absorption preserves NULLs in an independent comparison") {
+ val b = AttributeReference("b", IntegerType)()
+ val wide = GreaterThan(a, Literal(1))
+ val narrow = GreaterThan(a, Literal(2))
+ val other = LessThan(b, Literal(4))
+ val cases = Seq(
+ Or(wide, And(narrow, other)) -> wide,
+ Or(And(other, narrow), wide) -> wide,
+ And(narrow, Or(wide, other)) -> narrow,
+ And(Or(other, wide), narrow) -> narrow)
+ for ((input, expected) <- cases) {
+ checkSimplification(input, expected)
+ val original = BindReferences.bindReference(input, Seq(a, b))
+ val simplified = BindReferences.bindReference(optimize(input), Seq(a, b))
+ for (av <- Seq(null, 0, 1, 2, 3); bv <- Seq(null, 3, 4, 5)) {
+ val row = InternalRow(av, bv)
+ assert(original.eval(row) == simplified.eval(row))
+ }
+ }
+ }
+
+ test("SPARK-31760: preserve three-valued logic for all integral types and extreme bounds") {
+ val typesAndBounds = Seq(
+ (ByteType, Seq(Byte.MinValue, -1, 0, 1, Byte.MaxValue).map(_.toByte)),
+ (ShortType, Seq(Short.MinValue, -1, 0, 1, Short.MaxValue).map(_.toShort)),
+ (IntegerType, Seq(Int.MinValue, -1, 0, 1, Int.MaxValue)),
+ (LongType, Seq(Long.MinValue, -1L, 0L, 1L, Long.MaxValue)))
+
+ for ((dataType, bounds) <- typesAndBounds; nullable <- Seq(false, true)) {
+ val column = AttributeReference("a", dataType, nullable)()
+ val comparisons = bounds.flatMap { value =>
+ val literal = Literal.create(value, dataType)
+ Seq(GreaterThan(column, literal), GreaterThanOrEqual(column, literal),
+ LessThan(column, literal), LessThanOrEqual(column, literal))
+ }
+ val values = if (nullable) bounds :+ null else bounds
+ var changed = 0
+ for {
+ left <- comparisons
+ right <- comparisons
+ input <- Seq(And(left, right), Or(left, right))
+ } {
+ val simplified = optimize(input)
+ if (!input.semanticEquals(simplified)) changed += 1
+ val original = BindReferences.bindReference(input, Seq(column))
+ val result = BindReferences.bindReference(simplified, Seq(column))
+ for (value <- values) {
+ val row = InternalRow(value)
+ assert(original.eval(row) == result.eval(row),
+ s"$dataType, nullable=$nullable, value=$value: $input -> $simplified")
+ }
+ }
+ assert(changed > 0)
+ }
+ }
+
+ test("SPARK-31760: generated evaluation preserves NULL and range boundaries") {
+ val inputs = Seq(
+ And(GreaterThan(a, Literal(5)), GreaterThan(a, Literal(0))),
+ Or(GreaterThan(a, Literal(1)), GreaterThan(a, Literal(2))),
+ Or(GreaterThan(a, Literal(1)),
+ And(GreaterThan(a, Literal(2)), LessThan(a, Literal(4)))))
+ for (input <- inputs; value <- Seq(null, Int.MinValue, 0, 1, 2, 5, 6, Int.MaxValue)) {
+ val original = BindReferences.bindReference(input, Seq(a))
+ val simplified = BindReferences.bindReference(optimize(input), Seq(a))
+ val row = InternalRow(value)
+ checkEvaluation(simplified, original.eval(row), row)
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/IntegralRangeQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/IntegralRangeQuerySuite.scala
new file mode 100644
index 000000000000..d7f624125f32
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/IntegralRangeQuerySuite.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.expressions.BinaryComparison
+import org.apache.spark.sql.catalyst.optimizer.BooleanSimplification
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+class IntegralRangeQuerySuite extends QueryTest with SharedSparkSession {
+ test("SPARK-31760: range simplification preserves filter and projection results") {
+ withTempView("ranges") {
+ val rows = Seq(null, Int.MinValue, -1, 0, 1, 2, 3, 4, 5, 6, 6, Int.MaxValue).map(Row(_))
+ spark.createDataFrame(spark.sparkContext.parallelize(rows, 2),
+ StructType(Seq(StructField("a", IntegerType, nullable = true))))
+ .createOrReplaceTempView("ranges")
+
+ for {
+ aqe <- Seq(false, true)
+ condition <- Seq("a > 5 AND a > 0", "a > 1 OR a > 2",
+ "a > 1 OR (a > 2 AND a < 4)", "5 >= a AND a < 5",
+ "a > (2 + 2) AND a > (1 - 1)", "a > (1 + 1) OR a > (2 * 2)")
+ sqlText <- Seq(s"SELECT a FROM ranges WHERE $condition",
+ s"SELECT $condition AS result FROM ranges")
+ } {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) {
+ val expected = withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+ BooleanSimplification.ruleName) {
+ spark.sql(sqlText).collect().toSeq
+ }
+ val query = spark.sql(sqlText)
+ val comparisons = query.queryExecution.optimizedPlan.flatMap { plan =>
+ plan.expressions.flatMap(_.collect { case comparison: BinaryComparison => comparison })
+ }
+ assert(comparisons.size == 1, query.queryExecution.optimizedPlan.toString)
+ checkAnswer(query, expected)
+ }
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org