You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ho...@apache.org on 2026/09/26 02:45:05 UTC
(spark) branch master updated: [SPARK-55014][SQL] Split projection for expensive filters
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits 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 9b37040a25db [SPARK-55014][SQL] Split projection for expensive filters
9b37040a25db is described below
commit 9b37040a25dbe431150e442832cd31438f5061e8
Author: Holden Karau <ho...@pigscanfly.ca>
AuthorDate: Fri Sep 25 19:44:51 2026 -0700
[SPARK-55014][SQL] Split projection for expensive filters
### What changes were proposed in this pull request?
Follow up to SPARK-47672. That change stopped `PushPredicateThroughNonJoin` from pushing a filter below a `Project` when the filter references expensive projected aliases, because pushing caused them to be evaluated twice, and parked those conditions above the projection.
This splits the projection into a stack of `Project`/`Filter` layers instead, so that the expensive expressions a condition does not need are only evaluated for the rows that condition kept:
Filter f AND g Filter g
Project a, rlike(e,'magic') AS f, Project a, f, rlike(e,'other') AS g
rlike(e,'other') AS g --> Filter f
child Project a, e, rlike(e,'magic') AS f
child
Conditions are grouped by the aliases they reference, and the group needing the fewest not-yet-computed aliases is split off first, breaking ties towards the condition written first. Two restrictions keep the rule from costing more than it saves:
- A projection is only split off while doing so leaves an expensive alias for a later layer. An extra operator has a per-row cost of its own, so it has to buy a real deferral.
- Aliases sharing an expensive sub-expression are treated as one indivisible unit. Subexpression elimination works within a single projection and cannot reach across a filter, so splitting such aliases apart would evaluate the shared part once for every row below the filter and again for every row that survived it. A struct-returning UDF read field by field is the common shape.
Gated on the new `spark.sql.optimizer.splitProjectionForExpensiveFilters` (default true), which also requires `spark.sql.optimizer.avoidDoubleFilterEval`.
### Why are the changes needed?
SPARK-47672 stops the double evaluation but leaves the projection computing every expensive element for every row reaching it, even when a filter above it is about to discard most of them. Whole stage codegen already defers a projected expression past a filter that does not need it (`CodegenSupport.evaluateRequiredVariables`), so what this buys is the same saving on the paths codegen does not cover: interpreted projections, operators it bails out of, and Python UDF evaluation.
### Does this PR introduce _any_ user-facing change?
Plans for queries filtering on expensive projected expressions gain projection/filter layers. Results, schemas and column ordering are unchanged.
### How was this patch tested?
Thirteen new plan tests in `FilterPushdownSuite` covering splitting, the ordering heuristic, the no-split cases (a condition needing every expensive element, only cheap elements left to defer, shared expensive work), column ordering, and the config off. Three new end-to-end tests in `DataFrameSuite` count UDF invocations to show the saving is real (30 -> 15), that results match with splitting and codegen each on and off, that the split follows the filter's order rather than the projec [...]
`PlanStabilitySuite` confirms the TPCDS golden plans are unaffected.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5 using my previous work in the original PR as a starting point & Cursor with Kimi K3
Closes #58403 from holdenk/SPARK-55014-split-projection-for-expensive-filters.
Lead-authored-by: Holden Karau <ho...@pigscanfly.ca>
Co-authored-by: Holden Karau <ho...@snowflake.com>
Signed-off-by: Holden Karau <ho...@snowflake.com>
---
.../spark/sql/catalyst/optimizer/Optimizer.scala | 179 +++++++++++--
...ushdownPredicatesAndPruneColumnsForCTEDef.scala | 10 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 14 +
.../catalyst/optimizer/FilterPushdownSuite.scala | 281 +++++++++++++++++++++
.../org/apache/spark/sql/DataFrameSuite.scala | 96 +++++++
5 files changed, 558 insertions(+), 22 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 3f012689c43e..7ba4493c0961 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
@@ -2581,8 +2581,11 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
// resulting in double evaluation, but only of inexpensive items -- worth it to filter
// records sooner.
// (Case 1 & 2 are treated as "cheap" predicates)
- // 3) When an a filter references expensive to compute references we do not push it.
- // Note that a given filter may contain parts (sepereated by logical ands) from all cases.
+ // 3) When a filter references expensive-to-compute references we do not push it. Instead we
+ // split the projection around it (see `splitProjectForExpensiveConditions`); whatever is
+ // left over stays above the projection, which is where all of case 3 lands when nothing
+ // is worth splitting.
+ // Note that a given filter may contain parts (separated by logical ands) from all cases.
// We handle each part separately according to the logic above.
// Additional restriction:
// SPARK-13473: We can't push the predicate down when the underlying projection output non-
@@ -2624,26 +2627,44 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
false
}
}
- // Short circuit if we do not have any cheap filters return the original filter as is.
- if (cheapWithUsed.isEmpty) {
+ // Cheap filters (cases 1 & 2) pushed below the projection.
+ val baseChild: LogicalPlan = if (cheapWithUsed.nonEmpty) {
+ val combinedCheapFilter = cheapWithUsed.map(_._3).reduce(And)
+ Filter(combinedCheapFilter, child = grandChild)
+ } else {
+ grandChild
+ }
+ // Case 3: split the projection around the expensive conditions. Non-deterministic
+ // conditions stay in the top filter, as they do everywhere else in this rule -- the
+ // split orders conditions by cost, and cost is no way to decide how many rows a
+ // non-deterministic expression sees.
+ val (splitChild, splitAliases, stayUp) =
+ if (expensiveWithUsed.nonEmpty && SQLConf.get.splitProjectionForExpensiveFilters) {
+ val (deterministic, nonDeterministic) =
+ expensiveWithUsed.partition { case (cond, _, _) => cond.deterministic }
+ val (child, aliases, unplaced) = splitProjectForExpensiveConditions(
+ project, aliasMap, baseChild,
+ deterministic.map { case (cond, used, _) => (cond, used) })
+ (child, aliases, unplaced ++ nonDeterministic.map(_._1))
+ } else {
+ (baseChild, AttributeSet.empty, expensiveWithUsed.map(_._1))
+ }
+ if (cheapWithUsed.isEmpty && splitAliases.isEmpty) {
f
} else {
- val cheap: Seq[Expression] = cheapWithUsed.map(_._3)
- // Make a base instance which has all of the cheap filters pushed down.
- // For all filter which do not reference any expensive aliases then
- // just push the filter while resolving the non-expensive aliases.
- val combinedCheapFilter = cheap.reduce(And)
- val baseChild = Filter(combinedCheapFilter, child = grandChild)
- // Take our projection and place it on top of the pushed filters.
- val topProjection = project.copy(child = baseChild)
-
- // If we pushed all the filters we can return the projection
- if (expensiveWithUsed.isEmpty) {
+ // Reference the aliases the split already computed; only the aliases change so the
+ // column ordering (and schema) is preserved.
+ val topProjectList = if (splitAliases.isEmpty) {
+ fields
+ } else {
+ fields.map(field => if (splitAliases.contains(field)) field.toAttribute else field)
+ }
+ val topProjection = project.copy(projectList = topProjectList, child = splitChild)
+
+ if (stayUp.isEmpty) {
topProjection
} else {
- // Finally add any filters which could not be pushed
- val remainingConditions = expensiveWithUsed.map(_._1)
- Filter(remainingConditions.reduce(And), topProjection)
+ Filter(stayUp.reduce(And), topProjection)
}
}
}
@@ -2803,6 +2824,128 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
}
}
+ /**
+ * Splits `project` into a stack of [[Project]]s with the expensive conditions in
+ * `expensiveConds` between them, so an expensive expression only runs on the rows the filters
+ * below it kept (SPARK-55014). For example, with both regexes considered expensive and a child
+ * producing `a` and `e`:
+ *
+ * {{{
+ * Filter f AND g
+ * Project a, rlike(e, 'magic') AS f, rlike(e, 'other') AS g
+ * child
+ * }}}
+ * becomes
+ * {{{
+ * Filter g
+ * Project a, f, rlike(e, 'other') AS g
+ * Filter f
+ * Project a, e, rlike(e, 'magic') AS f
+ * child
+ * }}}
+ * so `rlike(e, 'other')` only runs on the rows where `rlike(e, 'magic')` was true.
+ *
+ * Conditions are grouped by the aliases they reference, then the group needing the fewest
+ * not-yet-computed aliases is split off first -- least demanding first keeps the expensive
+ * expressions as high in the stack, and so over as few rows, as we can manage. On a tie, the
+ * group making the most conditions evaluable wins, falling back to the condition written first
+ * (the only selectivity signal we have). Whole-stage codegen already defers a projected
+ * expression past a filter that does not need it (`CodegenSupport.evaluateRequiredVariables`),
+ * so this rule only buys the same saving on the paths codegen does not cover -- interpreted
+ * projections, operators it bails out of, Python UDFs.
+ *
+ * Two restrictions keep the split from costing more than it saves:
+ * - Only split a [[Project]] off while it leaves an expensive alias for a later layer; an extra
+ * operator has to buy a real deferral.
+ * - Aliases sharing an expensive sub-expression move together. Subexpression elimination
+ * works within one projection and cannot reach across a [[Filter]], so splitting them apart
+ * would re-evaluate the shared part on every row below the filter and again on every
+ * survivor (a struct-returning UDF read field by field is the common shape).
+ *
+ * The most demanding group is left un-placed for the caller to put above the projection --
+ * where expensive conditions go when there is nothing worth splitting.
+ *
+ * Callers must pass deterministic conditions only -- the grouping below orders conditions by
+ * cost, and cost is no way to decide how many rows a non-deterministic expression sees.
+ *
+ * Returns the new plan, the alias attributes it has already computed, and the un-placed
+ * conditions (in their original, alias-referencing form).
+ */
+ private def splitProjectForExpensiveConditions(
+ project: Project,
+ aliasMap: AttributeMap[Alias],
+ baseChild: LogicalPlan,
+ expensiveConds: Seq[(Expression, AttributeMap[Alias])])
+ : (LogicalPlan, AttributeSet, Seq[Expression]) = {
+ val expensiveAliases = AttributeSet(
+ aliasMap.collect { case (attr, alias) if alias.child.expensive => attr })
+ // Expensive sub-expressions behind each expensive alias, to spot when two aliases are the
+ // same piece of work. Cheap aliases have none, so we skip the (possibly wide) cheap rest.
+ val expensivePartsOf = AttributeMap(expensiveAliases.toSeq.map { attr =>
+ attr -> aliasMap(attr).child.collect { case e if e.expensive => e.canonicalized }.toSet
+ })
+ def expensiveParts(attr: Attribute): Set[Expression] =
+ expensivePartsOf.getOrElse(attr, Set.empty)
+ val aliasesSharing = expensivePartsOf.toSeq
+ .flatMap { case (attr, parts) => parts.map(part => part -> attr) }
+ .groupBy(_._1)
+ .map { case (part, pairs) => part -> AttributeSet(pairs.map(_._2)) }
+ // Grow a set of aliases into the whole unit of shared expensive work it belongs to,
+ // transitively (see the docstring restriction on shared sub-expressions).
+ def unitOf(aliases: AttributeSet): AttributeSet = {
+ var unit = aliases
+ var grew = true
+ while (grew) {
+ val next = unit.flatMap(expensiveParts).foldLeft(unit) {
+ case (acc, part) => acc ++ aliasesSharing(part)
+ }
+ grew = next.size > unit.size
+ unit = next
+ }
+ unit
+ }
+ val indexed = expensiveConds.zipWithIndex.map {
+ case ((cond, used), idx) => (unitOf(AttributeSet(used.keys)), idx, cond)
+ }
+ // Group conditions over the same aliases; group on sorted exprIds (not AttributeSet) for
+ // stable hashing, and order groups by first condition to keep plans stable.
+ var pending: Seq[(AttributeSet, Seq[(Int, Expression)])] = indexed
+ .groupBy { case (used, _, _) => used.toSeq.map(_.exprId.id).sorted }
+ .toSeq
+ .map { case (_, group) => (group.head._1, group.map { case (_, idx, cond) => (idx, cond) }) }
+ .sortBy { case (_, conds) => conds.head._1 }
+ def conditionsOf(groups: Seq[(AttributeSet, Seq[(Int, Expression)])]): Seq[Expression] =
+ groups.flatMap(_._2).sortBy(_._1).map(_._2)
+
+ // Projection order is stable, unlike the alias map's iteration order.
+ val orderedAliases = project.projectList.collect { case a: Alias => a }
+
+ var plan = baseChild
+ var computed = AttributeSet.empty
+ var searching = true
+ while (searching) {
+ // Only split while it leaves an expensive alias for a later layer.
+ val stillToCompute = expensiveAliases -- computed
+ val candidates = pending.filter { case (used, _) => !stillToCompute.subsetOf(used) }
+ if (candidates.isEmpty) {
+ searching = false
+ } else {
+ val (bestUsed, _) = candidates.minBy {
+ case (used, conds) => ((used -- computed).size, -conds.size)
+ }
+ val newAliases = orderedAliases.filter(a => bestUsed.contains(a) && !computed.contains(a))
+ computed ++= AttributeSet(newAliases.map(_.toAttribute))
+ // Keep the plan's existing outputs available for later projections; column pruning drops
+ // the extras.
+ val newProject = project.copy(projectList = plan.output ++ newAliases, child = plan)
+ val (evaluable, rest) = pending.partition { case (used, _) => used.subsetOf(computed) }
+ plan = Filter(conditionsOf(evaluable).reduce(And), newProject)
+ pending = rest
+ }
+ }
+ (plan, computed, conditionsOf(pending))
+ }
+
/**
* Check if we can safely push a filter through a projection, by making sure that predicate
* subqueries in the condition do not contain the same attributes as the plan they are moved
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala
index d33384152e8b..b7e875554400 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala
@@ -191,10 +191,12 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] with
* the very same `AliasHelper` utilities, so the two cannot drift apart): through projection
* and grouping-key aliases, positionally into each branch (`Union`), and unchanged
* through operators that pass the referenced attributes verbatim (`Join`, `Window`, and
- * output-preserving unary nodes like `Filter`, `Sort`, `Repartition`). Descent stops at
- * operators that remap attributes in other ways (e.g. `Generate`, `Expand`): if the filter
- * cannot be located, the input plan is returned unchanged, and the caller re-pushes on top,
- * which is redundant but always semantics-preserving.
+ * output-preserving unary nodes like `Filter`, `Sort`, `Repartition`). The descent is recursive
+ * because the filter can end up several operators down -- in particular between the projections
+ * `PushPredicateThroughNonJoin` splits a `Project` into for expensive projected expressions.
+ * Descent stops at operators that remap attributes in other ways (e.g. `Generate`, `Expand`):
+ * if the filter cannot be located, the input plan is returned unchanged, and the caller
+ * re-pushes on top, which is redundant but always semantics-preserving.
*
* Ancestors of the removed filter are rebuilt with `withNewChildren` rather than direct
* case-class copies so that `TreeNode` tags (e.g. `Project.hiddenOutputTag`, which the
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 0dde915fa80f..be19322ff495 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -610,6 +610,17 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS =
+ buildConf("spark.sql.optimizer.splitProjectionForExpensiveFilters")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .doc("When true, split a projection into several projections with the filters which " +
+ "reference its expensive (UDF, etc.) elements in between, so an expensive element is only " +
+ "evaluated for the rows which survived the filters below it. Costs an extra operator per " +
+ s"split, and only applies when ${AVOID_DOUBLE_FILTER_EVAL.key} is also enabled.")
+ .version("4.3.0")
+ .booleanConf
+ .createWithDefault(true)
+
val ATTEMPT_TRANSPILATION_OF_PYTHON_UDFS =
buildConf("spark.sql.experimental.optimizer.transpilePyUDFs")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
@@ -9880,6 +9891,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def avoidDoubleFilterEval: Boolean = getConf(AVOID_DOUBLE_FILTER_EVAL)
+ def splitProjectionForExpensiveFilters: Boolean =
+ getConf(SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS)
+
def structPredicateDecomposeEnabled: Boolean = getConf(STRUCT_PREDICATE_DECOMPOSE_ENABLED)
def structPredicateDecomposeMaxFields: Int = getConf(STRUCT_PREDICATE_DECOMPOSE_MAX_FIELDS)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
index e21798eeea0f..601f8e36f624 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
@@ -61,6 +61,8 @@ class FilterPushdownSuite extends PlanTest {
val testStringRelation = LocalRelation(attrA, attrB, attrE)
+ val testArrayStringRelation = LocalRelation($"a".int, $"s_arr".array(StringType))
+
val simpleDisjunctivePredicate =
("x.a".attr > 3) && ("y.a".attr > 13) || ("x.a".attr > 1) && ("y.a".attr > 11)
val expectedPredicatePushDownResult = {
@@ -333,6 +335,285 @@ class FilterPushdownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}
+ test("SPARK-55014: split the projection so a second expensive element sees fewer rows") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"b")
+ .where($"f" && $"g")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // Both regexes are expensive and neither condition needs both, so f is computed first and
+ // 'other' only runs on the rows 'magic' matched.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f")
+ .where($"f")
+ .select($"a", $"f", $"e".rlike("other") as "g", $"b")
+ .where($"g")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: split the projection even for a single expensive condition") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"b")
+ .where($"f")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // g is never filtered on but is still expensive, so it is worth deferring past the filter on f.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f")
+ .where($"f")
+ .select($"a", $"f", $"e".rlike("other") as "g", $"b")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: split the projection once per group of expensive elements") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"e".rlike("third") as
+ "j")
+ .where($"f" && $"g" && $"j")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // One projection per condition except the last -- nothing left to defer past it.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f")
+ .where($"f")
+ .select($"a", $"b", $"e", $"f", $"e".rlike("other") as "g")
+ .where($"g")
+ .select($"a", $"f", $"g", $"e".rlike("third") as "j")
+ .where($"j")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: a condition needing every expensive element can not be split off") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"b")
+ .where($"f" || $"g")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // Splitting here would compute both regexes below the filter -- the plan we already have
+ // plus a redundant projection.
+ val correctAnswer = originalQuery
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: a condition over several expensive elements can still be split off") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"e".rlike("third") as
+ "j")
+ .where($"f" || $"g")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // A third expensive element to defer, so f and g compute together below the filter and j
+ // waits above it.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f", $"e".rlike("other") as "g")
+ .where($"f" || $"g")
+ .select($"a", $"f", $"g", $"e".rlike("third") as "j")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: the least demanding condition goes lowest") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"e".rlike("third") as
+ "j")
+ .where(($"f" || $"g") && $"j")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // j needs one alias and (f || g) needs two, so j goes first even though it is projected
+ // last -- leaving two expensive elements to run on its survivors.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("third") as "j")
+ .where($"j")
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"j")
+ .where($"f" || $"g")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: split the projection while also pushing the cheap conditions") {
+ val originalQuery = testStringRelation
+ .select($"a" + $"b" as "sum", $"e".rlike("magic") as "f", $"e".rlike("other") as "g")
+ .where($"sum" > 10 && $"f")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // sum is cheap (case 2, pushed down); f is expensive (split around); g deferred on top.
+ val correctAnswer = testStringRelation
+ .where($"a" + $"b" > 10)
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f")
+ .where($"f")
+ .select($"a" + $"b" as "sum", $"f", $"e".rlike("other") as "g")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: no split when the only elements left to defer are cheap") {
+ val originalQuery = testStringRelation
+ .select($"a" + $"b" as "sum", $"e".rlike("magic") as "f")
+ .where($"f")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // A projection is not free; recomputing a cheap sum later is not worth an extra operator.
+ val correctAnswer = originalQuery
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: the projection on top keeps the original column ordering") {
+ val originalQuery = testStringRelation
+ .select($"b", $"e".rlike("magic") as "f", $"a", $"e".rlike("other") as "g")
+ .where($"f" && $"g")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // f is computed below but must come back out in its original position.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f")
+ .where($"f")
+ .select($"b", $"f", $"a", $"e".rlike("other") as "g")
+ .where($"g")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ assert(optimized.schema === originalQuery.schema)
+ }
+
+ test("SPARK-55014: all the conditions over the same elements are applied together") {
+ val joined = ArrayJoin($"s_arr", Literal(","), None)
+ val otherJoined = ArrayJoin($"s_arr", Literal("-"), None)
+ val originalQuery = testArrayStringRelation
+ .select($"a", joined as "j", otherJoined as "k")
+ .where($"j" > "a" && $"j" < "z")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // Conditions over the same aliases run at the same point of the stack, so both on j land in
+ // one filter and k is deferred past them.
+ val correctAnswer = testArrayStringRelation
+ .select($"a", $"s_arr", joined as "j")
+ .where($"j" > "a" && $"j" < "z")
+ .select($"a", $"j", otherJoined as "k")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: no split when the expensive work is shared between elements") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", Not($"e".rlike("magic")) as "g")
+ .where($"f" && $"g")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // f and g are the same regex. Subexpression elimination collapses it in one projection but
+ // cannot reach across a filter, so splitting would run it on every row below and again on
+ // every survivor -- more work than leaving the projection alone.
+ val correctAnswer = originalQuery
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: elements sharing expensive work are split off together") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", Not($"e".rlike("magic")) as "g",
+ $"e".rlike("other") as "j")
+ .where($"f" && $"j")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // j is its own work needing one alias; f drags g along, so j splits off first and f and g
+ // stay together on top.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("other") as "j")
+ .where($"j")
+ .select($"a", $"e".rlike("magic") as "f", Not($"e".rlike("magic")) as "g", $"j")
+ .where($"f")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: a non-deterministic condition is not split off into a lower layer") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"e".rlike("third") as
+ "j")
+ .where((Rand(10) > 0.5 || $"j") && ($"f" || $"g"))
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // The non-deterministic condition needs one alias and (f || g) needs two, so on cost alone
+ // it would go lowest and see every row. It stays in the top filter instead, and (f || g)
+ // takes the layer it would have had.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f", $"e".rlike("other") as "g")
+ .where($"f" || $"g")
+ .select($"a", $"f", $"g", $"e".rlike("third") as "j")
+ .where(Rand(10) > 0.5 || $"j")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: no split when the expensive conditions are non-deterministic") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"b")
+ .where(Rand(10) > 0.5 || $"f")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // Nothing deterministic to split around, and the non-deterministic condition must not
+ // move, so the plan is unchanged.
+ val correctAnswer = originalQuery
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-55014: do not split the projection when configured") {
+ withSQLConf(SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key -> "false") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g", $"b")
+ .where($"f" && $"g")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ val correctAnswer = originalQuery
+
+ comparePlans(optimized, correctAnswer)
+ }
+ }
+
test("nondeterministic: can always push down filter through project with deterministic field") {
val originalQuery = testRelation
.select($"a")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
index 423dcc83dbed..05c2f37e22ec 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
@@ -2808,6 +2808,102 @@ class DataFrameSuite extends SharedSparkSession
val df = classic.Dataset.ofRows(spark, relation)
checkAnswer(df.select("b"), Row(2))
}
+
+ test("SPARK-55014: only evaluate expensive projections for the rows a filter kept") {
+ val magicCalls = sparkContext.longAccumulator("magicCalls")
+ val otherCalls = sparkContext.longAccumulator("otherCalls")
+ val magic = udf((i: Long) => {
+ magicCalls.add(1)
+ i % 2 == 0
+ })
+ val other = udf((i: Long) => {
+ otherCalls.add(1)
+ i % 3 == 0
+ })
+ // A single partition so that the call counts below are exact.
+ def query(): DataFrame = spark.range(0, 30, 1, 1)
+ .select($"id", magic($"id") as "m", other($"id") as "o")
+ .where($"m" && $"o")
+ val expected = Seq(0L, 6L, 12L, 18L, 24L).map(Row(_, true, true))
+
+ // Codegen already defers a projected expression past a filter that does not need it
+ // (CodegenSupport.evaluateRequiredVariables); splitting buys the same saving on the paths
+ // codegen does not cover -- interpreted projections, operators it bails out of, Python UDFs.
+ // Measure with codegen off.
+ Seq(true, false).foreach { split =>
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
+ SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key -> split.toString) {
+ magicCalls.reset()
+ otherCalls.reset()
+ // Collect, not checkAnswer, so the plan runs exactly once.
+ assert(query().collect().toSeq === expected)
+ assert(magicCalls.value === 30)
+ // Splitting around the filter on `m` means `other` never sees the odd ids.
+ assert(otherCalls.value === (if (split) 15 else 30))
+ }
+ }
+
+ // However the plan is built, the answer is the same.
+ for (codegen <- Seq(true, false); split <- Seq(true, false)) {
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString,
+ SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key -> split.toString) {
+ checkAnswer(query(), expected)
+ }
+ }
+ }
+
+ test("SPARK-55014: no split when projected expressions share the expensive work") {
+ val calls = sparkContext.longAccumulator("udfCalls")
+ val u = udf((i: Long) => {
+ calls.add(1)
+ i
+ })
+ // x and y share one UDF call: subexpression elimination collapses it in one projection but
+ // cannot reach across a filter, so splitting would re-run it on every row below and again on
+ // every survivor. Leave the projection alone; the count stays at one per input row.
+ def query(): DataFrame = spark.range(0, 30, 1, 1)
+ .select($"id", u($"id") as "x", (u($"id") + 1) as "y")
+ .where($"x" % 2 === 0 && $"y" % 3 === 0)
+
+ Seq(true, false).foreach { codegen =>
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString) {
+ calls.reset()
+ assert(query().collect().length === 5)
+ assert(calls.value === 30)
+ }
+ }
+ }
+
+ test("SPARK-55014: splitting follows the filter's order, not the projection's") {
+ val earlyCalls = sparkContext.longAccumulator("earlyCalls")
+ val lateCalls = sparkContext.longAccumulator("lateCalls")
+ val early = udf((i: Long) => {
+ earlyCalls.add(1)
+ i % 3 == 0
+ })
+ val late = udf((i: Long) => {
+ lateCalls.add(1)
+ i % 2 == 0
+ })
+ // `l` is projected before `e` but filtered on after it. Split order must follow the filter:
+ // the condition order is the only selectivity signal we have, and ordering by the projection
+ // would run `late` over every row and could do more work than not splitting at all.
+ def query(): DataFrame = spark.range(0, 30, 1, 1)
+ .select($"id", late($"id") as "l", early($"id") as "e")
+ .where($"e" && $"l")
+ val expected = Seq(0L, 6L, 12L, 18L, 24L).map(Row(_, true, true))
+
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
+ SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key -> "true") {
+ assert(query().collect().toSeq === expected)
+ assert(earlyCalls.value === 30)
+ // Only the ten ids divisible by three reach `late`.
+ assert(lateCalls.value === 10)
+ }
+ }
}
case class GroupByKey(a: Int, b: Int)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org