This is an automated email from the ASF dual-hosted git repository.
JkSelf 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 0ff76818b6 [GLUTEN-12008][VL] Align Expand projection types with
output (#12009)
0ff76818b6 is described below
commit 0ff76818b6ae7bbc61d6e623a44e8084099c7dbc
Author: jianzhenwu <[email protected]>
AuthorDate: Thu Jul 16 10:40:18 2026 +0800
[GLUTEN-12008][VL] Align Expand projection types with output (#12009)
---
.../gluten/backendsapi/velox/VeloxRuleApi.scala | 11 +-
.../spark/sql/execution/VeloxExpandSuite.scala | 137 ++++++++++++++++++++-
.../columnar/rewrite/AlignExpandOutputTypes.scala | 82 ++++++++++++
.../rewrite/AlignExpandOutputTypesSuite.scala | 77 ++++++++++++
4 files changed, 302 insertions(+), 5 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
index d63928527d..dd492c90bb 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
@@ -34,6 +34,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.datasources.noop.GlutenNoopWriterRule
+import org.apache.spark.util.SparkVersionUtil
class VeloxRuleApi extends RuleApi {
import VeloxRuleApi._
@@ -96,10 +97,12 @@ object VeloxRuleApi {
Seq(
RewriteIn,
RewriteMultiChildrenCount,
- RewriteJoin,
- PullOutPreProject,
- PullOutPostProject,
- ProjectColumnPruning)
+ RewriteJoin) ++
+ (if (SparkVersionUtil.eqSpark33) Seq(AlignExpandOutputTypes) else
Seq.empty) ++
+ Seq(
+ PullOutPreProject,
+ PullOutPostProject,
+ ProjectColumnPruning)
injector.injectTransform(
c =>
HeuristicTransform.WithRewrites(
diff --git
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxExpandSuite.scala
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxExpandSuite.scala
index 1af4e9fba0..6fead0ce2a 100644
---
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxExpandSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxExpandSuite.scala
@@ -18,11 +18,14 @@ package org.apache.spark.sql.execution
import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.events.GlutenPlanFallbackEvent
-import org.apache.gluten.execution.VeloxWholeStageTransformerSuite
+import org.apache.gluten.execution.{ExpandExecTransformer,
VeloxWholeStageTransformerSuite}
import org.apache.spark.SparkConf
import org.apache.spark.internal.config.UI.UI_ENABLED
import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+
+import java.sql.Date
import scala.collection.mutable.ArrayBuffer
@@ -33,10 +36,18 @@ class VeloxExpandSuite extends
VeloxWholeStageTransformerSuite {
override def sparkConf: SparkConf = {
super.sparkConf
.set(GlutenConfig.GLUTEN_UI_ENABLED.key, "true")
+ .set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
// The gluten ui event test suite expects the spark ui to be enable
.set(UI_ENABLED, true)
}
+ private def assertContainsNativeExpand(df: DataFrame): Unit = {
+ assert(
+ getExecutedPlan(df).exists(_.isInstanceOf[ExpandExecTransformer]),
+ s"Expected ExpandExecTransformer in plan,
got:\n${df.queryExecution.executedPlan}"
+ )
+ }
+
test("Expand with duplicated group keys") {
withTable("t1") {
val events = new ArrayBuffer[GlutenPlanFallbackEvent]
@@ -77,4 +88,128 @@ class VeloxExpandSuite extends
VeloxWholeStageTransformerSuite {
}
}
}
+
+ test("Expand with round(avg(decimal)) and multiple distinct aggregates") {
+ withTempPath {
+ pendingPath =>
+ withTempPath {
+ verifiedPath =>
+ withTempView("pending_events", "verified_events") {
+ spark
+ .sql("""
+ |SELECT * FROM VALUES
+ | (1L, DATE'2026-04-22', 'A24', 0L),
+ | (2L, DATE'2026-04-22', 'A24', 0L)
+ |AS pending_events(order_id, pending_date,
pending_reason, pending_timestamp)
+ |""".stripMargin)
+ .write
+ .mode("overwrite")
+ .parquet(pendingPath.getCanonicalPath)
+
+ spark
+ .sql("""
+ |SELECT * FROM VALUES
+ | (1L, 90000L),
+ | (2L, 180000L)
+ |AS verified_events(order_id, verified_timestamp)
+ |""".stripMargin)
+ .write
+ .mode("overwrite")
+ .parquet(verifiedPath.getCanonicalPath)
+
+ spark.read
+ .parquet(pendingPath.getCanonicalPath)
+ .createOrReplaceTempView("pending_events")
+ spark.read
+ .parquet(verifiedPath.getCanonicalPath)
+ .createOrReplaceTempView("verified_events")
+
+ val df = spark.sql(
+ """
+ |WITH sla_calc AS (
+ | SELECT
+ | p.pending_date,
+ | p.pending_reason,
+ | p.order_id,
+ | round(
+ | cast((v.verified_timestamp - p.pending_timestamp) as
decimal(38, 18)) /
+ | 3600.000000000000000000,
+ | 1) AS sla_hours
+ | FROM pending_events p
+ | JOIN verified_events v
+ | ON p.order_id = v.order_id
+ |)
+ |SELECT
+ | pending_date,
+ | pending_reason,
+ | COUNT(DISTINCT order_id) AS total_order,
+ | round(AVG(sla_hours), 1) AS avg_sla_hours,
+ | COUNT(DISTINCT CASE WHEN sla_hours > 24 THEN order_id
END) AS backlog_24,
+ | COUNT(DISTINCT CASE WHEN sla_hours > 48 THEN order_id
END) AS backlog_48
+ |FROM sla_calc
+ |GROUP BY pending_date, pending_reason
+ |""".stripMargin)
+
+ checkAnswer(
+ df,
+ Row(Date.valueOf("2026-04-22"), "A24", 2L, BigDecimal("37.5"),
2L, 1L))
+ assertContainsNativeExpand(df)
+ }
+ }
+ }
+ }
+
+ test("Expand with decimal case-when sum and multiple distinct aggregates") {
+ withTempPath {
+ eventsPath =>
+ withTempView("smart_events") {
+ spark
+ .sql("""
+ |SELECT * FROM VALUES
+ | (1, 101L, 1001L, 1, 0, 1,
+ | CAST(1.1000000000 AS DECIMAL(25, 10)),
+ | CAST(2.2000000000 AS DECIMAL(25, 10)),
+ | CAST(3.3000000000 AS DECIMAL(25, 10))),
+ | (1, 102L, 1002L, 0, 1, 0,
+ | CAST(4.4000000000 AS DECIMAL(25, 10)),
+ | CAST(5.5000000000 AS DECIMAL(25, 10)),
+ | CAST(6.6000000000 AS DECIMAL(25, 10)))
+ |AS smart_events(
+ | campaign_id,
+ | order_id,
+ | checkout_id,
+ | has_dd,
+ | has_ccb,
+ | has_fsv,
+ | dd_cost_usd,
+ | ccb_cost_usd,
+ | fsv_cost_usd)
+ |""".stripMargin)
+ .write
+ .mode("overwrite")
+ .parquet(eventsPath.getCanonicalPath)
+
+
spark.read.parquet(eventsPath.getCanonicalPath).createOrReplaceTempView("smart_events")
+
+ val df =
+ spark.sql("""
+ |SELECT
+ | campaign_id,
+ | COUNT(DISTINCT order_id) AS total_order,
+ | COUNT(DISTINCT CASE WHEN has_dd = 1 THEN order_id
END) AS dd_order,
+ | COUNT(DISTINCT checkout_id) AS checkout_count,
+ | SUM(
+ | CASE WHEN has_dd = 1 THEN dd_cost_usd ELSE 0 END +
+ | CASE WHEN has_ccb = 1 THEN ccb_cost_usd ELSE 0
END +
+ | CASE WHEN has_fsv = 1 THEN fsv_cost_usd ELSE 0 END
+ | ) AS smart_voucher_cost_usd
+ |FROM smart_events
+ |GROUP BY campaign_id
+ |""".stripMargin)
+
+ checkAnswer(df, Row(1, 2L, 1L, 2L, BigDecimal("9.9000000000")))
+ assertContainsNativeExpand(df)
+ }
+ }
+ }
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypes.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypes.scala
new file mode 100644
index 0000000000..1fe963ab45
--- /dev/null
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypes.scala
@@ -0,0 +1,82 @@
+/*
+ * 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.extension.columnar.rewrite
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.execution.{ExpandExec, SparkPlan}
+import org.apache.spark.sql.types.DataType
+
+/**
+ * Spark 3.3 may produce Expand projections whose expression output types do
not exactly match the
+ * corresponding Expand output attributes. Spark's row path tolerates this,
but native Expand
+ * conversion requires each projection column to have a consistent type.
+ *
+ * This rule rewrites each projection column: null literals are replaced with
a typed null matching
+ * the output type; non-matching expressions are wrapped in a Cast to the
output type.
+ */
+object AlignExpandOutputTypes extends RewriteSingleNode {
+ override def isRewritable(plan: SparkPlan): Boolean = {
+ plan match {
+ case _: ExpandExec => true
+ case _ => false
+ }
+ }
+
+ override def rewrite(plan: SparkPlan): SparkPlan = plan match {
+ case expand: ExpandExec =>
+ val alignedProjections = ExpandOutputTypeAlignment.alignProjections(
+ expand.projections,
+ expand.output,
+ expand.child.output)
+ if (alignedProjections == expand.projections) {
+ expand
+ } else {
+ val newExpand = expand.copy(projections = alignedProjections)
+ newExpand.copyTagsFrom(expand)
+ newExpand
+ }
+ case _ => plan
+ }
+}
+
+private[gluten] object ExpandOutputTypeAlignment {
+ def alignProjections(
+ projections: Seq[Seq[Expression]],
+ output: Seq[Attribute],
+ inputAttributes: Seq[Attribute]): Seq[Seq[Expression]] = {
+ projections.map {
+ projection =>
+ projection.zipWithIndex.map {
+ case (expression, colIdx) if colIdx < output.length =>
+ alignExpression(expression, output(colIdx).dataType)
+ case (expression, _) =>
+ expression
+ }
+ }
+ }
+
+ private def alignExpression(expression: Expression, outputType: DataType):
Expression = {
+ expression match {
+ case Literal(null, _) =>
+ Literal.create(null, outputType)
+ case _ if expression.dataType != outputType =>
+ Cast(expression, outputType)
+ case _ =>
+ expression
+ }
+ }
+}
diff --git
a/gluten-substrait/src/test/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypesSuite.scala
b/gluten-substrait/src/test/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypesSuite.scala
new file mode 100644
index 0000000000..56bab3a745
--- /dev/null
+++
b/gluten-substrait/src/test/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypesSuite.scala
@@ -0,0 +1,77 @@
+/*
+ * 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.extension.columnar.rewrite
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.types._
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class AlignExpandOutputTypesSuite extends AnyFunSuite {
+ test("align null literal type to expand output type") {
+ val outputType = DecimalType(20, 6)
+ val expression = Literal.create(null, NullType)
+ val output = AttributeReference("value", outputType, nullable = true)()
+
+ val aligned = ExpandOutputTypeAlignment.alignProjections(
+ Seq(Seq(expression)),
+ Seq(output),
+ Seq.empty)
+
+ assert(aligned.head.head == Literal.create(null, outputType))
+ }
+
+ test("cast decimal expression whose result type differs from output type") {
+ val inputType = DecimalType(20, 6)
+ val outputType = DecimalType(30, 6)
+ val amount = AttributeReference("amount", inputType, nullable = true)()
+ val expression = Add(amount, Literal(Decimal(1), inputType))
+ val output = AttributeReference("value", outputType, nullable = true)()
+
+ val aligned = ExpandOutputTypeAlignment.alignProjections(
+ Seq(Seq(expression)),
+ Seq(output),
+ Seq(amount))
+
+ aligned.head.head match {
+ case Cast(child, castType, _, _) =>
+ assert(child == expression)
+ assert(castType == outputType)
+ case other =>
+ fail(s"Expected decimal expression to be cast to $outputType, got
$other")
+ }
+ }
+
+ test("cast non-decimal expression whose result type differs from output
type") {
+ val id = AttributeReference("id", IntegerType, nullable = false)()
+ val expression = Add(id, Literal(1))
+ val output = AttributeReference("value", LongType, nullable = false)()
+
+ val aligned = ExpandOutputTypeAlignment.alignProjections(
+ Seq(Seq(expression)),
+ Seq(output),
+ Seq(id))
+
+ aligned.head.head match {
+ case Cast(child, castType, _, _) =>
+ assert(child == expression)
+ assert(castType == LongType)
+ case other =>
+ fail(s"Expected integer expression to be cast to LongType, got $other")
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
