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 4b4c443f4df9 [SPARK-56032][SQL][FOLLOWUP] Skip FilterExec CSE codegen
when every common subexpression is cheap
4b4c443f4df9 is described below
commit 4b4c443f4df936291f4f9949a140a103830c42c7
Author: Wenchen Fan <[email protected]>
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 <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
---
.../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 87500f0ca514..077d6cbe8d47 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: [email protected]
For additional commands, e-mail: [email protected]
