[SPARK-50589][SQL] Avoid extra expression duplication when push filter #49202

zml1206 · 2024-12-16T09:47:38Z

What changes were proposed in this pull request?

Push predicates to Project/Aggregate will extra expression duplication, this will cause the execution plan to become larger, and may also cause performance regression if common expression is non-cheap.

This PR use With expression to rewrite condition which contains attribute that are not cheap and be consumed multiple times. Each predicate generates one or 0 With. Before rewrite With, extra expression duplication is reduced again through the new rule MergeWithExpression. In addition, optimized rewrite With rule, don't pre-evaluate common expressions which are only used once.

Why are the changes needed?

Avoid nested expansion of filters expression causing execution plans to become too large.
Avoid duplicate non-cheap expressions for better performance

Does this PR introduce any user-facing change?

No.

How was this patch tested?

UT.

Was this patch authored or co-authored using generative AI tooling?

No.

zml1206 · 2024-12-17T03:49:10Z

@cloud-fan Can you help first to see if the overall design is feasible? It involves a lot of UT, so I will modify it step by step.

cloud-fan · 2024-12-19T09:39:50Z

I'm very confused. When pushing predicates through Project, the problem we hit is we need to rewrite the attributes in the predicate with the corresponding expression in the Project. If the attribute appears more than once in the predicate, and the expression from Project is expensive, filter pushdown may make the query slower.

The solution is to wrap the predicate with With expression, adding the expression from Project as the common expression, and replace attributes in predicate with the common expression reference. But this PR does some rewriting and I don't understand what it does.

zml1206 · 2024-12-19T09:56:39Z

I'm very confused. When pushing predicates through Project, the problem we hit is we need to rewrite the attributes in the predicate with the corresponding expression in the Project. If the attribute appears more than once in the predicate, and the expression from Project is expensive, filter pushdown may make the query slower.

The solution is to wrap the predicate with With expression, adding the expression from Project as the common expression, and replace attributes in predicate with the common expression reference. But this PR does some rewriting and I don't understand what it does.

Filters support partial pushdown. If an common expression exists in two conditions, one condition is pushed to one level, and the other condition is pushed down to multiple levels. Therefore, after splitting by and, each filter need generates a With, then they need to share commonexprdef. Only the lowest commonexprdef will generate a common expression project. In addition, the child of the original common expression Alias also needs to be rewrite with new common expression attribute.

cloud-fan · 2024-12-19T10:00:31Z

Why do we care about common expressions in the original predicate? This is an optimization opportunity but what we should focus on is to avoid perf regression when pushing filter through project that duplicate expensive expressions.

Maybe we should discuss this with a concrete example.

zml1206 · 2024-12-19T10:14:37Z

Why do we care about common expressions in the original predicate? This is an optimization opportunity but what we should focus on is to avoid perf regression when pushing filter through project that duplicate expensive expressions.

Maybe we should discuss this with a concrete example.

As the ut in pr, I hope that the common expression will only be calculated only once. _common_expr_0 finally used in two Filters and as e.

zml1206 · 2024-12-19T10:18:26Z

Do you want to make it simpler, commonexprdef is not shared, and the original alias is ignored? This way the common expression will be evaluated multiple times, but less than before, especially in nested cases.

cloud-fan · 2024-12-19T10:28:54Z

OK let's make the scope clear. We do not aim to find all duplicated expressions in the query plan tree and optimize them with With. We aim to avoid extra expression duplication that happens during pushing filter through project. So the change should be we update the filter pushdown rule and use With to avoid the aforementioned expression duplication.

zml1206 · 2024-12-19T10:36:19Z

OK let's make the scope clear. We do not aim to find all duplicated expressions in the query plan tree and optimize them with With. We aim to avoid extra expression duplication that happens during pushing filter through project. So the change should be we update the filter pushdown rule and use With to avoid the aforementioned expression duplication.

OK, I'll do it as later, thank you.

cloud-fan · 2024-12-20T06:40:22Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/With.scala

