This is an automated email from the ASF dual-hosted git repository.
weiting-chen 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 2830381db7 [CORE][VL] Add columnar EmptyRelationExec offload to Velox
backend (#12766)
2830381db7 is described below
commit 2830381db71b137ca58cf87178b604e0b9cdfabd
Author: Minni Mittal <[email protected]>
AuthorDate: Thu Sep 10 20:43:00 2026 +0530
[CORE][VL] Add columnar EmptyRelationExec offload to Velox backend (#12766)
* [CORE][VL] Add native EmptyRelationExec offload to Velox backend
Offload EmptyRelationExec (a leaf node AQE's Propagate Empty Relations
optimization creates on Spark 4.0+ when it proves a subtree produces no
output) to a native EmptyRelationExecTransformer. The transformer produces
an empty RDD[ColumnarBatch] so surrounding columnar operators no longer
need to be wrapped in ColumnarToRow / RowToColumnar transitions around the
empty relation.
EmptyRelationExec only exists on Spark 4.0+ (SPARK-47217), so the node is
never referenced from version-agnostic modules: detection goes through the
new SparkShims.isEmptyRelationExec, overridden only in the Spark 4.0 and 4.1
shims and defaulting to false elsewhere. The shared OffloadOthers rule and
the SparkPlanExecApi trait therefore compile unchanged against Spark
3.3-3.5.
The offload is gated by spark.gluten.sql.columnar.emptyRelation (default
true). The Velox backend implements isSupportEmptyRelationExec; other
backends inherit the trait default and keep vanilla execution.
Adds VeloxEmptyRelationSuite (empty-result correctness on all supported
Spark versions, plus plan-shape and config-gate assertions gated to Spark
4.0+) and re-enables the upstream SPARK-35585 AQE test on Spark 4.0/4.1 with
a Gluten-aware assertion that accepts either EmptyRelationExec or the
transformer.
Co-authored-by: Copilot <[email protected]>
* Reword native->columnar for EmptyRelationExec offload and sync config docs
The EmptyRelationExecTransformer is a JVM-side columnar leaf that returns
an empty RDD[ColumnarBatch]; it does not invoke native execution. Reword the
config doc string and scaladocs accordingly, and regenerate the
Configuration.md row so it matches the config doc() string (fixes
AllGlutenConfiguration check).
* Fix test compile error in VeloxEmptyRelationSuite
SharedSparkSession mixes in Spark's SQLTestUtilsBase, which overrides
withSQLConf to return Unit rather than the block value. Capturing the collected
rows via the block return type therefore inferred Unit and failed to compile
(isEmpty / checkAnswer on Unit). Assign the vanilla result to a var inside the
block instead.
* Fix VeloxEmptyRelationSuite plan-shape tests to actually create
EmptyRelationExec
The plan-shape tests queried 'WHERE 1 = 0', which the logical optimizer
folds to an empty LocalRelation (physical LocalTableScan) before physical
planning, so no EmptyRelationExec is ever produced and the offload assertion
failed (0 transformers; plan was LocalTableScan <empty>). EmptyRelationExec is
an AQE-runtime node created by Propagate Empty Relations when a materialized
query stage is empty at runtime. Drive it through an AQE INTERSECT whose left
side (l_orderkey < 0) is empty o [...]
* Make EmptyRelationExecTransformer extend LeafExecNode
Extend LeafExecNode with ValidatablePlan (matching ColumnarRangeBaseExec)
instead of manually overriding children and withNewChildrenInternal.
LeafExecNode/LeafLike supplies an empty children list and a
withNewChildrenInternal that enforces the no-children invariant, rather than
silently ignoring newChildren and returning this.
* [VL] Make EmptyRelationExecTransformer a dual-mode, schema-independent
leaf
Model EmptyRelationExecTransformer as a dual-mode GlutenPlan instead of a
batch-only ValidatablePlan. The node carries no data and never sends rows to
the backend, so native schema validation is irrelevant and previously forced
needless fallback whenever the (unused) output schema contained a
backend-unsupported type.
It now advertises both VanillaRowType and the primary backend batch type and
implements doExecute (emptyRDD[InternalRow]) alongside doExecuteColumnar
(emptyRDD[ColumnarBatch]). This lets the transition framework consume the
empty
relation directly from either a row or columnar context, so no
ColumnarToRow /
RowToColumnar sandwich is inserted around it.
Also add a strict-adjacency assertion in VeloxEmptyRelationSuite that
verifies
no RowToColumnar transition wraps the transformer (unwrapping AQE
query-stage
wrappers), guarding the offload's core performance invariant against future
transition-planning regressions.
Co-authored-by: Copilot <[email protected]>
* [VL] Fix EmptyRelation suite CI checks
Replace the non-ASCII dash rejected by Scalastyle and import
ReusedExchangeExec from its execution.exchange package so the Spark 4.0 test
sources compile.
Co-authored-by: Copilot <[email protected]>
---------
Co-authored-by: Copilot <[email protected]>
---
.../backendsapi/velox/VeloxSparkPlanExecApi.scala | 13 ++
.../gluten/execution/VeloxEmptyRelationSuite.scala | 204 +++++++++++++++++++++
docs/Configuration.md | 1 +
.../gluten/backendsapi/SparkPlanExecApi.scala | 11 ++
.../org/apache/gluten/config/GlutenConfig.scala | 12 ++
.../columnar/offload/OffloadSingleNodeRules.scala | 4 +
.../execution/EmptyRelationExecTransformer.scala | 68 +++++++
.../gluten/utils/velox/VeloxTestSettings.scala | 1 -
.../velox/VeloxAdaptiveQueryExecSuite.scala | 17 ++
.../gluten/utils/velox/VeloxTestSettings.scala | 1 -
.../velox/VeloxAdaptiveQueryExecSuite.scala | 17 ++
.../org/apache/gluten/sql/shims/SparkShims.scala | 6 +
.../gluten/sql/shims/spark40/Spark40Shims.scala | 5 +
.../gluten/sql/shims/spark41/Spark41Shims.scala | 5 +
14 files changed, 363 insertions(+), 2 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
index 5bcc03c64d..c00803c521 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
@@ -1423,6 +1423,16 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi
with Logging {
VeloxColumnarToCarrierRowExec.enforce(plan)
}
+ override def isSupportEmptyRelationExec(plan: SparkPlan): Boolean = {
+ if (!GlutenConfig.get.enableColumnarEmptyRelation) {
+ logDebug(
+ "EmptyRelationExec offload skipped: " +
+ s"${GlutenConfig.COLUMNAR_EMPTY_RELATION_ENABLED.key}=false")
+ return false
+ }
+ true
+ }
+
override def isSupportLocalTableScanExec(plan: LocalTableScanExec): Boolean
= {
// `rows` is @transient, so it becomes null after Java serialization (e.g.
an AQE sub-plan
// shipped across an RPC boundary). A null rows payload signals a
deserialized plan that can
@@ -1445,6 +1455,9 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with
Logging {
true
}
+ override def getEmptyRelationExecTransform(plan: SparkPlan):
EmptyRelationExecTransformer =
+ EmptyRelationExecTransformer(plan.output)
+
override def getLocalTableScanTransform(plan: LocalTableScanExec):
LocalTableScanTransformer =
VeloxLocalTableScanTransformer.replace(plan)
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxEmptyRelationSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxEmptyRelationSuite.scala
new file mode 100644
index 0000000000..9ef39c7eb6
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxEmptyRelationSuite.scala
@@ -0,0 +1,204 @@
+/*
+ * 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.sql.shims.SparkShimLoader
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.EmptyRelationExecTransformer
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Test suite for EmptyRelationExecTransformer.
+ *
+ * EmptyRelationExec is a leaf node AQE creates (Spark 4.0+) when it proves a
subtree produces no
+ * output. Gluten offloads it to EmptyRelationExecTransformer so that
surrounding columnar operators
+ * do not need ColumnarToRow / RowToColumnar transitions around the empty
relation.
+ *
+ * Empty-result correctness is asserted on every supported Spark version; the
plan-shape assertions
+ * that depend on EmptyRelationExec are gated to Spark 4.0+, where the node
exists. The raw node is
+ * detected through `SparkShims.isEmptyRelationExec` so this suite does not
reference a class that
+ * is absent on Spark 3.x.
+ */
+class VeloxEmptyRelationSuite extends VeloxWholeStageTransformerSuite with
AdaptiveSparkPlanHelper {
+
+ override protected val resourcePath: String = "/tpch-data-parquet"
+ override protected val fileFormat: String = "parquet"
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set(GlutenConfig.COLUMNAR_EMPTY_RELATION_ENABLED.key, "true")
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ createTPCHNotNullTables()
+ }
+
+ private def countTransformers(plan:
org.apache.spark.sql.execution.SparkPlan): Int =
+ collectWithSubqueries(plan) { case _: EmptyRelationExecTransformer => true
}.size
+
+ private def countRawEmptyRelations(plan:
org.apache.spark.sql.execution.SparkPlan): Int =
+ collectWithSubqueries(plan) {
+ case p if SparkShimLoader.getSparkShims.isEmptyRelationExec(p) => true
+ }.size
+
+ /**
+ * Number of RowToColumnar transitions placed directly on top of an
EmptyRelationExecTransformer.
+ * Because the transformer is dual-mode (it produces both columnar batches
and rows), a columnar
+ * consumer must read it directly; a RowToColumnar immediately wrapping it
would mean the empty
+ * relation was executed in row mode and re-columnarized -- exactly the
ColumnarToRow /
+ * RowToColumnar sandwich this offload exists to remove. The expected count
is always 0. AQE stage
+ * wrappers between the transition and the transformer are unwrapped so the
adjacency check holds
+ * after query stages are materialized.
+ */
+ private def rowToColumnarAroundTransformer(
+ plan: org.apache.spark.sql.execution.SparkPlan): Int = {
+ def unwrap(p: org.apache.spark.sql.execution.SparkPlan)
+ : org.apache.spark.sql.execution.SparkPlan =
+ p match {
+ case stage: org.apache.spark.sql.execution.adaptive.QueryStageExec =>
unwrap(stage.plan)
+ case reused: ReusedExchangeExec => unwrap(reused.child)
+ case other => other
+ }
+ collectWithSubqueries(plan) {
+ case r2c: RowToColumnarExecBase
+ if unwrap(r2c.child).isInstanceOf[EmptyRelationExecTransformer] =>
+ true
+ }.size
+ }
+
+ // --- Empty-result correctness (all supported Spark versions) --------------
+
+ test("WHERE 1=0 produces empty result") {
+ val df = spark.sql("SELECT l_orderkey, l_partkey FROM lineitem WHERE 1 =
0")
+ assert(df.collect().isEmpty, "Expected empty result for WHERE 1=0")
+ }
+
+ test("empty UNION ALL produces empty result") {
+ val df = spark.sql("""SELECT l_orderkey FROM lineitem WHERE 1 = 0
+ |UNION ALL
+ |SELECT l_orderkey FROM lineitem WHERE 1 =
0""".stripMargin)
+ assert(df.collect().isEmpty)
+ }
+
+ test("empty result preserves string-column schema") {
+ val df = spark.sql("SELECT l_returnflag, l_linestatus, l_comment FROM
lineitem WHERE 1 = 0")
+ assert(df.collect().isEmpty)
+ assert(df.schema.fieldNames.toSeq == Seq("l_returnflag", "l_linestatus",
"l_comment"))
+ }
+
+ test("empty result preserves mixed-type schema") {
+ val df = spark.sql("""SELECT CAST(1 AS BOOLEAN) AS b,
+ | CAST(1 AS INT) AS i,
+ | CAST(1 AS BIGINT) AS l,
+ | CAST(1.0 AS DOUBLE) AS d,
+ | 'x' AS str,
+ | CAST(1.00 AS DECIMAL(10,2)) AS dec
+ |WHERE 1 = 0""".stripMargin)
+ assert(df.collect().isEmpty)
+ assert(df.schema.fields.length == 6)
+ }
+
+ test("AQE empty propagation through inner join") {
+ val df = spark.sql("""SELECT t1.l_orderkey, t2.l_partkey
+ |FROM (SELECT * FROM lineitem WHERE 1 = 0) t1
+ |INNER JOIN lineitem t2 ON t1.l_orderkey =
t2.l_orderkey""".stripMargin)
+ assert(df.collect().isEmpty)
+ }
+
+ test("AQE empty propagation through aggregation") {
+ val df = spark.sql("""SELECT l_returnflag, sum(l_quantity) AS total
+ |FROM lineitem
+ |WHERE 1 = 0
+ |GROUP BY l_returnflag""".stripMargin)
+ assert(df.collect().isEmpty)
+ }
+
+ test("empty result matches vanilla Spark") {
+ val query = "SELECT l_orderkey, l_partkey, l_quantity FROM lineitem WHERE
1 = 0"
+ var vanillaResult: Seq[Row] = Seq.empty
+ withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
+ vanillaResult = spark.sql(query).collect().toSeq
+ }
+ assert(vanillaResult.isEmpty)
+ checkAnswer(spark.sql(query), vanillaResult)
+ }
+
+ // --- Plan-shape verification (Spark 4.0+, where EmptyRelationExec exists)
--
+
+ test("EmptyRelationExec is offloaded to the columnar transformer") {
+ assume(isSparkVersionGE("4.0"))
+ // A statically-empty predicate (e.g. WHERE 1 = 0) is folded to an empty
LocalRelation by the
+ // logical optimizer and never becomes an EmptyRelationExec. That node is
produced by AQE's
+ // Propagate Empty Relations optimization when a materialized query stage
turns out to be empty
+ // at runtime, so drive it through an AQE INTERSECT whose left side is
empty only at runtime.
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
+ val df =
+ spark.sql("SELECT * FROM lineitem WHERE l_orderkey < 0 INTERSECT
SELECT * FROM lineitem")
+ assert(df.collect().isEmpty)
+ val plan = df.queryExecution.executedPlan
+ assert(
+ countTransformers(plan) > 0,
+ "Expected EmptyRelationExecTransformer in plan:\n" + plan.treeString)
+ assert(
+ countRawEmptyRelations(plan) == 0,
+ "EmptyRelationExec should be fully offloaded to the transformer:\n" +
plan.treeString)
+ assert(
+ rowToColumnarAroundTransformer(plan) == 0,
+ "No RowToColumnar transition should wrap the empty-relation
transformer; a dual-mode " +
+ "columnar leaf must be consumed directly without a
ColumnarToRow/RowToColumnar " +
+ "sandwich:\n" + plan.treeString
+ )
+ }
+ }
+
+ test("EmptyRelationExec is not offloaded when the config is disabled") {
+ assume(isSparkVersionGE("4.0"))
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ GlutenConfig.COLUMNAR_EMPTY_RELATION_ENABLED.key -> "false") {
+ val df =
+ spark.sql("SELECT * FROM lineitem WHERE l_orderkey < 0 INTERSECT
SELECT * FROM lineitem")
+ assert(df.collect().isEmpty)
+ val plan = df.queryExecution.executedPlan
+ assert(
+ countTransformers(plan) == 0,
+ "Transformer must not appear when the config is disabled:\n" +
plan.treeString)
+ }
+ }
+
+ test("EmptyRelationExec offload works under AQE runtime propagation") {
+ assume(isSparkVersionGE("4.0"))
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") {
+ val df = spark.sql("""SELECT l.l_orderkey, r.l_partkey
+ |FROM lineitem l
+ |INNER JOIN (
+ | SELECT l_orderkey, l_partkey
+ | FROM lineitem
+ | WHERE l_orderkey = -999999
+ |) r ON l.l_orderkey = r.l_orderkey""".stripMargin)
+ assert(df.collect().isEmpty, "Expected empty result from join with
impossible predicate")
+ }
+ }
+}
diff --git a/docs/Configuration.md b/docs/Configuration.md
index 3926053a17..a9b5c669e1 100644
--- a/docs/Configuration.md
+++ b/docs/Configuration.md
@@ -55,6 +55,7 @@ nav_order: 15
| spark.gluten.sql.columnar.coalesce | 🔄
Dynamic | true | Enable or disable columnar coalesce.
[...]
| spark.gluten.sql.columnar.collectLimit | 🔄
Dynamic | true | Enable or disable columnar collectLimit.
[...]
| spark.gluten.sql.columnar.collectTail | 🔄
Dynamic | true | Enable or disable columnar collectTail.
[...]
+| spark.gluten.sql.columnar.emptyRelation | 🔄
Dynamic | true | Enable or disable columnar execution of
EmptyRelationExec (Spark 4.0+). When true, Gluten replaces EmptyRelationExec (a
leaf node AQE creates when it proves a subtree produces no output) with a
columnar transformer, avoiding unnecessary ColumnarToRow / RowToColumnar
transitions around the empty relation.
[...]
| spark.gluten.sql.columnar.enableNestedColumnPruningInHiveTableScan | 🔄
Dynamic | true | Enable or disable nested column pruning in
hivetablescan.
[...]
| spark.gluten.sql.columnar.enableVanillaVectorizedReaders | âš“
Static | true | Enable or disable vanilla vectorized scan.
[...]
| spark.gluten.sql.columnar.executor.libpath | 🔄
Dynamic || The gluten executor library path.
[...]
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
index 4ca08a5ad6..56baa918c2 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
@@ -662,6 +662,17 @@ trait SparkPlanExecApi {
def getRDDScanTransform(plan: RDDScanExec): RDDScanTransformer =
throw new GlutenNotSupportException("RDDScanExec is not supported")
+ /**
+ * Whether the backend supports offloading the given empty-relation plan to
a columnar
+ * transformer. Typed as [[SparkPlan]] because EmptyRelationExec only exists
on Spark 4.0+;
+ * callers must first confirm the type through
`SparkShims.isEmptyRelationExec`.
+ */
+ def isSupportEmptyRelationExec(plan: SparkPlan): Boolean = false
+
+ /** Returns the backend transformer that replaces the given empty-relation
plan. */
+ def getEmptyRelationExecTransform(plan: SparkPlan):
EmptyRelationExecTransformer =
+ throw new GlutenNotSupportException("EmptyRelationExec is not supported")
+
def copyColumnarBatch(batch: ColumnarBatch): ColumnarBatch =
throw new GlutenNotSupportException("Copying ColumnarBatch is not
supported")
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 2bb3da9a39..0f2805c8d5 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
@@ -95,6 +95,8 @@ class GlutenConfig(conf: SQLConf) extends
GlutenCoreConfig(conf) {
def enableColumnarWindowGroupLimit: Boolean =
getConf(COLUMNAR_WINDOW_GROUP_LIMIT_ENABLED)
+ def enableColumnarEmptyRelation: Boolean =
getConf(COLUMNAR_EMPTY_RELATION_ENABLED)
+
def enableColumnarLocalTableScan: Boolean =
getConf(COLUMNAR_LOCAL_TABLE_SCAN_ENABLED)
def enableAppendData: Boolean = getConf(COLUMNAR_APPEND_DATA_ENABLED)
@@ -951,6 +953,16 @@ object GlutenConfig extends ConfigRegistry {
.booleanConf
.createWithDefault(true)
+ val COLUMNAR_EMPTY_RELATION_ENABLED =
+ buildConf("spark.gluten.sql.columnar.emptyRelation")
+ .doc(
+ "Enable or disable columnar execution of EmptyRelationExec (Spark
4.0+). When " +
+ "true, Gluten replaces EmptyRelationExec (a leaf node AQE creates
when it proves a " +
+ "subtree produces no output) with a columnar transformer, avoiding
unnecessary " +
+ "ColumnarToRow / RowToColumnar transitions around the empty
relation.")
+ .booleanConf
+ .createWithDefault(true)
+
val COLUMNAR_APPEND_DATA_ENABLED =
buildConf("spark.gluten.sql.columnar.appendData")
.doc("Enable or disable columnar v2 command append data.")
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala
index 650fbaa3d9..61520c6e42 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala
@@ -323,6 +323,10 @@ object OffloadOthers {
child)
case plan: RDDScanExec if
RDDScanTransformer.isSupportRDDScanExec(plan) =>
RDDScanTransformer.getRDDScanTransform(plan)
+ case plan
+ if SparkShimLoader.getSparkShims.isEmptyRelationExec(plan) &&
+ EmptyRelationExecTransformer.isSupportEmptyRelationExec(plan) =>
+ EmptyRelationExecTransformer.getEmptyRelationExecTransform(plan)
case plan: LocalTableScanExec
if LocalTableScanTransformer.isSupportLocalTableScanExec(plan) =>
LocalTableScanTransformer.getLocalTableScanTransform(plan)
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/EmptyRelationExecTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/EmptyRelationExecTransformer.scala
new file mode 100644
index 0000000000..5db78cafd0
--- /dev/null
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/EmptyRelationExecTransformer.scala
@@ -0,0 +1,68 @@
+/*
+ * 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
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.execution.GlutenPlan
+import org.apache.gluten.extension.columnar.transition.Convention
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+/**
+ * Columnar-aware replacement for Spark's EmptyRelationExec (Spark 4.0+).
+ *
+ * The node is dual-mode: it advertises both a columnar (primary backend
batch) output and a vanilla
+ * row output, and implements execution for both. This lets the transition
framework consume it
+ * directly from either a columnar or a row context, so an empty relation
propagated by AQE never
+ * forces surrounding operators into ColumnarToRow / RowToColumnar transitions.
+ *
+ * It carries no data, so it deliberately extends [[GlutenPlan]] rather than
+ * [[org.apache.gluten.execution.ValidatablePlan]]: native schema validation
is irrelevant for a
+ * relation that never sends rows to the backend, and applying it would
needlessly force fallback
+ * whenever the (unused) output schema contains a type the backend cannot
execute natively.
+ */
+case class EmptyRelationExecTransformer(output: Seq[Attribute])
+ extends LeafExecNode
+ with GlutenPlan {
+
+ override def rowType(): Convention.RowType =
Convention.RowType.VanillaRowType
+
+ override def batchType(): Convention.BatchType =
BackendsApiManager.getSettings.primaryBatchType
+
+ override protected def doExecute(): RDD[InternalRow] =
+ sparkContext.emptyRDD[InternalRow]
+
+ override protected def doExecuteColumnar(): RDD[ColumnarBatch] =
+ sparkContext.emptyRDD[ColumnarBatch]
+}
+
+object EmptyRelationExecTransformer {
+
+ /**
+ * Whether the backend supports offloading the given empty-relation plan to
a columnar
+ * transformer. The plan is typed as [[SparkPlan]] because EmptyRelationExec
only exists on Spark
+ * 4.0+; callers must first confirm the type through
`SparkShims.isEmptyRelationExec`.
+ */
+ def isSupportEmptyRelationExec(plan: SparkPlan): Boolean =
+
BackendsApiManager.getSparkPlanExecApiInstance.isSupportEmptyRelationExec(plan)
+
+ def getEmptyRelationExecTransform(plan: SparkPlan):
EmptyRelationExecTransformer =
+
BackendsApiManager.getSparkPlanExecApiInstance.getEmptyRelationExecTransform(plan)
+}
diff --git
a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index f7d68c0170..10f467378e 100644
---
a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -304,7 +304,6 @@ class VeloxTestSettings extends BackendTestSettings {
"SPARK-32649",
"SPARK-34533",
"SPARK-34781",
- "SPARK-35585",
"SPARK-32932",
"SPARK-33494",
"SPARK-33933",
diff --git
a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
index 67e9baed04..e105f5b7e4 100644
---
a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
+++
b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
@@ -1553,4 +1553,21 @@ class VeloxAdaptiveQueryExecSuite extends
AdaptiveQueryExecSuite with GlutenSQLT
}
}
}
+
+ // Gluten offloads EmptyRelationExec to EmptyRelationExecTransformer, so the
upstream
+ // `instanceof EmptyRelationExec` assertion no longer holds. Accept either
node.
+ testGluten("SPARK-35585: empty relation is correctly handled") {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
+ val df = spark.sql("SELECT * FROM testData WHERE key < 0 INTERSECT
SELECT * FROM testData")
+ df.collect()
+ val plan = df.queryExecution.executedPlan
+ val emptyNodes = collectWithSubqueries(plan) {
+ case e: EmptyRelationExec => e
+ case e: EmptyRelationExecTransformer => e
+ }
+ assert(
+ emptyNodes.nonEmpty,
+ "Expected EmptyRelationExec or EmptyRelationExecTransformer in
plan:\n" + plan.treeString)
+ }
+ }
}
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 9070e2626b..5b2757a95c 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -313,7 +313,6 @@ class VeloxTestSettings extends BackendTestSettings {
"SPARK-32649",
"SPARK-34533",
"SPARK-34781",
- "SPARK-35585",
"SPARK-32932",
"SPARK-33494",
"SPARK-33933",
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
index 6ebefbe15a..cb9df9e064 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
@@ -1557,4 +1557,21 @@ class VeloxAdaptiveQueryExecSuite extends
AdaptiveQueryExecSuite with GlutenSQLT
}
}
}
+
+ // Gluten offloads EmptyRelationExec to EmptyRelationExecTransformer, so the
upstream
+ // `instanceof EmptyRelationExec` assertion no longer holds. Accept either
node.
+ testGluten("SPARK-35585: empty relation is correctly handled") {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
+ val df = spark.sql("SELECT * FROM testData WHERE key < 0 INTERSECT
SELECT * FROM testData")
+ df.collect()
+ val plan = df.queryExecution.executedPlan
+ val emptyNodes = collectWithSubqueries(plan) {
+ case e: EmptyRelationExec => e
+ case e: EmptyRelationExecTransformer => e
+ }
+ assert(
+ emptyNodes.nonEmpty,
+ "Expected EmptyRelationExec or EmptyRelationExecTransformer in
plan:\n" + plan.treeString)
+ }
+ }
}
diff --git
a/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
b/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
index 9e29d934a1..5b25a59900 100644
--- a/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
+++ b/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
@@ -83,6 +83,12 @@ trait SparkShims {
def isWindowGroupLimitExec(plan: SparkPlan): Boolean = false
+ /**
+ * Whether the given plan is an EmptyRelationExec. The node only exists on
Spark 4.0+
+ * (SPARK-47217) so the default implementation returns false; Spark 4.0+
shims override it.
+ */
+ def isEmptyRelationExec(plan: SparkPlan): Boolean = false
+
def getWindowGroupLimitExecShim(plan: SparkPlan): WindowGroupLimitExecShim =
null
def getWindowGroupLimitExec(windowGroupLimitExecShim:
WindowGroupLimitExecShim): SparkPlan = null
diff --git
a/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
b/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
index a5ac265985..9ac1683dff 100644
---
a/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
+++
b/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
@@ -125,6 +125,11 @@ class Spark40Shims extends SparkShims {
case _ => false
}
+ override def isEmptyRelationExec(plan: SparkPlan): Boolean = plan match {
+ case _: EmptyRelationExec => true
+ case _ => false
+ }
+
override def getWindowGroupLimitExecShim(plan: SparkPlan):
WindowGroupLimitExecShim = {
val windowGroupLimitPlan = plan.asInstanceOf[WindowGroupLimitExec]
val mode = windowGroupLimitPlan.mode match {
diff --git
a/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
b/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
index ae145a1f0a..c7cd8fed15 100644
---
a/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
+++
b/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
@@ -124,6 +124,11 @@ class Spark41Shims extends SparkShims {
case _ => false
}
+ override def isEmptyRelationExec(plan: SparkPlan): Boolean = plan match {
+ case _: EmptyRelationExec => true
+ case _ => false
+ }
+
override def getWindowGroupLimitExecShim(plan: SparkPlan):
WindowGroupLimitExecShim = {
val windowGroupLimitPlan = plan.asInstanceOf[WindowGroupLimitExec]
val mode = windowGroupLimitPlan.mode match {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
