This is an automated email from the ASF dual-hosted git repository.
philo-he pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 57e7bac77d [CORE] Eliminate local sort through partial project in
EliminateLocalSort (#13208)
57e7bac77d is described below
commit 57e7bac77d9eab5a3af0662811f2e80e260674c4
Author: Felix Loesing <[email protected]>
AuthorDate: Wed Oct 7 18:29:35 2026 -0700
[CORE] Eliminate local sort through partial project in EliminateLocalSort
(#13208)
---
.../execution/ColumnarPartialProjectExec.scala | 6 ++---
.../execution/ColumnarPartialProjectExec.scala | 6 ++---
.../gluten/expression/UDFPartialProjectSuite.scala | 14 ++++++++++-
.../BasicPhysicalOperatorTransformer.scala | 8 +++---
.../extension/columnar/EliminateLocalSort.scala | 29 ++++++++++++----------
...PartitioningAndOrderingPreservingNodeShim.scala | 20 ---------------
...PartitioningAndOrderingPreservingNodeShim.scala | 20 ---------------
...PartitioningAndOrderingPreservingNodeShim.scala | 20 ---------------
...PartitioningAndOrderingPreservingNodeShim.scala | 20 ---------------
...PartitioningAndOrderingPreservingNodeShim.scala | 20 ---------------
10 files changed, 39 insertions(+), 124 deletions(-)
diff --git
a/backends-bolt/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
b/backends-bolt/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
index 5cbed106d8..27b26e6d5b 100644
---
a/backends-bolt/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
+++
b/backends-bolt/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
@@ -29,7 +29,7 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.QueryPlan
-import org.apache.spark.sql.execution.{ExplainUtils, OrderPreservingNodeShim,
PartitioningPreservingNodeShim, ProjectExec, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.execution.{ExplainUtils,
OrderPreservingUnaryExecNode, PartitioningPreservingUnaryExecNode, ProjectExec,
SparkPlan, UnaryExecNode}
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
import org.apache.spark.sql.hive.{BoltHiveUDFTransformer, HiveUDFTransformer}
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
@@ -52,8 +52,8 @@ import scala.collection.mutable.ListBuffer
case class ColumnarPartialProjectExec(projectList: Seq[Expression], child:
SparkPlan)(
replacedAlias: Seq[Alias])
extends UnaryExecNode
- with OrderPreservingNodeShim
- with PartitioningPreservingNodeShim
+ with OrderPreservingUnaryExecNode
+ with PartitioningPreservingUnaryExecNode
with ValidatablePlan {
private val projectAttributes: ListBuffer[Attribute] = ListBuffer()
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
b/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
index b8f7d34663..96bf21f418 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
@@ -31,7 +31,7 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.QueryPlan
-import org.apache.spark.sql.execution.{ExplainUtils, OrderPreservingNodeShim,
PartitioningPreservingNodeShim, ProjectExec, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.execution.{ExplainUtils,
OrderPreservingUnaryExecNode, PartitioningPreservingUnaryExecNode, ProjectExec,
SparkPlan, UnaryExecNode}
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
import org.apache.spark.sql.hive.{HiveUDFTransformer, VeloxHiveUDFTransformer}
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
@@ -55,8 +55,8 @@ import scala.collection.mutable.ListBuffer
case class ColumnarPartialProjectExec(projectList: Seq[Expression], child:
SparkPlan)(
replacedAlias: Seq[Alias])
extends UnaryExecNode
- with OrderPreservingNodeShim
- with PartitioningPreservingNodeShim
+ with OrderPreservingUnaryExecNode
+ with PartitioningPreservingUnaryExecNode
with ValidatablePlan {
private val projectAttributes: ListBuffer[Attribute] = ListBuffer()
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala
index 71b60aeefa..a8287f7e0a 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala
@@ -17,7 +17,7 @@
package org.apache.gluten.expression
import org.apache.gluten.config.GlutenConfig
-import org.apache.gluten.execution.{ColumnarPartialProjectExec,
WholeStageTransformerSuite}
+import org.apache.gluten.execution.{ColumnarPartialProjectExec,
SortExecTransformer, WholeStageTransformerSuite}
import org.apache.spark.SparkConf
import org.apache.spark.sql.catalyst.optimizer.{ConstantFolding,
NullPropagation}
@@ -119,6 +119,18 @@ class UDFPartialProjectSuite extends
WholeStageTransformerSuite {
}.size == 2)
}
+ test("eliminate local sort below partial project to keep collect_list
order") {
+ // A non-stable native sort on the grouping key would reorder collected
values.
+ runQueryAndCompare(
+ """SELECT k, collect_list(concat_concat(v)) FROM (
+ | SELECT id % 10 AS k, cast(id AS string) AS v FROM range(0, 10000,
1, 1)
+ |) GROUP BY k""".stripMargin) {
+ df =>
+ checkGlutenPlan[ColumnarPartialProjectExec](df)
+ assert(getExecutedPlan(df).collect { case s: SortExecTransformer => s
}.isEmpty)
+ }
+ }
+
test("test plus_one with many columns in project") {
runQueryAndCompare("SELECT plus_one(cast(l_orderkey as long)),
hash(l_partkey) from lineitem") {
checkGlutenPlan[ColumnarPartialProjectExec]
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
index 1514aca532..19864d2960 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
@@ -39,8 +39,8 @@ import scala.collection.JavaConverters._
abstract class FilterExecTransformerBase(val cond: Expression, val input:
SparkPlan)
extends UnaryTransformSupport
- with OrderPreservingNodeShim
- with PartitioningPreservingNodeShim
+ with OrderPreservingUnaryExecNode
+ with PartitioningPreservingUnaryExecNode
with PredicateHelper
with Logging {
@@ -168,8 +168,8 @@ abstract class FilterExecTransformerBase(val cond:
Expression, val input: SparkP
abstract class ProjectExecTransformerBase(val list: Seq[NamedExpression], val
input: SparkPlan)
extends UnaryTransformSupport
- with OrderPreservingNodeShim
- with PartitioningPreservingNodeShim
+ with OrderPreservingUnaryExecNode
+ with PartitioningPreservingUnaryExecNode
with PredicateHelper
with Logging {
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EliminateLocalSort.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EliminateLocalSort.scala
index 17e7f29eec..fdd0cdbffe 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EliminateLocalSort.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EliminateLocalSort.scala
@@ -20,7 +20,7 @@ import org.apache.gluten.execution.{ProjectExecTransformer,
ShuffledHashJoinExec
import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.{ProjectExec, SortExec, SparkPlan,
UnaryExecNode}
+import org.apache.spark.sql.execution.{OrderPreservingUnaryExecNode,
ProjectExec, SortExec, SparkPlan, UnaryExecNode}
/**
* This rule is used to eliminate unnecessary local sort.
@@ -44,6 +44,8 @@ object EliminateLocalSort extends Rule[SparkPlan] {
private def canThrough(p: SparkPlan): Boolean = p match {
case _: ProjectExec => true
case _: ProjectExecTransformer => true
+ // e.g. ColumnarPartialProjectExec, which sits between a pulled-out
project and its sort.
+ case p: OrderPreservingUnaryExecNode =>
p.requiredChildOrdering.forall(_.isEmpty)
case _ => false
}
@@ -56,22 +58,23 @@ object EliminateLocalSort extends Rule[SparkPlan] {
case p if canEliminateLocalSort(p) =>
val requiredChildOrdering = p.requiredChildOrdering
assert(requiredChildOrdering.size == p.children.size)
- val newChildren = p.children.zipWithIndex.map {
- case (SortWithChild(gChild), i) if orderingSatisfies(gChild,
requiredChildOrdering(i)) =>
- gChild
- case (p: UnaryExecNode, i) if canThrough(p) =>
- // There may be more than one project between target operator and
sort,
- // e.g., both hash aggregate and sort pull out project
- p.child match {
- case SortWithChild(gChild) if orderingSatisfies(gChild,
requiredChildOrdering(i)) =>
- p.withNewChildren(gChild :: Nil)
- case _ => p
- }
- case p => p._1
+ val newChildren = p.children.zip(requiredChildOrdering).map {
+ case (child, ordering) => eliminateSort(child, ordering)
}
p.withNewChildren(newChildren)
}
}
+
+ // There may be more than one project between target operator and sort,
e.g., both hash
+ // aggregate and sort pull out project, or a partial project is inserted.
+ private def eliminateSort(plan: SparkPlan, requiredOrdering:
Seq[SortOrder]): SparkPlan =
+ plan match {
+ case SortWithChild(gChild) if orderingSatisfies(gChild,
requiredOrdering) =>
+ gChild
+ case p: UnaryExecNode if canThrough(p) =>
+ p.withNewChildren(eliminateSort(p.child, requiredOrdering) :: Nil)
+ case p => p
+ }
}
object SortWithChild {
diff --git
a/shims/spark34/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
b/shims/spark34/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
deleted file mode 100644
index f0ffcc8b69..0000000000
---
a/shims/spark34/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution
-
-trait OrderPreservingNodeShim extends OrderPreservingUnaryExecNode
-trait PartitioningPreservingNodeShim extends
PartitioningPreservingUnaryExecNode
diff --git
a/shims/spark35/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
b/shims/spark35/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
deleted file mode 100644
index f0ffcc8b69..0000000000
---
a/shims/spark35/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution
-
-trait OrderPreservingNodeShim extends OrderPreservingUnaryExecNode
-trait PartitioningPreservingNodeShim extends
PartitioningPreservingUnaryExecNode
diff --git
a/shims/spark40/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
b/shims/spark40/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
deleted file mode 100644
index f0ffcc8b69..0000000000
---
a/shims/spark40/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution
-
-trait OrderPreservingNodeShim extends OrderPreservingUnaryExecNode
-trait PartitioningPreservingNodeShim extends
PartitioningPreservingUnaryExecNode
diff --git
a/shims/spark41/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
b/shims/spark41/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
deleted file mode 100644
index f0ffcc8b69..0000000000
---
a/shims/spark41/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution
-
-trait OrderPreservingNodeShim extends OrderPreservingUnaryExecNode
-trait PartitioningPreservingNodeShim extends
PartitioningPreservingUnaryExecNode
diff --git
a/shims/spark42/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
b/shims/spark42/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
deleted file mode 100644
index f0ffcc8b69..0000000000
---
a/shims/spark42/src/main/scala/org/apache/spark/sql/execution/PartitioningAndOrderingPreservingNodeShim.scala
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution
-
-trait OrderPreservingNodeShim extends OrderPreservingUnaryExecNode
-trait PartitioningPreservingNodeShim extends
PartitioningPreservingUnaryExecNode
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
