weiting-chen commented on code in PR #12766: URL: https://github.com/apache/gluten/pull/12766#discussion_r3795168769
########## gluten-substrait/src/main/scala/org/apache/spark/sql/execution/EmptyRelationExecTransformer.scala: ########## @@ -0,0 +1,64 @@ +/* + * 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.{ValidatablePlan, ValidationResult} +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+). It produces an empty + * RDD[ColumnarBatch] so that surrounding columnar operators do not need to be wrapped in + * unnecessary ColumnarToRow / RowToColumnar transitions when AQE propagates an empty relation + * through the plan. + */ +case class EmptyRelationExecTransformer(output: Seq[Attribute]) + extends LeafExecNode Review 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:** ```scala 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: ```scala 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] } ``` ########## backends-velox/src/test/scala/org/apache/gluten/execution/VeloxEmptyRelationSuite.scala: ########## @@ -0,0 +1,172 @@ +/* + * 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.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 + + // --- 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, Review 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:** ```scala 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()`: ```scala 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) ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
