This is an automated email from the ASF dual-hosted git repository.
zhztheplayer 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 223fd820b6 [VL] Introduce config to allow fail on fallback (#12279)
223fd820b6 is described below
commit 223fd820b6d74962fd9e9d3fef75ba36acc9474a
Author: Felix Loesing <[email protected]>
AuthorDate: Wed Jun 17 01:11:48 2026 -0700
[VL] Introduce config to allow fail on fallback (#12279)
---
.../org/apache/gluten/config/GlutenConfig.scala | 11 ++++
.../spark/sql/execution/GlutenExplainUtils.scala | 61 ++++++++++++++-----
.../sql/execution/GlutenFallbackReporter.scala | 68 ++++++++++++++-------
.../execution/GlutenFailOnFallbackSuite.scala | 70 ++++++++++++++++++++++
4 files changed, 172 insertions(+), 38 deletions(-)
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
index f32f225be6..fad32a0489 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
@@ -325,6 +325,8 @@ class GlutenConfig(conf: SQLConf) extends
GlutenCoreConfig(conf) {
def enableFallbackReport: Boolean = getConf(FALLBACK_REPORTER_ENABLED)
+ def failOnFallback: Boolean = getConf(FALLBACK_FAIL_ON_FALLBACK)
+
def debug: Boolean = getConf(DEBUG_ENABLED)
/** Full scan SQL metrics; also enabled when [[debug]] is true. */
@@ -1430,6 +1432,15 @@ object GlutenConfig extends ConfigRegistry {
.booleanConf
.createWithDefault(true)
+ val FALLBACK_FAIL_ON_FALLBACK =
+ buildConf("spark.gluten.sql.columnar.failOnFallback")
+ .internal()
+ .doc(
+ "When true, throw an exception if any operator falls back to Spark" +
+ " instead of running on the native engine.")
+ .booleanConf
+ .createWithDefault(false)
+
val TEXT_INPUT_ROW_MAX_BLOCK_SIZE =
buildConf("spark.gluten.sql.text.input.max.block.size")
.doc("the max block size for text input rows")
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 dd115efa54..4e15b3e09d 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
@@ -84,18 +84,39 @@ object GlutenExplainUtils extends AdaptiveSparkPlanHelper {
}
}
- private def collectFallbackNodes(plan: QueryPlan[_]): FallbackInfo = {
- var numGlutenNodes = 0
- val fallbackNodeToReason = new mutable.HashMap[String, String]
+ /**
+ * Walks `plan` and classifies each operator into one of three buckets:
+ * - 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
+ *
+ * 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
+ * visited here; callers that need full subquery coverage should use
[[processPlan]], which
+ * performs an explicit subquery walk.
+ */
+ def visitFallbackNodes(
+ plan: QueryPlan[_],
+ onFallback: (SparkPlan, String) => Unit,
+ onGluten: SparkPlan => Unit = _ => ()): Unit = {
- def collect(tmp: QueryPlan[_]): Unit = {
+ def fallbackReason(p: SparkPlan): String =
+ FallbackTags
+ .getOption(p)
+ .orElse(p.logicalLink.flatMap(FallbackTags.getOption))
+ .map(_.reason())
+ .getOrElse("Gluten does not touch it or does not support it")
+
+ def visit(tmp: QueryPlan[_]): Unit = {
tmp.foreachUp {
case _: ExecutedCommandExec =>
- case cmd: CommandResultExec => collect(cmd.commandPhysicalPlan)
+ case cmd: CommandResultExec => visit(cmd.commandPhysicalPlan)
case p: V2CommandExec
if FallbackTags.nonEmpty(p) ||
p.logicalLink.exists(FallbackTags.getOption(_).nonEmpty) =>
- handleVanillaSparkPlan(p, fallbackNodeToReason)
+ onFallback(p, fallbackReason(p))
case _: V2CommandExec =>
case _: DataWritingCommandExec =>
case _: WholeStageCodegenExec =>
@@ -108,26 +129,36 @@ object GlutenExplainUtils extends AdaptiveSparkPlanHelper
{
case _: ReusedExchangeExec =>
case _: NoopLeaf =>
case w: WriteFilesExec if w.child.isInstanceOf[NoopLeaf] =>
- case sub: AdaptiveSparkPlanExec if sub.isSubquery =>
collect(sub.executedPlan)
+ case sub: AdaptiveSparkPlanExec if sub.isSubquery =>
visit(sub.executedPlan)
case _: AdaptiveSparkPlanExec =>
- case p: QueryStageExec => collect(p.plan)
+ case p: QueryStageExec => visit(p.plan)
case p: GlutenPlan =>
- numGlutenNodes += 1
- p.innerChildren.foreach(collect)
+ onGluten(p)
+ p.innerChildren.foreach(visit)
case i: InMemoryTableScanExec =>
if (PlanUtil.isGlutenTableCache(i)) {
- numGlutenNodes += 1
+ onGluten(i)
} else {
- addFallbackNodeWithReason(i, "Columnar table cache is disabled",
fallbackNodeToReason)
+ onFallback(i, "Columnar table cache is disabled")
}
case _: AQEShuffleReadExec => // Ignore
case p: SparkPlan =>
- handleVanillaSparkPlan(p, fallbackNodeToReason)
- p.innerChildren.foreach(collect)
+ onFallback(p, fallbackReason(p))
+ p.innerChildren.foreach(visit)
case _ =>
}
}
- collect(plan)
+ visit(plan)
+ }
+
+ private def collectFallbackNodes(plan: QueryPlan[_]): FallbackInfo = {
+ var numGlutenNodes = 0
+ val fallbackNodeToReason = new mutable.HashMap[String, String]
+ visitFallbackNodes(
+ plan,
+ onGluten = _ => numGlutenNodes += 1,
+ onFallback = (p, reason) => addFallbackNodeWithReason(p, reason,
fallbackNodeToReason)
+ )
(numGlutenNodes, fallbackNodeToReason.toMap)
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenFallbackReporter.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenFallbackReporter.scala
index 0ef0b6a28c..920b8b71b5 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenFallbackReporter.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenFallbackReporter.scala
@@ -18,7 +18,7 @@ package org.apache.spark.sql.execution
import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.events.GlutenPlanFallbackEvent
-import org.apache.gluten.execution.GlutenPlan
+import org.apache.gluten.exception.GlutenException
import org.apache.gluten.extension.columnar.FallbackTags
import org.apache.gluten.logging.LogLevelUtil
@@ -27,20 +27,33 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.StringUtils.PlanStringConcat
import org.apache.spark.sql.execution.ui.GlutenUIUtils
+import scala.collection.mutable.ArrayBuffer
+
/**
- * This rule is used to collect all fallback reason.
- * 1. print fallback reason for each plan node 2. post all fallback reason
using one event
+ * Collects fallback information from a finalized columnar plan and, depending
on configuration:
+ * 1. logs the fallback reason for every non-Gluten operator 2. throws a
[[GlutenException]] when
+ * `spark.gluten.sql.columnar.failOnFallback` is enabled and any fallback
is present 3. posts a
+ * single [[GlutenPlanFallbackEvent]] for the SQL UI
+ *
+ * The set of operators that count as a fallback is shared with
[[GlutenExplainUtils]] via
+ * [[GlutenExplainUtils.visitFallbackNodes]], so the reporter, the SQL UI, and
any future consumers
+ * always agree on what "fallback" means.
*/
case class GlutenFallbackReporter(glutenConf: GlutenConfig, spark:
SparkSession)
extends Rule[SparkPlan]
with LogLevelUtil {
override def apply(plan: SparkPlan): SparkPlan = {
- if (!glutenConf.enableFallbackReport) {
+ val report = glutenConf.enableFallbackReport
+ val fail = glutenConf.failOnFallback
+ if (!report && !fail) {
return plan
}
- printFallbackReason(plan)
- if (GlutenUIUtils.uiEnabled(spark.sparkContext)) {
+ val fallbacks = collectAndLogFallbacks(plan, log = report)
+ if (fail) {
+ throwIfFallback(fallbacks)
+ }
+ if (report && GlutenUIUtils.uiEnabled(spark.sparkContext)) {
postFallbackReason(plan)
}
plan
@@ -53,26 +66,35 @@ case class GlutenFallbackReporter(glutenConf: GlutenConfig,
spark: SparkSession)
logOnLevel(logLevel, s"Validation failed for plan:
$nodeName$executionIdInfo, due to: $reason")
}
- private def printFallbackReason(plan: SparkPlan): Unit = {
+ private def collectAndLogFallbacks(plan: SparkPlan, log: Boolean):
Seq[(String, String)] = {
val validationLogLevel = glutenConf.validationLogLevel
- def printPlan(p: SparkPlan): Unit = {
- p.foreachUp {
- case _: GlutenPlan => // ignore
- case cmd: CommandResultExec =>
- printPlan(cmd.commandPhysicalPlan)
- case p: SparkPlan if FallbackTags.nonEmpty(p) =>
- val tag = FallbackTags.get(p)
- logFallbackReason(validationLogLevel, p.nodeName, tag.reason())
- // With in next round stage in AQE, the physical plan would be a new
instance that
- // can not preserve the tag, so we need to set the fallback reason
to logical plan.
- // Then we can be aware of the fallback reason for the whole plan.
- // If a logical plan mapping to several physical plan, we add all
reason into
- // that logical plan to make sure we do not lose any fallback reason.
- p.logicalLink.foreach(logicalPlan => FallbackTags.add(logicalPlan,
tag))
- case _ =>
+ val fallbacks = ArrayBuffer[(String, String)]()
+ GlutenExplainUtils.visitFallbackNodes(
+ plan,
+ onFallback = (node, reason) => {
+ if (log) {
+ logFallbackReason(validationLogLevel, node.nodeName, reason)
+ // Within the next AQE stage, the physical plan would be a new
instance that does not
+ // preserve the tag, so we propagate the fallback reason to the
logical plan.
+ // If a logical plan maps to several physical plans, we add all
reasons onto that
+ // logical plan to make sure we do not lose any fallback reason.
+ FallbackTags
+ .getOption(node)
+ .foreach(tag => node.logicalLink.foreach(FallbackTags.add(_, tag)))
+ }
+ fallbacks += ((node.nodeName, reason))
}
+ )
+ fallbacks.toSeq
+ }
+
+ private def throwIfFallback(fallbacks: Seq[(String, String)]): Unit = {
+ if (fallbacks.nonEmpty) {
+ val reasons = fallbacks.map { case (name, reason) => s" $name: $reason"
}.mkString("\n")
+ throw new GlutenException(
+ s"Gluten fallback detected
(${GlutenConfig.FALLBACK_FAIL_ON_FALLBACK.key} is enabled). " +
+ s"${fallbacks.size} operator(s) fell back to Spark:\n$reasons")
}
- printPlan(plan)
}
private def postFallbackReason(plan: SparkPlan): Unit = {
diff --git
a/gluten-ut/test/src/test/scala/org/apache/gluten/execution/GlutenFailOnFallbackSuite.scala
b/gluten-ut/test/src/test/scala/org/apache/gluten/execution/GlutenFailOnFallbackSuite.scala
new file mode 100644
index 0000000000..b05b6c952e
--- /dev/null
+++
b/gluten-ut/test/src/test/scala/org/apache/gluten/execution/GlutenFailOnFallbackSuite.scala
@@ -0,0 +1,70 @@
+/*
+ * 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.gluten.execution
+
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.exception.GlutenException
+import org.apache.gluten.utils.BackendTestUtils
+
+import org.apache.spark.SparkConf
+
+import java.util.Locale
+
+class GlutenFailOnFallbackSuite extends WholeStageTransformerSuite {
+
+ protected val resourcePath: String = null
+ protected val fileFormat: String = null
+
+ override protected def sparkConf: SparkConf = {
+ val conf = super.sparkConf
+ if (BackendTestUtils.isCHBackendLoaded()) {
+ conf.set(GlutenConfig.NATIVE_VALIDATION_ENABLED.key, "false")
+ }
+ conf
+ }
+
+ test("failOnFallback throws only when a fallback is present") {
+ withTable("t") {
+ spark.range(10).write.format("parquet").saveAsTable("t")
+
+ // failOnFallback explicitly disabled: a fallback must NOT throw.
+ withSQLConf(
+ GlutenConfig.FALLBACK_FAIL_ON_FALLBACK.key -> "false",
+ GlutenConfig.COLUMNAR_FILESCAN_ENABLED.key -> "false") {
+ sql("SELECT * FROM t").collect()
+ }
+
+ // failOnFallback enabled, fully native query (no fallback): must NOT
throw.
+ withSQLConf(GlutenConfig.FALLBACK_FAIL_ON_FALLBACK.key -> "true") {
+ sql("SELECT * FROM t").collect()
+ }
+
+ // failOnFallback enabled, fallback present: must throw with the config
key
+ // and the offending operator/reason in the message.
+ withSQLConf(
+ GlutenConfig.FALLBACK_FAIL_ON_FALLBACK.key -> "true",
+ GlutenConfig.COLUMNAR_FILESCAN_ENABLED.key -> "false") {
+ val ex = intercept[GlutenException] {
+ sql("SELECT * FROM t").collect()
+ }
+
assert(ex.getMessage.contains(GlutenConfig.FALLBACK_FAIL_ON_FALLBACK.key))
+ assert(ex.getMessage.contains("fell back to Spark"))
+ assert(ex.getMessage.toLowerCase(Locale.ROOT).contains("scan"))
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
