[CORE][VL] Add columnar EmptyRelationExec offload to Velox backend - #12766
[CORE][VL] Add columnar EmptyRelationExec offload to Velox backend#12766minni31 wants to merge 9 commits into
Conversation
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 <223556219+Copilot@users.noreply.github.com>
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
This PR adds a Spark 4.0+ shim-based detection path and a new columnar leaf transformer for EmptyRelationExec, allowing Gluten (Velox backend) to avoid unnecessary ColumnarToRow / RowToColumnar transitions around proven-empty subtrees produced by AQE’s “Propagate Empty Relations”.
Changes:
- Add
SparkShims.isEmptyRelationExec(defaultfalse) with Spark 4.0/4.1 overrides, and wireEmptyRelationExecTransformerintoOffloadOthers. - Introduce
EmptyRelationExecTransformer(emptyRDD[ColumnarBatch]) and a new dynamic configspark.gluten.sql.columnar.emptyRelation(defaulttrue). - Add/adjust Spark 4.0/4.1 Velox tests to validate plan shape + correctness and handle SPARK-35585 expectations.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala | Adds Spark 4.1 shim detection for EmptyRelationExec. |
| shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala | Adds Spark 4.0 shim detection for EmptyRelationExec. |
| shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala | Adds version-agnostic isEmptyRelationExec API with safe default. |
| gluten-substrait/src/main/scala/org/apache/spark/sql/execution/EmptyRelationExecTransformer.scala | New columnar leaf transformer producing an empty RDD[ColumnarBatch]. |
| gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala | Adds EmptyRelationExec → transformer offload case guarded by shims + backend support. |
| gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala | Adds new dynamic config accessor + entry for empty relation offload. |
| gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala | Adds backend API hooks to support and create the transformer. |
| backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala | Implements support gate + transformer construction for Velox backend. |
| docs/Configuration.md | Documents the new spark.gluten.sql.columnar.emptyRelation config. |
| backends-velox/src/test/scala/org/apache/gluten/execution/VeloxEmptyRelationSuite.scala | New suite validating correctness + Spark 4.0+ plan-shape behavior and config disablement. |
| gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala | Adds Gluten-specific SPARK-35585 assertion accepting either raw or transformer node. |
| gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala | Adjusts included AQE test selection to account for new Gluten-specific coverage. |
| gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala | Same SPARK-35585 Gluten-specific assertion for Spark 4.0 test module. |
| gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala | Adjusts included AQE test selection for Spark 4.0 module. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
… 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).
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 14 out of 14 changed files in this pull request and generated no new comments.
Suppressed comments (1)
gluten-substrait/src/main/scala/org/apache/spark/sql/execution/EmptyRelationExecTransformer.scala:52
withNewChildrenInternalcurrently ignoresnewChildrenunconditionally. For a leaf node, it’s safer to enforce the invariant that no children are provided (otherwise a caller attempting a tree rewrite could silently fail to apply updates). Add arequire(newChildren.isEmpty, ...)guard (or otherwise validate) before returningthis.
override protected def withNewChildrenInternal(
newChildren: IndexedSeq[SparkPlan]): SparkPlan = this
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.
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 14 out of 14 changed files in this pull request and generated no new comments.
Suppressed comments (1)
gluten-substrait/src/main/scala/org/apache/spark/sql/execution/EmptyRelationExecTransformer.scala:53
- EmptyRelationExecTransformer is intended as a semantic replacement for Spark's EmptyRelationExec, but
doExecute()currently throws. If this node ever ends up on a row-based path (e.g., under a non-offloaded parent, or if a transition is missed), the query would fail instead of producing an empty result. Consider returning an emptyRDD[InternalRow]here. Also, since this is a leaf exec, extendingLeafExecNodelets Spark handlechildren/withNewChildrenInternalinvariants and avoids silently ignoring non-emptynewChildren.
override protected def doExecute(): RDD[InternalRow] =
throw new UnsupportedOperationException(
"EmptyRelationExecTransformer does not support row execution.")
override protected def doExecuteColumnar(): RDD[ColumnarBatch] =
sparkContext.emptyRDD[ColumnarBatch]
override def children: Seq[SparkPlan] = Seq.empty
override protected def withNewChildrenInternal(
newChildren: IndexedSeq[SparkPlan]): SparkPlan = this
…RelationExec 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 only at runtime, mirroring the upstream SPARK-35585 trigger, and traverse with collectWithSubqueries. The config-gate test now exercises the same real EmptyRelationExec instead of passing vacuously.
|
Run Gluten Clickhouse CI on x86 |
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.
|
Run Gluten Clickhouse CI on x86 |
minni31
changed the title
| * through the plan. | ||
| */ | ||
| case class EmptyRelationExecTransformer(output: Seq[Attribute]) | ||
| extends LeafExecNode |
There was a problem hiding this comment.
Consider making this data-free JVM leaf dual-mode and schema-independent
Problem: This transformer emits empty JVM RDDs and never sends data to Velox, but ValidatablePlan still applies backend schema validation and RowType.None makes the node batch-only. Consequently, an empty relation with a Velox-unsupported output type can remain vanilla, and a row consumer still needs a terminal columnar-to-row transition even though producing an empty row RDD is trivial.
Evidence:
case class EmptyRelationExecTransformer(output: Seq[Attribute])
extends LeafExecNode
with ValidatablePlan {
override def rowType(): Convention.RowType = Convention.RowType.None
override protected def doValidateInternal(): ValidationResult = ValidationResult.succeeded
override protected def doExecute(): RDD[InternalRow] =
throw new UnsupportedOperationException(
"EmptyRelationExecTransformer does not support row execution.")Suggested Fix: Consider modeling this as a dual-mode GlutenPlan directly, avoiding irrelevant native schema validation while supporting both empty execution modes:
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]
}There was a problem hiding this comment.
Thanks @weiting-chen — great point, done in ba8302e12. The transformer now extends GlutenPlan instead of ValidatablePlan, so it no longer goes through FallbackByNativeValidation (which only validates ValidatablePlan nodes) — native schema validation is irrelevant for a leaf that never sends data to Velox, and this removes the spurious fallback when the unused output schema has a backend-unsupported type.
It's now genuinely dual-mode: rowType() = VanillaRowType + batchType() = primaryBatchType, with both doExecute → emptyRDD[InternalRow] and doExecuteColumnar → emptyRDD[ColumnarBatch] implemented. So a row consumer reads it directly (no terminal C2R) and a columnar consumer reads the empty batch RDD — the transition framework picks either with zero transitions. I verified ConventionFunc.checkRowType/checkBatchType stay consistent since GlutenPlan derives supportsRowBased/supportsColumnar from these (both non-None).
| countTransformers(plan) > 0, | ||
| "Expected EmptyRelationExecTransformer in plan:\n" + plan.treeString) | ||
| assert( | ||
| countRawEmptyRelations(plan) == 0, |
There was a problem hiding this comment.
Assert that the transition sandwich is actually eliminated
Problem: These assertions prove that EmptyRelationExec was replaced, but they do not verify the PR's central performance invariant: no ColumnarToRow -> RowToColumnar sandwich remains around the transformer. A future transition-planning regression could preserve the transformer and still pass this test while losing the optimization.
Evidence:
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)Suggested Fix: Add an explicit adjacency check for the redundant sandwich while still allowing a terminal columnar-to-row transition required by collect():
val transitionSandwiches = collectWithSubqueries(plan) {
case r2c
if r2c.nodeName.contains("RowToColumnar") &&
r2c.children.exists(c2r =>
c2r.nodeName.contains("ColumnarToRow") &&
c2r.children.exists(_.isInstanceOf[EmptyRelationExecTransformer])) => true
}
assert(
transitionSandwiches.isEmpty,
"Unexpected C2R/R2C sandwich around EmptyRelationExecTransformer:\n" + plan.treeString)There was a problem hiding this comment.
Done in ba8302e12. Added a rowToColumnarAroundTransformer helper that asserts no RowToColumnar transition sits directly on top of the transformer, so a transition-planning regression that reintroduces the C2R→R2C sandwich now fails the test — while still allowing the terminal C2R that collect() legitimately needs.
I used a strict-adjacency check (unwrap(r2c.child).isInstanceOf[EmptyRelationExecTransformer]) matching your suggestion, and additionally unwrap AQE QueryStageExec/ReusedExchangeExec wrappers so the adjacency holds after query stages materialize under the AQE-driven test path.
…t 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 <223556219+Copilot@users.noreply.github.com>
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
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 <223556219+Copilot@users.noreply.github.com>
812f2f6 to
18061b8
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 14 out of 14 changed files in this pull request and generated 1 comment.
Suppressed comments (2)
gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala:954
- The docstring implies this flag controls behavior everywhere, but
EmptyRelationExecdoesn’t exist on Spark < 4.0 so the config is effectively a no-op there. Consider explicitly documenting that it only applies on Spark 4.0+ (and is ignored on Spark 3.x) both here and indocs/Configuration.md, to prevent confusion when users toggle it on older Spark versions.
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)
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala:1433
- Avoid
returnin Scala here; it makes the method less idiomatic and slightly harder to refactor. This can be expressed as a single expression (e.g., compute the boolean, log when false, and return the boolean), which also reduces the chance of accidentally adding code after areturnthat never runs.
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
}
| // 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) | ||
| } | ||
| } |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
What changes are proposed in this pull request?
This PR offloads
EmptyRelationExecto the Velox backend so that empty relations are executed columnarly instead of forcing a fallback to vanilla row execution.EmptyRelationExecis a leaf node that AQE's Propagate Empty Relations optimization creates on Spark 4.0+ when a materialized query stage turns out to be empty at runtime (e.g. anINTERSECTor join whose input is empty only at runtime, or an aggregation over an empty runtime stage). A statically-empty predicate such asWHERE 1 = 0does not reach this path — the logical optimizer folds it to an emptyLocalRelation(aLocalTableScan) before physical planning. Because its defaultsupportsColumnarisfalse, Gluten currently wraps it inColumnarToRow/RowToColumnartransitions even though it returns zero rows. This PR addsEmptyRelationExecTransformer, a columnar leaf that returns an emptyRDD[ColumnarBatch], eliminating those transitions.EmptyRelationExeconly exists on Spark 4.0+ (SPARK-47217), so the node is never referenced from version-agnostic modules. Detection is routed through a newSparkShims.isEmptyRelationExec, which defaults tofalseand is overridden only in the Spark 4.0 and 4.1 shims. The sharedOffloadOthersrule and theSparkPlanExecApitrait therefore continue to compile unchanged against Spark 3.3–3.5.The offload is gated by a new config
spark.gluten.sql.columnar.emptyRelation(defaulttrue). The Velox backend implementsisSupportEmptyRelationExec; other backends inherit the trait default and keep vanilla execution.How was this patch tested?
VeloxEmptyRelationSuite: empty-result correctness across various schemas, emptyUNION ALL, AQE propagation through joins/aggregations, and side-by-side parity with vanilla Spark (asserted on all supported Spark versions). Plan-shape assertions (transformer present, no residualEmptyRelationExec) and the config-disabled negative case are gated to Spark 4.0+, where the node exists.SPARK-35585AQE test on Spark 4.0/4.1 with a Gluten-aware assertion that accepts eitherEmptyRelationExecorEmptyRelationExecTransformer.Was this patch authored or co-authored using generative AI tooling?
Generated-by: GitHub Copilot (Claude Opus 4.8)