From ac4006aed0fde4060461f42888adcfbe03eb2321 Mon Sep 17 00:00:00 2001 From: xiaoh1024 Date: Mon, 21 Sep 2026 17:51:08 +0800 Subject: [PATCH 1/2] fix: simplify contained integral range predicates (SPARK-31760) --- .../sql/catalyst/optimizer/expressions.scala | 70 +++++++ .../IntegralRangeSimplificationSuite.scala | 175 ++++++++++++++++++ .../spark/sql/IntegralRangeQuerySuite.scala | 56 ++++++ 3 files changed, 301 insertions(+) create mode 100644 sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/IntegralRangeSimplificationSuite.scala create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/IntegralRangeQuerySuite.scala 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 cdc0444d74cc1..746f1fd0e2e01 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,70 @@ 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 + } + } + + 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 0000000000000..1cc2e4185018c --- /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 0000000000000..b3dad20a72570 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/IntegralRangeQuerySuite.scala @@ -0,0 +1,56 @@ +/* + * 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") + 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) + } + } + } + } +} From bec3163111d3df314a02c7c53045de83e00e5102 Mon Sep 17 00:00:00 2001 From: xiaoh1024 Date: Tue, 22 Sep 2026 20:47:50 +0800 Subject: [PATCH 2/2] test: cover folded integral range bounds --- .../org/apache/spark/sql/catalyst/optimizer/expressions.scala | 2 ++ .../scala/org/apache/spark/sql/IntegralRangeQuerySuite.scala | 3 ++- 2 files changed, 4 insertions(+), 1 deletion(-) 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 746f1fd0e2e01..23eb8009e2475 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 @@ -593,6 +593,8 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper { } } + // 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) 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 index b3dad20a72570..d7f624125f325 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/IntegralRangeQuerySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/IntegralRangeQuerySuite.scala @@ -34,7 +34,8 @@ class IntegralRangeQuerySuite extends QueryTest with SharedSparkSession { 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 > 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") } {