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/06/21 14:34:32 UTC
(spark) branch branch-4.x updated: [SPARK-56032][SQL][FOLLOWUP] Skip FilterExec CSE codegen when every common subexpression is cheap
This is an automated email from the ASF dual-hosted git repository.
uros-b 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 a8547910df12 [SPARK-56032][SQL][FOLLOWUP] Skip FilterExec CSE codegen when every common subexpression is cheap
a8547910df12 is described below
commit a8547910df125b2834799ca4e29a702cf1d54fb5
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Sun Jun 21 16:34:10 2026 +0200
[SPARK-56032][SQL][FOLLOWUP] Skip FilterExec CSE codegen when every common subexpression is cheap
### What changes were proposed in this pull request?
A follow-up of #54862 (which introduced subexpression elimination (CSE) in `FilterExec`
whole-stage codegen) and #56209 (which gated the CSE path on whether `otherPreds` contain a
common subexpression).
The #56209 gate takes the CSE path whenever `otherPredsEquivalentExpressions.getCommonSubexpressions`
is non-empty. That correctly excludes the no-shared-subexpression case, but it still takes the CSE
path when the only shared subexpressions are **cheap**. The realistic case is a shared non-leaf slot
read such as a struct field access: `s.x > 5 AND s.x < 100` shares `GetStructField(s, x)`. Caching
such a cheap read gains nothing -- the non-CSE path already loads each column lazily into a variable
on demand, just before the predicate that needs it -- so taking the CSE path only adds the eager
`inputVarsEvalCode` prologue, which evaluates **every** column referenced by `otherPreds` at the top
of the per-row loop and defeats the short-circuiting the non-CSE path gets for free.
This PR requires a **non-cheap** common subexpression (per `CollapseProject.isCheap`) before taking
the CSE path. Filters with a genuine repeated computation (e.g. `a + b`) are unaffected and still
benefit from CSE.
`CollapseProject.isCheap` is the canonical "cheap to recompute" predicate (attributes, foldables,
`Alias`/`ExtractValue` of cheap children). The gate runs on predicates already bound for codegen, so
this PR also teaches `isCheap` that `BoundReference` -- the codegen-bound form of an `Attribute`, an
equally cheap slot read -- is cheap. Without that, `GetStructField(BoundReference, ...)` would not be
recognized as cheap (an `ExtractValue` is only cheap when its child is), and the struct-field filter
above would still wrongly take the CSE path. Reusing `isCheap` here also lets the gate share the
single `EquivalentExpressions` analysis it already builds for the CSE codegen rather than
re-analyzing the predicates.
Note that bare columns never reach this gate in the first place: `EquivalentExpressions` skips
`LeafExpression`s (`BoundReference`/`Attribute`), and `splitConjunctivePredicates` feeds each
conjunct to a separate `addExprTree` call, so a column repeated across conjuncts is never recorded
as a common subexpression. The cheap case this gate filters out is therefore the shared **non-leaf**,
with struct field access as the realistic example.
### Why are the changes needed?
When the only common subexpressions are cheap, the CSE path's eager prologue is pure overhead: it
decodes every referenced column up front, including columns needed only by a later predicate that an
earlier cheaper predicate would have short-circuited past. For `s.x > 5 AND s.x < 100`, the #56209
gate takes the CSE path solely because `GetStructField(s, x)` is shared, then evaluates the struct
field for every row at the top of the loop instead of lazily after the first comparison fails.
Requiring a non-cheap common subexpression keeps such filters on the lazy, short-circuiting path
while preserving CSE for genuine repeated computation -- completing the intent of the #56209 gate.
### Does this PR introduce _any_ user-facing change?
No. This is a codegen-only change; query results are unchanged.
### How was this patch tested?
Two new unit tests in `WholeStageCodegenSuite`:
- A **cheap non-leaf** common subexpression (`s.x > 5 AND s.x < 100` over `struct<x:int>`, sharing
`GetStructField`): CSE-enabled generated code is identical to CSE-disabled code -- i.e. it falls
back to the lazy, short-circuiting non-CSE path. This shape took the CSE path before this change,
so the test genuinely pins the new behavior.
- A **non-cheap** common subexpression (`(a + b) > 0 AND (a + b) < 100`, sharing `Add`): CSE-enabled
code differs from CSE-disabled code -- i.e. the gate still takes the CSE path that computes the
shared result once.
The existing `FilterExec` CSE tests, which use genuine non-cheap common subexpressions, still
exercise the CSE path and pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude (Claude Code)
Closes #56604 from cloud-fan/filterexec-cse-leaf-gate.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
(cherry picked from commit 4b4c443f4df936291f4f9949a140a103830c42c7)
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../spark/sql/catalyst/optimizer/Optimizer.scala | 9 ++-
.../sql/execution/basicPhysicalOperators.scala | 17 ++++-
.../sql/execution/WholeStageCodegenSuite.scala | 72 ++++++++++++++++++++++
3 files changed, 96 insertions(+), 2 deletions(-)
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 51c6a3399b39..6753dd9a723c 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
@@ -1559,9 +1559,16 @@ object CollapseProject extends Rule[LogicalPlan] with AliasHelper {
/**
* Check if the given expression is cheap that we can inline it.
+ *
+ * This is consumed both by logical-stage callers (which only ever see `Attribute`) and by the
+ * `FilterExec` whole-stage-codegen CSE gate, which runs on predicates already bound for codegen
+ * and so sees `BoundReference` instead. The `BoundReference` branch therefore only fires on the
+ * codegen path -- logical plans never carry `BoundReference` -- and leaves the logical callers
+ * unaffected.
*/
def isCheap(e: Expression): Boolean = e match {
- case _: Attribute | _: OuterReference => true
+ // `BoundReference` is the codegen-bound form of an `Attribute`; a slot read, equally cheap.
+ case _: Attribute | _: OuterReference | _: BoundReference => true
case _ if e.foldable => true
// PythonUDF is handled by the rule ExtractPythonUDFs
case _: PythonUDF =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
index 88c74ab7adc4..8c96f1ff9579 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
@@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.BindReferences.bindReferences
import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.optimizer.CollapseProject
import org.apache.spark.sql.catalyst.plans.physical._
import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
@@ -316,12 +317,26 @@ case class FilterExec(condition: Expression, child: SparkPlan)
// (e.g. decoding a decimal column for rows a cheaper earlier predicate would reject), so we
// fall back to `generatePredicateCode`.
//
+ // A *cheap* common subexpression does not count either. Caching a cheap load saves nothing:
+ // the non-CSE path already loads each column lazily into a variable on demand, so taking the
+ // CSE path for it would only add the eager prologue that decodes every referenced column up
+ // front. Note bare columns never reach this point: `EquivalentExpressions` skips
+ // `LeafExpression`s (which includes `BoundReference`/`Attribute`), and
+ // `splitConjunctivePredicates` feeds each conjunct to a separate `addExprTree` call, so a
+ // column repeated across conjuncts (e.g. the `c >= lo` / `c <= hi` that `c BETWEEN lo AND hi`
+ // lowers to) is never recorded as a common subexpression. The cheap-but-recorded case is a
+ // shared *non-leaf* such as a struct field access -- `s.x > 5 AND s.x < 100` shares
+ // `GetStructField(s, x)` -- which is just a slot read. Require a non-cheap common
+ // subexpression (per `CollapseProject.isCheap`) so such filters keep the lazy,
+ // short-circuiting path and only genuine repeated computation takes the CSE path.
+ //
// `subexpressionElimination.filterExec.enabled` additionally gates this path so it can be
// turned off independently of subexpression elimination elsewhere.
val (prologueCode, predicateCode) =
if (conf.subexpressionEliminationEnabled && conf.subexpressionEliminationFilterExecEnabled &&
otherPreds.nonEmpty &&
- otherPredsEquivalentExpressions.getCommonSubexpressions.nonEmpty) {
+ otherPredsEquivalentExpressions.getCommonSubexpressions
+ .exists(!CollapseProject.isCheap(_))) {
// Pre-evaluate input variables before CSE analysis: CSE clears
// ctx.currentVars[i].code as a side effect; without this pre-evaluation, Janino
// fails when otherPreds reference the same input columns that CSE already
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
index 886df9184aca..bcd2f5369932 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
@@ -1225,6 +1225,78 @@ class WholeStageCodegenSuite extends SharedSparkSession
"CSE-disabled codegen (i.e. fall back to the lazy, short-circuiting non-CSE path)")
}
+ test("SPARK-56032: FilterExec skips CSE codegen when the common subexpression is cheap") {
+ // A column repeated across conjuncts never becomes a common subexpression -- a bare column is a
+ // `LeafExpression`, which `EquivalentExpressions` skips, and `splitConjunctivePredicates` feeds
+ // each conjunct to a separate `addExprTree` call. The realistic cheap-but-recorded case is a
+ // shared *non-leaf* slot read such as a struct field access: `s.x > 5 AND s.x < 100` shares
+ // `GetStructField(s, x)`. Caching that gains nothing over the non-CSE path's lazy load, so the
+ // gate must fall back. (Pre-`isCheap`-gate this took the CSE path, emitting the eager
+ // prologue.)
+ val schema = StructType(Seq(
+ StructField("s", StructType(Seq(StructField("x", IntegerType, nullable = true))),
+ nullable = true)))
+ val data = spark.sparkContext.parallelize(Seq(
+ Row(Row(10)), Row(Row(3)), Row(Row(200)), Row(Row(50)), Row(Row(null)), Row(null)))
+ val expected = Seq(Row(Row(10)), Row(Row(50)))
+
+ def filterCode(cseEnabled: Boolean): String = {
+ withSQLConf(
+ SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> cseEnabled.toString,
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val df = spark.createDataFrame(data, schema)
+ // Both conjuncts share `GetStructField(s, x)`, a cheap non-leaf common subexpression.
+ val filtered = df.where("s.x > 5 AND s.x < 100")
+ val plan = filtered.queryExecution.executedPlan
+ assert(plan.exists(_.isInstanceOf[WholeStageCodegenExec]),
+ "Filter should be in whole-stage codegen")
+ checkAnswer(filtered, expected)
+ codegenString(plan)
+ }
+ }
+
+ def normalize(code: String): String = code.replaceAll("#\\d+", "#")
+ assert(normalize(filterCode(cseEnabled = true)) == normalize(filterCode(cseEnabled = false)),
+ "With only a cheap common subexpression, CSE-enabled FilterExec codegen should be " +
+ "identical to CSE-disabled codegen (i.e. fall back to the lazy, short-circuiting " +
+ "non-CSE path)")
+ }
+
+ test("SPARK-56032: FilterExec takes CSE codegen when the common subexpression is non-cheap") {
+ // The dual of the cheap-subexpression test: when `otherPreds` share a genuinely non-cheap
+ // computation (`a + b`, whose `isCheap` is false), the gate must take the CSE path so the
+ // shared result is computed once. Verify the CSE-enabled code differs from CSE-disabled here,
+ // pinning down that the gate still fires for real repeated computation.
+ val schema = StructType(Seq(
+ StructField("a", IntegerType, nullable = true),
+ StructField("b", IntegerType, nullable = true)))
+ val data = spark.sparkContext.parallelize(Seq(
+ Row(1, 5), Row(60, 50), Row(10, 20), Row(0, 0), Row(null, 5)))
+ val expected = Seq(Row(1, 5), Row(10, 20))
+
+ def filterCode(cseEnabled: Boolean): String = {
+ withSQLConf(
+ SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> cseEnabled.toString,
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val df = spark.createDataFrame(data, schema)
+ // Both conjuncts share `a + b`, a non-cheap common subexpression worth eliminating.
+ val filtered = df.where("(a + b) > 0 AND (a + b) < 100")
+ val plan = filtered.queryExecution.executedPlan
+ assert(plan.exists(_.isInstanceOf[WholeStageCodegenExec]),
+ "Filter should be in whole-stage codegen")
+ checkAnswer(filtered, expected)
+ codegenString(plan)
+ }
+ }
+
+ def normalize(code: String): String = code.replaceAll("#\\d+", "#")
+ assert(normalize(filterCode(cseEnabled = true)) != normalize(filterCode(cseEnabled = false)),
+ "With a non-cheap common subexpression, CSE-enabled FilterExec codegen should differ from " +
+ "CSE-disabled codegen (i.e. take the CSE path that computes the shared result once)")
+ }
+
test("SPARK-56032: subexpressionElimination.filterExec.enabled gates FilterExec CSE " +
"independently of subexpression elimination") {
// The conf disables CSE specifically for FilterExec while leaving subexpression elimination
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org