+   * @param replaceMap Replaced attributes and common expressions
+   */
+  def apply(expr: Expression, replaceMap: Map[Attribute, Expression]): With = {
+    val commonExprDefsMap = replaceMap.map(m => m._1 -> CommonExpressionDef(m._2))


why do we create a map if we never look up from it?

Convenient to generate commonExprRefsMap.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/With.scala

cloud-fan · 2024-12-21T01:53:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala

@@ -65,6 +65,8 @@ class SparkOptimizer(
      RewriteDistinctAggregates),
    Batch("Pushdown Filters from PartitionPruning", fixedPoint,
      PushDownPredicates),
+    Batch("Rewrite With expression", fixedPoint,


shall we just rewrite With once at the very end of the optimizer?

Let me try.

We can't push With filter into scan, so rewrite with at the end of operatorOptimizationBatch.

cloud-fan · 2024-12-23T05:39:24Z

how about the rule PushPredicateThroughJoin?

zml1206 · 2024-12-23T05:48:38Z

how about the rule PushPredicateThroughJoin?

There is no replaceAlias logic in PushPredicateThroughJoin.

cloud-fan · 2024-12-23T07:47:18Z

ah ok. Shall we implement the better predicate split-and-combine in this PR to make it more useful? Otherwise we can't handle common cases like a + a > 5 AND a < 10.

zml1206 · 2024-12-23T10:03:03Z

@cloud-fan It seems that the change in filters order causes the execution plan to mismatch. Is there any good solution?

cloud-fan · 2024-12-23T11:51:07Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala

@@ -107,6 +107,14 @@ trait PredicateHelper extends AliasHelper with Logging {
    }
  }

+  /**
+   * First split the predicates by And, then combine the predicates with the same references.


I think there is a misunderstanding. What I proposed is splitConjunctivePredicates should split AND/OR under With: With(c1, c2, c1 + c1 > 0 AND c2 + c2 > 0 AND c1 + c2 > 0) can be split into With(c1, c1 + c1 > 0), With(c2, c2 + c2 > 0), With(c1, c2, c1 + c2 > 0). When combine the predicates later, we also recognize the With and merge them to avoid duplicated expressions.

why split OR

Merge With requires a certain cost. It seems that we only need to merge once before rewriting With.

note: when splitting the predicates under With (by AND), we don't refresh the CTE ids, so when merging the With back, it should be pretty easy.

It is unnecessary to continuously split and merge with. We only need to split by and to generate with, and finally merge with before rewriting with.

cloud-fan · 2024-12-24T08:05:38Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/With.scala

@@ -112,6 +112,22 @@ object With {
    With(replaced(commonExprRefs), commonExprDefs)
  }

+  /**
+   * Helper function to create a [[With]] statement when push down filter.


Since this is very specific to filter pushdown, shall we put this method in the filter pushdown rule?

OK, wait for CI to finish running.

cloud-fan · 2024-12-24T08:07:40Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -1831,8 +1841,8 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
      }

      if (pushDown.nonEmpty) {
-        val pushDownPredicate = pushDown.reduce(And)
-        val replaced = replaceAlias(pushDownPredicate, aliasMap)
+        val replacedByWith = rewriteConditionByWith(pushDown.reduce(And), aliasMap)


rewriteConditionByWith will split the predicate anyway, why do we combine them with And here?

Calculate the number of times each common expression is used through the entire condition.

cloud-fan · 2024-12-24T08:08:47Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

-        val pushDownPredicate = pushDown.reduce(And)
-        val replaced = replaceAlias(pushDownPredicate, aliasMap)
+        val replacedByWith = rewriteConditionByWith(pushDown.reduce(And), aliasMap)
+        val replaced = replaceAlias(replacedByWith, aliasMap)


doesn't rewriteConditionByWith already replace the aliases?

No, rewriteConditionByWith only rewrite common attribute to common expression ref.

cloud-fan · 2024-12-24T08:17:03Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala

@@ -96,14 +96,17 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
        val defs = w.defs.map(rewriteWithExprAndInputPlans(_, inputPlans, isNestedWith = true))
        val refToExpr = mutable.HashMap.empty[CommonExpressionId, Expression]
        val childProjections = Array.fill(inputPlans.length)(mutable.ArrayBuffer.empty[Alias])
+        val refsCount = child.collect { case r: CommonExpressionRef => r}


we should only collect references to the current With, not the nested With.

Suggested change

val refsCount = child.collect { case r: CommonExpressionRef => r}

val refsCount = child.collect { case r: CommonExpressionRef if defs.exists(_.id == r.id) => r}

cloud-fan · 2024-12-24T08:19:52Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/MergeWithExpression.scala

+    val newDefs = left.defs.toBuffer
+    val replaceMap = mutable.HashMap.empty[CommonExpressionId, CommonExpressionRef]
+    right.defs.foreach {rDef =>
+      val index = left.defs.indexWhere(lDef => rDef.child.fastEquals(lDef.child))


plan comparison can be expensive. Since it's a very special case for filter pushdown, shall we use the same CommonExpressionDef ids for different Withs in the split predicates?

fix fix fix fix

zml1206 · 2024-12-25T07:57:50Z

CI failure is not relevant.

cloud-fan · 2024-12-25T08:42:37Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+    if (!SQLConf.get.getConf(SQLConf.ALWAYS_INLINE_COMMON_EXPR)) {
+      val canRewriteConf = cond.filter(canRewriteByWith)
+      if (canRewriteConf.nonEmpty) {
+        val replaceWithMap = canRewriteConf.reduce(And)


we can do canRewriteConf.flatMap(_.collect...)

cloud-fan · 2024-12-25T08:43:21Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+
+  // With does not support inline subquery
+  private def canRewriteByWith(expr: Expression): Boolean = {
+    !expr.containsPattern(PLAN_EXPRESSION)


is this check too strong? We only require the common expression to not contain subqueries.

It seems not feasible, https://github.com/zml1206/spark/actions/runs/12491712984/job/34858074548

Both rewriting subqueries and pushing down predicates are in batch "operator optimization before inferring filters". Pushing down predicates may cause SubqueryExpression to contain common refs, and then rewriting subqueries cannot replace common refs.

cloud-fan · 2024-12-25T08:46:10Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+          .groupBy(identity)
+          .transform((_, v) => v.size)
+          .filter(m => aliasMap.contains(m._1) && m._2 > 1)
+          .map(m => m._1 -> trimAliases(aliasMap.getOrElse(m._1, m._1)))


Slo we've done the work of def replaceAlias here, why do we need to call replaceAlias?

Here just get the map of attribute and common expression to generate With, without rewriting the condition.

cloud-fan · 2024-12-25T08:47:14Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+    val defs = mutable.HashSet.empty[CommonExpressionDef]
+    val replaced = expr.transform {
+      case a: Attribute if refsMap.contains(a) =>
+        defs.add(defsMap.get(a).get)


Suggested change

defs.add(defsMap.get(a).get)

defs.add(defsMap(a))

cloud-fan · 2024-12-25T08:50:09Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala

+        val refsCount = child.collect {
+            case r: CommonExpressionRef
+              if defs.exists {
+                case d: CommonExpressionDef => d.id == r.id


nit: we can create an id set to avoid repeated linear search.

cloud-fan · 2024-12-25T08:51:24Z

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala

-    Batch("Replace CTE with Repartition", Once, ReplaceCTERefWithRepartition)))
+    Batch("Replace CTE with Repartition", Once, ReplaceCTERefWithRepartition),
+    Batch("Merge With expression", fixedPoint, MergeWithExpression),
+    Batch("Rewrite With expression", fixedPoint,


how many places do we need to put this batch?

Before Extract Python UDFs, infer filters and last of each optimizer.

cloud-fan · 2024-12-25T08:58:50Z

sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q21.sf100/explain.txt

 Input [4]: [w_warehouse_name#12, i_item_id#7, inv_before#19, inv_after#20]
-Condition : (CASE WHEN (inv_before#19 > 0) THEN ((cast(inv_after#20 as double) / cast(inv_before#19 as double)) >= 0.666667) END AND CASE WHEN (inv_before#19 > 0) THEN ((cast(inv_after#20 as double) / cast(inv_before#19 as double)) <= 1.5) END)


ah so this is a positive plan change that we avoid computing the CASE WHEN multiple times in the Filter.

zml1206 marked this pull request as draft December 16, 2024 09:47

github-actions bot added the SQL label Dec 16, 2024

Avoid extra expression duplication when push filter

ca81109

zml1206 force-pushed the push_with branch from 536aae0 to ca81109 Compare December 19, 2024 14:52

fix

8b17a9f

zml1206 changed the title ~~[SPARK-50589][SQL] Avoid produce duplicate non-cheap expressions when push down predicate~~ [SPARK-50589][SQL] Avoid extra expression duplication when push filter Dec 19, 2024

zml1206 added 2 commits December 20, 2024 10:51

fix

808752e

fix

53b1852

cloud-fan reviewed Dec 20, 2024

View reviewed changes

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala Show resolved Hide resolved

cloud-fan reviewed Dec 20, 2024

View reviewed changes

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/With.scala Outdated Show resolved Hide resolved

zml1206 added 2 commits December 20, 2024 16:08

fix

ea68366

fix

b212228

zml1206 marked this pull request as ready for review December 20, 2024 08:35

cloud-fan reviewed Dec 21, 2024

View reviewed changes

zml1206 added 3 commits December 21, 2024 12:25

update

b9df556

update

5acc5fa

fix

044ffaa

optimize split condition

f1aa618

cloud-fan reviewed Dec 23, 2024

View reviewed changes

zml1206 added 3 commits December 24, 2024 13:15

optimize split condition

6247c34

update

8fb567e

fix

58459d0

cloud-fan reviewed Dec 24, 2024

View reviewed changes

fix

776122d

fix fix fix fix

zml1206 force-pushed the push_with branch from 16a8b15 to 776122d Compare December 25, 2024 03:00

fix

a94d51b

cloud-fan reviewed Dec 25, 2024

View reviewed changes

zml1206 added 2 commits December 25, 2024 17:28

update

925f80e

fix

9683b68

zml1206 force-pushed the push_with branch from 541b9ed to 9683b68 Compare December 25, 2024 13:25

	val refsCount = child.collect { case r: CommonExpressionRef => r}
	val refsCount = child.collect { case r: CommonExpressionRef if defs.exists(_.id == r.id) => r}

		Input [4]: [w_warehouse_name#12, i_item_id#7, inv_before#19, inv_after#20]
		Condition : (CASE WHEN (inv_before#19 > 0) THEN ((cast(inv_after#20 as double) / cast(inv_before#19 as double)) >= 0.666667) END AND CASE WHEN (inv_before#19 > 0) THEN ((cast(inv_after#20 as double) / cast(inv_before#19 as double)) <= 1.5) END)

[SPARK-50589][SQL] Avoid extra expression duplication when push filter #49202

Are you sure you want to change the base?

[SPARK-50589][SQL] Avoid extra expression duplication when push filter #49202

Conversation

zml1206 commented Dec 16, 2024 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

zml1206 commented Dec 17, 2024

cloud-fan commented Dec 19, 2024

zml1206 commented Dec 19, 2024 • edited Loading

cloud-fan commented Dec 19, 2024

zml1206 commented Dec 19, 2024

zml1206 commented Dec 19, 2024 • edited Loading

cloud-fan commented Dec 19, 2024

zml1206 commented Dec 19, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Dec 23, 2024 • edited Loading

zml1206 commented Dec 23, 2024

cloud-fan commented Dec 23, 2024

zml1206 commented Dec 23, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zml1206 commented Dec 25, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zml1206 commented Dec 16, 2024 •

edited

Loading

zml1206 commented Dec 19, 2024 •

edited

Loading

zml1206 commented Dec 19, 2024 •

edited

Loading

cloud-fan commented Dec 23, 2024 •

edited

Loading