This is an automated email from the ASF dual-hosted git repository.
rui-mo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 1e896874b0 [GLUTEN-13179][VL] Combine pushed-down filters into a
balanced And tree and re-enable GlutenWholeStageCodegenSuite for Spark 4.2
(#13435)
1e896874b0 is described below
commit 1e896874b0ef4ddf2f8ceecd3c771d0970209836
Author: Akshay Tayal <[email protected]>
AuthorDate: Mon Oct 26 15:38:47 2026 -0700
[GLUTEN-13179][VL] Combine pushed-down filters into a balanced And tree and
re-enable GlutenWholeStageCodegenSuite for Spark 4.2 (#13435)
---
.../BasicPhysicalOperatorTransformer.scala | 10 ++-
.../execution/BasicScanExecTransformer.scala | 5 +-
.../gluten/utils/velox/VeloxTestSettings.scala | 5 ++
.../execution/GlutenWholeStageCodegenSuite.scala | 82 ++++++++++++++++++++--
4 files changed, 93 insertions(+), 9 deletions(-)
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
index 19864d2960..9cde1e7939 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
@@ -118,7 +118,7 @@ abstract class FilterExecTransformerBase(val cond:
Expression, val input: SparkP
} else {
val remainingFilters =
FilterHandler.subtractFilters(splitConjunctivePredicates(cond),
scanFilters)
- remainingFilters.reduceLeftOption(And).orNull
+ FilterHandler.combineConjuncts(remainingFilters).orNull
}
}
@@ -340,4 +340,12 @@ object FilterHandler extends PredicateHelper {
def combineFilters(left: Seq[Expression], right: Seq[Expression]):
Seq[Expression] = {
(left.toSet ++ right.toSet).toSeq
}
+
+ /**
+ * Combines the conjuncts into a balanced And tree. A left-deep tree grows
linearly in depth with
+ * the number of conjuncts, which can overflow the stack in the recursive
expression conversion.
+ */
+ def combineConjuncts(conjuncts: Seq[Expression]): Option[Expression] = {
+ if (conjuncts.isEmpty) None else Some(buildBalancedPredicate(conjuncts,
And))
+ }
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
index faa19881e3..b4a7fc0159 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
@@ -194,9 +194,8 @@ trait BasicScanExecTransformer extends LeafTransformSupport
with BaseDataSource
val nameList = ConverterUtils.collectAttributeNamesWithoutExprId(output)
val columnTypeNodes = output.map(makeColumnTypeNode).asJava
// Will put all filter expressions into an AND expression
- val exprNode = filterExprs()
- .map(ExpressionConverter.replaceAttributeReference)
- .reduceLeftOption(And)
+ val exprNode = FilterHandler
+
.combineConjuncts(filterExprs().map(ExpressionConverter.replaceAttributeReference))
.map(ExpressionConverter.replaceWithExpressionTransformer(_, output))
.map(_.doTransform(context))
.orNull
diff --git
a/gluten-ut/spark42/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark42/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index c85c1ceac9..adce5b7beb 100644
---
a/gluten-ut/spark42/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark42/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -809,6 +809,11 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("Give up splitting aggregate code if a parameter length goes over
the limit")
.exclude("Give up splitting subexpression code if a parameter length goes
over the limit")
.exclude("SPARK-47238: Test broadcast threshold for generated code")
+ .exclude("SPARK-56431: CSE in FilterExec preserves null guards for notNull
columns")
+ .exclude("SPARK-56032: FilterExec skips CSE codegen when there is no
common subexpression")
+ .exclude("SPARK-56032: FilterExec skips CSE codegen when the common
subexpression is cheap")
+ .exclude(
+ "SPARK-56032: FilterExec takes CSE codegen when the common subexpression
is non-cheap")
enableSuite[GlutenBroadcastExchangeSuite]
.exclude("SPARK-52962: broadcast exchange should not reset metrics") //
Add Gluten test
enableSuite[GlutenLocalBroadcastExchangeSuite]
diff --git
a/gluten-ut/spark42/src/test/scala/org/apache/spark/sql/execution/GlutenWholeStageCodegenSuite.scala
b/gluten-ut/spark42/src/test/scala/org/apache/spark/sql/execution/GlutenWholeStageCodegenSuite.scala
index 5a43c1846b..29329d5e60 100644
---
a/gluten-ut/spark42/src/test/scala/org/apache/spark/sql/execution/GlutenWholeStageCodegenSuite.scala
+++
b/gluten-ut/spark42/src/test/scala/org/apache/spark/sql/execution/GlutenWholeStageCodegenSuite.scala
@@ -23,14 +23,12 @@ import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, GlutenSQLTestsTrait, Row, SaveMode}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
+import org.apache.spark.sql.types.{DecimalType, IntegerType, StringType,
StructType}
+
+import java.math.{BigDecimal => JBigDecimal}
import scala.reflect.ClassTag
-// GLUTEN-12569 Spark 4.2 UT enablement: disabled whole suite - a test
triggers a
-// StackOverflowError in Gluten ExpressionConverter.transformExpression that
aborts the
-// entire group2 CI run. Disabled here so the rest of the group can run and be
triaged.
[email protected]
class GlutenWholeStageCodegenSuite extends WholeStageCodegenSuite with
GlutenSQLTestsTrait {
import testImplicits._
@@ -764,4 +762,78 @@ class GlutenWholeStageCodegenSuite extends
WholeStageCodegenSuite with GlutenSQL
}
}
}
+
+ testGluten("SPARK-56431: CSE in FilterExec preserves null guards for notNull
columns") {
+ val schema = new StructType()
+ .add("a", DecimalType(10, 2))
+ .add("b", DecimalType(10, 2))
+ val data = spark.sparkContext.parallelize(
+ Seq(
+ Row(null, new JBigDecimal("100.00")),
+ Row(new JBigDecimal("50.00"), new JBigDecimal("200.00")),
+ Row(new JBigDecimal("80.00"), new JBigDecimal("100.00"))))
+ withSQLConf(
+ SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false"
+ ) {
+ val filtered = spark
+ .createDataFrame(data, schema)
+ .where("a IS NOT NULL AND abs(b - a) > 0.10 AND abs(b - a) < 1000.00")
+ assertWholeStageContains[FilterExecTransformerBase](filtered)
+ checkAnswer(
+ filtered,
+ Seq(
+ Row(new JBigDecimal("50.00"), new JBigDecimal("200.00")),
+ Row(new JBigDecimal("80.00"), new JBigDecimal("100.00"))))
+ }
+ }
+
+ private def checkFilterWithAndWithoutCSE(
+ data: Seq[Row],
+ schema: StructType,
+ condition: String,
+ expected: Seq[Row]): Unit = {
+ Seq(true, false).foreach {
+ cseEnabled =>
+ withSQLConf(
+ SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> cseEnabled.toString,
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false"
+ ) {
+ val filtered =
+ spark.createDataFrame(spark.sparkContext.parallelize(data),
schema).where(condition)
+ assertWholeStageContains[FilterExecTransformerBase](filtered)
+ checkAnswer(filtered, expected)
+ }
+ }
+ }
+
+ testGluten("SPARK-56032: FilterExec skips CSE codegen when there is no
common subexpression") {
+ checkFilterWithAndWithoutCSE(
+ Seq(Row(1, 5), Row(null, 3), Row(4, null), Row(5, 6), Row(7, 8), Row(2,
3)),
+ new StructType().add("a", IntegerType).add("b", IntegerType),
+ "a IS NOT NULL AND a > 4 AND b > 4",
+ Seq(Row(5, 6), Row(7, 8))
+ )
+ }
+
+ testGluten("SPARK-56032: FilterExec skips CSE codegen when the common
subexpression is cheap") {
+ checkFilterWithAndWithoutCSE(
+ Seq(Row(Row(10)), Row(Row(3)), Row(Row(200)), Row(Row(50)),
Row(Row(null)), Row(null)),
+ new StructType().add("s", new StructType().add("x", IntegerType)),
+ "s.x > 5 AND s.x < 100",
+ Seq(Row(Row(10)), Row(Row(50)))
+ )
+ }
+
+ testGluten(
+ "SPARK-56032: FilterExec takes CSE codegen when the common subexpression
is non-cheap") {
+ checkFilterWithAndWithoutCSE(
+ Seq(Row(1, 5), Row(60, 50), Row(10, 20), Row(0, 0), Row(null, 5)),
+ new StructType().add("a", IntegerType).add("b", IntegerType),
+ "(a + b) > 0 AND (a + b) < 100",
+ Seq(Row(1, 5), Row(10, 20))
+ )
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
