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 26d35367fb [GLUTEN-12702][VL] Introduce shouldIgnoreInFallbackStats to
skip certain plan nodes in fallback counting (#12703)
26d35367fb is described below
commit 26d35367fb61e84d34dbce555e33995b445483bd
Author: Wechar Yu <[email protected]>
AuthorDate: Fri Aug 28 08:47:38 2026 +0800
[GLUTEN-12702][VL] Introduce shouldIgnoreInFallbackStats to skip certain
plan nodes in fallback counting (#12703)
---
.../AutoAdjustStageResourceProfileSuite.scala | 35 +++++++++++++
.../GlutenAutoAdjustStageResourceProfile.scala | 11 ++++-
.../spark/sql/execution/GlutenExplainUtils.scala | 57 +++++++++++++---------
.../spark/sql/execution/GlutenImplicits.scala | 34 +++----------
4 files changed, 85 insertions(+), 52 deletions(-)
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/AutoAdjustStageResourceProfileSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/AutoAdjustStageResourceProfileSuite.scala
index 8e2eca4f78..2ef614e415 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/execution/AutoAdjustStageResourceProfileSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/AutoAdjustStageResourceProfileSuite.scala
@@ -23,6 +23,7 @@ import org.apache.spark.annotation.Experimental
import org.apache.spark.sql.execution.{ApplyResourceProfileExec,
ColumnarShuffleExchangeExec, SparkPlan}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.internal.SQLConf
@Experimental
class AutoAdjustStageResourceProfileSuite
@@ -131,6 +132,40 @@ class AutoAdjustStageResourceProfileSuite
}
}
+ test("Structural Spark nodes should not be counted as fallback nodes") {
+ withSQLConf(
+ GlutenConfig.AUTO_ADJUST_STAGE_RESOURCES_FALLEN_NODE_RATIO_THRESHOLD.key
-> "0.5",
+ SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true"
+ ) {
+ runQueryAndCompare(
+ """
+ |SELECT /*+ BROADCAST(t2) */ t1.c1, t2.c1
+ |FROM tmp1 t1
+ |JOIN (
+ | SELECT /*+ REBALANCE(c1) */ c1
+ | FROM tmp2
+ |) t2
+ |ON t1.c1 = t2.c1
+ |""".stripMargin) {
+ // scalastyle:off
+ // format: off
+ /*
+ ColumnarBroadcastExchange
HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false),
[plan_id=1267]
+ +- AQEShuffleRead coalesced
+ +- ShuffleQueryStage 0
+ +- ColumnarExchange hashpartitioning(c1#10, 5),
REBALANCE_PARTITIONS_BY_COL, [c1#10], [plan_id=1194],
[shuffle_writer_type=hash], [output=[c1#10: int]]
+ +- VeloxResizeBatches
+ +- ^(1) ProjectExecTransformer [hash(c1#10, 42) AS
hash_partition_key#31, c1#10]
+ +- ^(1) FilterExecTransformer isnotnull(c1#10)
+ +- ^(1) FileFileSourceScanExecTransformer parquet
spark_catalog.default.tmp2[c1#10] Batched: true, DataFilters: [isnotnull(c1#10)]
+ */
+ // format: on
+ // scalastyle:on
+ df =>
assert(collectApplyResourceProfileExec(df.queryExecution.executedPlan) == 0)
+ }
+ }
+ }
+
test("Apply new resource profile when whole stage fallback") {
withSQLConf(
GlutenConfig.COLUMNAR_FALLBACK_PREFER_COLUMNAR.key -> "false",
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
index 8682fa6889..14669c51ac 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution
import org.apache.gluten.config.{GlutenConfig, GlutenCoreConfig}
import org.apache.gluten.execution.{ColumnarToRowExecBase, CudfTag,
GlutenPlan, WholeStageTransformer}
import org.apache.gluten.logging.LogLevelUtil
+import org.apache.gluten.utils.PlanUtil
import org.apache.spark.SparkConf
import org.apache.spark.annotation.Experimental
@@ -29,6 +30,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{GlutenAutoAdjustStageResourceProfile =>
GlutenResourceProfile}
import org.apache.spark.sql.execution.adaptive.QueryStageExec
+import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
import org.apache.spark.sql.execution.command.{DataWritingCommandExec,
ExecutedCommandExec}
import org.apache.spark.sql.execution.exchange.Exchange
import org.apache.spark.sql.internal.SQLConf
@@ -145,8 +147,13 @@ case class
GlutenAutoAdjustStageResourceProfile(glutenConf: GlutenConfig, spark:
// case 2: check whether fallback exists and decide whether increase heap
memory
// and decrease offheap memory.
- val fallenNodeCnt = planNodes.count(p => !p.isInstanceOf[GlutenPlan])
- val totalCount = planNodes.size
+ val countedPlanNodes =
planNodes.filterNot(GlutenExplainUtils.shouldIgnoreInFallbackStats)
+ val fallenNodeCnt = countedPlanNodes.count {
+ case _: GlutenPlan => false
+ case i: InMemoryTableScanExec => !PlanUtil.isGlutenTableCache(i)
+ case _ => true
+ }
+ val totalCount = countedPlanNodes.size
if (1.0 * fallenNodeCnt / totalCount >=
glutenConf.autoAdjustStageFallenNodeThreshold) {
val newMemoryAmount = memoryRequest.get.amount *
glutenConf.autoAdjustStageRPHeapRatio
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenExplainUtils.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenExplainUtils.scala
index 21694eea07..1151f4584a 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenExplainUtils.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenExplainUtils.scala
@@ -45,6 +45,34 @@ import scala.collection.mutable.{ArrayBuffer, BitSet}
object GlutenExplainUtils extends AdaptiveSparkPlanHelper {
type FallbackInfo = (Int, Map[String, String])
+ /**
+ * Returns whether a plan should be ignored when collecting fallback
statistics.
+ *
+ * Such plans may be execution-framework or implementation wrappers.
+ */
+ def shouldIgnoreInFallbackStats(plan: SparkPlan): Boolean = plan match {
+ case _: ExecutedCommandExec => true
+ case _: CommandResultExec => true
+ case p: V2CommandExec =>
+ !FallbackTags.nonEmpty(p) &&
p.logicalLink.forall(FallbackTags.getOption(_).isEmpty)
+ case _: DataWritingCommandExec => true
+ case _: WholeStageCodegenExec => true
+ case _: WholeStageTransformer => true
+ case _: InputAdapter => true
+ case _: ColumnarInputAdapter => true
+ case _: InputIteratorTransformer => true
+ case _: ColumnarToRowTransition => true
+ case _: RowToColumnarTransition => true
+ case _: ReusedExchangeExec => true
+ case _: NoopLeaf => true
+ case w: WriteFilesExec => w.child.isInstanceOf[NoopLeaf]
+ case _: AdaptiveSparkPlanExec => true
+ case _: QueryStageExec => true
+ case _: AQEShuffleReadExec => true
+ case _: ColumnarAQEShuffleReadExec => true
+ case _ => false
+ }
+
def addFallbackNodeWithReason(
p: SparkPlan,
reason: String,
@@ -89,8 +117,7 @@ object GlutenExplainUtils extends AdaptiveSparkPlanHelper {
* - native Gluten operator: invokes `onGluten`
* - vanilla Spark operator considered a fallback: invokes `onFallback`
with the reason resolved
* from physical or logical [[FallbackTags]], or a synthetic default
when no tag is present
- * - structural / non-operator nodes (commands, transitions, codegen
wrappers, query stages, AQE
- * shuffle reads, reused exchanges, etc.): no callback
+ * - nodes ignored by [[shouldIgnoreInFallbackStats]]: no callback
*
* Recurses into AQE subqueries and query stages, and into each visited
operator's
* `innerChildren`. Subqueries reached purely via expressions on a vanilla
Spark plan are not
@@ -111,38 +138,22 @@ object GlutenExplainUtils extends AdaptiveSparkPlanHelper
{
def visit(tmp: QueryPlan[_]): Unit = {
tmp.foreachUp {
- case _: ExecutedCommandExec =>
case cmd: CommandResultExec => visit(cmd.commandPhysicalPlan)
case p: V2CommandExec
- if FallbackTags.nonEmpty(p) ||
- p.logicalLink.exists(FallbackTags.getOption(_).nonEmpty) =>
+ if !shouldIgnoreInFallbackStats(p) =>
onFallback(p, fallbackReason(p))
- case _: V2CommandExec =>
- case _: DataWritingCommandExec =>
- case _: WholeStageCodegenExec =>
- case _: WholeStageTransformer =>
- case _: InputAdapter =>
- case _: ColumnarInputAdapter =>
- case _: InputIteratorTransformer =>
- case _: ColumnarToRowTransition =>
- case _: RowToColumnarTransition =>
- case _: ReusedExchangeExec =>
- case _: NoopLeaf =>
- case w: WriteFilesExec if w.child.isInstanceOf[NoopLeaf] =>
case sub: AdaptiveSparkPlanExec if sub.isSubquery =>
visit(sub.executedPlan)
- case _: AdaptiveSparkPlanExec =>
case p: QueryStageExec => visit(p.plan)
- case p: GlutenPlan =>
- onGluten(p)
- p.innerChildren.foreach(visit)
case i: InMemoryTableScanExec =>
if (PlanUtil.isGlutenTableCache(i)) {
onGluten(i)
} else {
onFallback(i, "Columnar table cache is disabled")
}
- case _: AQEShuffleReadExec => // Ignore
- case _: ColumnarAQEShuffleReadExec => // Ignore
+ case p: SparkPlan if shouldIgnoreInFallbackStats(p) => // Ignore
+ case p: GlutenPlan =>
+ onGluten(p)
+ p.innerChildren.foreach(visit)
case p: SparkPlan =>
onFallback(p, fallbackReason(p))
p.innerChildren.foreach(visit)
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenImplicits.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenImplicits.scala
index 6e54ec5dfd..a4354220b7 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenImplicits.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenImplicits.scala
@@ -17,8 +17,7 @@
package org.apache.spark.sql.execution
import org.apache.gluten.exception.GlutenException
-import org.apache.gluten.execution.{GlutenPlan, WholeStageTransformer}
-import org.apache.gluten.extension.columnar.FallbackTags
+import org.apache.gluten.execution.GlutenPlan
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.utils.PlanUtil
@@ -27,13 +26,9 @@ import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.catalyst.plans.logical.{CommandResult, LogicalPlan}
import org.apache.spark.sql.catalyst.util.StringUtils.PlanStringConcat
import org.apache.spark.sql.classic.ClassicConversions._
-import org.apache.spark.sql.execution.ColumnarWriteFilesExec.NoopLeaf
-import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec,
AQEShuffleReadExec, ColumnarAQEShuffleReadExec, QueryStageExec}
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec,
QueryStageExec}
import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
-import org.apache.spark.sql.execution.command.{DataWritingCommandExec,
ExecutedCommandExec}
-import org.apache.spark.sql.execution.datasources.WriteFilesExec
import org.apache.spark.sql.execution.datasources.v2.V2CommandExec
-import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
import org.apache.spark.sql.internal.SQLConf
import scala.collection.mutable
@@ -108,24 +103,10 @@ object GlutenImplicits {
def collect(tmp: QueryPlan[_]): Unit = {
tmp.foreachUp {
- case _: ExecutedCommandExec =>
case cmd: CommandResultExec => collect(cmd.commandPhysicalPlan)
case p: V2CommandExec
- if FallbackTags.nonEmpty(p) ||
- p.logicalLink.exists(FallbackTags.getOption(_).nonEmpty) =>
+ if !GlutenExplainUtils.shouldIgnoreInFallbackStats(p) =>
GlutenExplainUtils.handleVanillaSparkPlan(p, fallbackNodeToReason)
- case _: V2CommandExec =>
- case _: DataWritingCommandExec =>
- case _: WholeStageCodegenExec =>
- case _: WholeStageTransformer =>
- case _: InputAdapter =>
- case _: ColumnarInputAdapter =>
- case _: InputIteratorTransformer =>
- case _: ColumnarToRowTransition =>
- case _: RowToColumnarTransition =>
- case p: ReusedExchangeExec =>
- case _: NoopLeaf =>
- case w: WriteFilesExec if w.child.isInstanceOf[NoopLeaf] =>
case p: AdaptiveSparkPlanExec if isFinalAdaptivePlan(p) =>
collect(p.executedPlan)
case p: AdaptiveSparkPlanExec =>
@@ -149,9 +130,6 @@ object GlutenImplicits {
numGlutenNodes += innerNumGlutenNodes
fallbackNodeToReason.++=(innerFallbackNodeToReason)
case p: QueryStageExec => collect(p.plan)
- case p: GlutenPlan =>
- numGlutenNodes += 1
- p.innerChildren.foreach(collect)
case i: InMemoryTableScanExec =>
if (PlanUtil.isGlutenTableCache(i)) {
numGlutenNodes += 1
@@ -162,8 +140,10 @@ object GlutenImplicits {
fallbackNodeToReason)
}
collect(i.relation.cachedPlan)
- case _: AQEShuffleReadExec => // Ignore
- case _: ColumnarAQEShuffleReadExec => // Ignore
+ case p: SparkPlan if GlutenExplainUtils.shouldIgnoreInFallbackStats(p)
=> // Ignore
+ case p: GlutenPlan =>
+ numGlutenNodes += 1
+ p.innerChildren.foreach(collect)
case p: SparkPlan =>
GlutenExplainUtils.handleVanillaSparkPlan(p, fallbackNodeToReason)
p.innerChildren.foreach(collect)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
