You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by yu...@apache.org on 2024/10/21 13:27:22 UTC
(incubator-gluten) branch main updated: [GLUTEN-7577][VL] Add pattern match for extension rules (#7584)
This is an automated email from the ASF dual-hosted git repository.
yuanzhou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 29b52f7e63 [GLUTEN-7577][VL] Add pattern match for extension rules (#7584)
29b52f7e63 is described below
commit 29b52f7e63fbc893ef5ceb1862abd18e3ff09331
Author: Mingliang Zhu <zh...@gmail.com>
AuthorDate: Mon Oct 21 21:27:18 2024 +0800
[GLUTEN-7577][VL] Add pattern match for extension rules (#7584)
this patch refined the patten matching rules
Fixes: #7577
---
.../gluten/extension/BloomFilterMightContainJointRewriteRule.scala | 2 +-
.../main/scala/org/apache/gluten/extension/CollectRewriteRule.scala | 4 ++--
.../org/apache/gluten/extension/FlushableHashAggregateRule.scala | 3 ++-
3 files changed, 5 insertions(+), 4 deletions(-)
diff --git a/backends-velox/src/main/scala/org/apache/gluten/extension/BloomFilterMightContainJointRewriteRule.scala b/backends-velox/src/main/scala/org/apache/gluten/extension/BloomFilterMightContainJointRewriteRule.scala
index 9a0a59e8e3..735d6ad41b 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/extension/BloomFilterMightContainJointRewriteRule.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/extension/BloomFilterMightContainJointRewriteRule.scala
@@ -28,7 +28,7 @@ import org.apache.spark.sql.execution.SparkPlan
case class BloomFilterMightContainJointRewriteRule(spark: SparkSession) extends Rule[SparkPlan] {
override def apply(plan: SparkPlan): SparkPlan = PhysicalPlanSelector.maybe(spark, plan) {
- if (!(GlutenConfig.getConf.enableNativeBloomFilter)) {
+ if (!GlutenConfig.getConf.enableNativeBloomFilter) {
return plan
}
val out = plan.transformWithSubqueries {
diff --git a/backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala b/backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala
index 86defb2fcb..85d31b8d02 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala
@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.{Expression, WindowExpression}
import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE_EXPRESSION, WINDOW_EXPRESSION}
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, AGGREGATE_EXPRESSION, WINDOW, WINDOW_EXPRESSION}
import scala.reflect.{classTag, ClassTag}
@@ -41,7 +41,7 @@ case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
return plan
}
- val newPlan = plan.transformUp {
+ val newPlan = plan.transformUpWithPruning(_.containsAnyPattern(WINDOW, AGGREGATE)) {
case node =>
replaceAggCollect(node)
}
diff --git a/backends-velox/src/main/scala/org/apache/gluten/extension/FlushableHashAggregateRule.scala b/backends-velox/src/main/scala/org/apache/gluten/extension/FlushableHashAggregateRule.scala
index 04bdbe1efb..2e53906977 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/extension/FlushableHashAggregateRule.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/extension/FlushableHashAggregateRule.scala
@@ -23,6 +23,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.aggregate.{Partial, PartialMerge}
import org.apache.spark.sql.catalyst.plans.physical.ClusteredDistribution
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.EXCHANGE
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
@@ -36,7 +37,7 @@ case class FlushableHashAggregateRule(session: SparkSession) extends Rule[SparkP
if (!GlutenConfig.getConf.enableVeloxFlushablePartialAggregation) {
return plan
}
- plan.transformUp {
+ plan.transformUpWithPruning(_.containsPattern(EXCHANGE)) {
case s: ShuffleExchangeLike =>
// If an exchange follows a hash aggregate in which all functions are in partial mode,
// then it's safe to convert the hash aggregate to flushable hash aggregate.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org