From 98a4b86ac1bb9880a3237fe2da170bbbb61b0e54 Mon Sep 17 00:00:00 2001 From: Kapil Singh Date: Thu, 5 Feb 2026 20:08:55 +0530 Subject: [PATCH 1/7] enable suites --- .../org/apache/gluten/utils/velox/VeloxTestSettings.scala | 6 +++--- .../org/apache/gluten/utils/velox/VeloxTestSettings.scala | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) 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 a2deb7740f8..fcd150e599d 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 @@ -770,7 +770,7 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenSSBQuerySuite] enableSuite[GlutenSessionStateSuite] // TODO: 4.x enableSuite[GlutenSetCommandSuite] // 1 failure - // TODO: 4.x enableSuite[GlutenSingleLevelAggregateHashMapSuite] // 1 failure + enableSuite[GlutenSingleLevelAggregateHashMapSuite] enableSuite[GlutenSparkSessionBuilderSuite] // TODO: 4.x enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite] // 1 failure enableSuite[GlutenTPCDSCollationQueryTestSuite] @@ -786,8 +786,8 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenTPCDSV2_7_PlanStabilityWithStatsSuite] enableSuite[GlutenTPCHPlanStabilitySuite] enableSuite[GlutenTPCHQuerySuite] - // TODO: 4.x enableSuite[GlutenTwoLevelAggregateHashMapSuite] // 1 failure - // TODO: 4.x enableSuite[GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite] // 1 failure + enableSuite[GlutenTwoLevelAggregateHashMapSuite] + enableSuite[GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite] enableSuite[GlutenUDFSuite] enableSuite[GlutenUDTRegistrationSuite] enableSuite[GlutenUnsafeRowSuite] 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 b247659b0d5..87b78fddc99 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 @@ -735,7 +735,7 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenSSBQuerySuite] enableSuite[GlutenSessionStateSuite] // TODO: 4.x enableSuite[GlutenSetCommandSuite] // 1 failure - // TODO: 4.x enableSuite[GlutenSingleLevelAggregateHashMapSuite] // 1 failure + enableSuite[GlutenSingleLevelAggregateHashMapSuite] enableSuite[GlutenSparkSessionBuilderSuite] // TODO: 4.x enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite] // 1 failure enableSuite[GlutenTPCDSCollationQueryTestSuite] @@ -751,8 +751,8 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenTPCDSV2_7_PlanStabilityWithStatsSuite] enableSuite[GlutenTPCHPlanStabilitySuite] enableSuite[GlutenTPCHQuerySuite] - // TODO: 4.x enableSuite[GlutenTwoLevelAggregateHashMapSuite] // 1 failure - // TODO: 4.x enableSuite[GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite] // 1 failure + enableSuite[GlutenTwoLevelAggregateHashMapSuite] + enableSuite[GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite] enableSuite[GlutenUDFSuite] enableSuite[GlutenUDTRegistrationSuite] enableSuite[GlutenUnsafeRowSuite] From 5012d5f21f5cf037cd695cbb79cf62773d7f687c Mon Sep 17 00:00:00 2001 From: Kapil Singh Date: Fri, 6 Feb 2026 19:04:38 +0530 Subject: [PATCH 2/7] Update trait --- .../org/apache/spark/sql/GlutenAggregateHashMapSuite.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala index 3f576c9c1a5..49d786b4e2b 100644 --- a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala +++ b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala @@ -18,12 +18,12 @@ package org.apache.spark.sql class GlutenSingleLevelAggregateHashMapSuite extends SingleLevelAggregateHashMapSuite - with GlutenTestsCommonTrait {} + with GlutenSQLTestsTrait {} class GlutenTwoLevelAggregateHashMapSuite extends TwoLevelAggregateHashMapSuite - with GlutenTestsCommonTrait {} + with GlutenSQLTestsTrait {} class GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite extends TwoLevelAggregateHashMapWithVectorizedMapSuite - with GlutenTestsCommonTrait {} + with GlutenSQLTestsTrait {} From 43dec933833e1de0d04a54f2baba623eff1c5f94 Mon Sep 17 00:00:00 2001 From: Kapil Singh Date: Sat, 7 Feb 2026 00:12:01 +0530 Subject: [PATCH 3/7] Fix for single suite --- .../utils/velox/VeloxTestSettings.scala | 6 + .../sql/GlutenAggregateHashMapSuite.scala | 149 +++++++++++++++++- 2 files changed, 154 insertions(+), 1 deletion(-) 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 87b78fddc99..1c53edf75df 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 @@ -736,6 +736,12 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenSessionStateSuite] // TODO: 4.x enableSuite[GlutenSetCommandSuite] // 1 failure enableSuite[GlutenSingleLevelAggregateHashMapSuite] + .exclude("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") + .exclude( + "SPARK-19471: AggregationIterator does not initialize the generated result projection before using it") + // Checks Spark specific codegen code for HashMap + .exclude("SPARK-43876: Enable fast hashmap for distinct queries") + .excludeByPrefix("SPARK-31620: agg with subquery (whole-stage-codegen =") enableSuite[GlutenSparkSessionBuilderSuite] // TODO: 4.x enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite] // 1 failure enableSuite[GlutenTPCDSCollationQueryTestSuite] diff --git a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala index 49d786b4e2b..ed89f43f115 100644 --- a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala +++ b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala @@ -16,9 +16,156 @@ */ package org.apache.spark.sql +import org.apache.gluten.execution.{HashAggregateExecTransformer, WholeStageTransformer} + +import org.apache.spark.sql.execution.aggregate.{ObjectHashAggregateExec, SortAggregateExec} +import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec +import org.apache.spark.sql.functions.{col, collect_list, expr, monotonically_increasing_id, rand, randn, spark_partition_id, sum} +import org.apache.spark.sql.internal.SQLConf + +import scala.util.Random + class GlutenSingleLevelAggregateHashMapSuite extends SingleLevelAggregateHashMapSuite - with GlutenSQLTestsTrait {} + with GlutenSQLTestsTrait { + import testImplicits._ + + private def assertNoExceptions(wholeStageBoundary: Boolean, c: Column): Unit = { + for ( + (wholeStage, useObjectHashAgg) <- + Seq((true, true), (true, false), (false, true), (false, false)) + ) { + withSQLConf( + (SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key, wholeStage.toString), + (SQLConf.USE_OBJECT_HASH_AGG.key, useObjectHashAgg.toString)) { + + val df = Seq(("1", 1), ("1", 2), ("2", 3), ("2", 4)).toDF("x", "y") + + // test case for HashAggregate + // Converts to HashAggregateExecTransformer + val hashAggDF = df.groupBy("x").agg(c, sum("y")) + hashAggDF.collect() + val hashAggPlan = hashAggDF.queryExecution.executedPlan + assert(find(hashAggPlan) { + case WholeStageTransformer(_: HashAggregateExecTransformer, _) if wholeStageBoundary => + true + case _: HashAggregateExecTransformer if !wholeStageBoundary => true + case _ => false + }.isDefined) + + // test case for ObjectHashAggregate and SortAggregate + // Both gets converted to HashAggregateExecTransformer + val objHashAggOrSortAggDF = df.groupBy("x").agg(c, collect_list("y")) + objHashAggOrSortAggDF.collect() + val objHashAggOrSortAggPlan = + stripAQEPlan(objHashAggOrSortAggDF.queryExecution.executedPlan) + assert(find(objHashAggOrSortAggPlan) { + case WholeStageTransformer(_: HashAggregateExecTransformer, _) if wholeStageBoundary => + true + case _: HashAggregateExecTransformer if !wholeStageBoundary => true + case _ => false + }.isDefined) + } + } + } + + testGluten( + "SPARK-19471: AggregationIterator does not initialize the generated result projection" + + " before using it") { + Seq( + (true, monotonically_increasing_id()), + (false, spark_partition_id()), + (false, rand(Random.nextLong())), + (false, randn(Random.nextLong())) + ).foreach(tup => assertNoExceptions(tup._1, tup._2)) + } + + testGluten("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") { + withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") { + val df = Seq(("1", "2", 1), ("1", "2", 2), ("2", "3", 3), ("2", "3", 4)) + .toDF("a", "b", "c") + .repartition(col("a")) + + val objHashAggDF = df + .withColumn("d", expr("(a, b, c)")) + .groupBy("a", "b") + .agg(collect_list("d").as("e")) + .withColumn("f", expr("(b, e)")) + .groupBy("a") + .agg(collect_list("f").as("g")) + val aggPlan = objHashAggDF.queryExecution.executedPlan + + val sortAggPlans = collect(aggPlan) { case sortAgg: SortAggregateExec => sortAgg } + // SortAggregate will be retained due velox_collect_list + assert(sortAggPlans.size == 4) + + val objHashAggPlans = collect(aggPlan) { + case objHashAgg: ObjectHashAggregateExec => objHashAgg + } + assert(objHashAggPlans.isEmpty) + + val exchangePlans = collect(aggPlan) { case shuffle: ShuffleExchangeExec => shuffle } + assert(exchangePlans.length == 1) + } + } + + Seq(true, false).foreach { + value => + testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") { + withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> value.toString) { + withTempView("t1", "t2") { + sql("create temporary view t1 as select * from values (1, 2) as t1(a, b)") + sql("create temporary view t2 as select * from values (3, 4) as t2(c, d)") + + // test without grouping keys + checkAnswer( + sql("select sum(if(c > (select a from t1), d, 0)) as csum from t2"), + Row(4) :: Nil) + + // test with grouping keys + checkAnswer( + sql( + "select c, sum(if(c > (select a from t1), d, 0)) as csum from " + + "t2 group by c"), + Row(3, 4) :: Nil) + + // test with distinct + checkAnswer( + sql( + "select avg(distinct(d)), sum(distinct(if(c > (select a from t1)," + + " d, 0))) as csum from t2 group by c"), + Row(4, 4) :: Nil) + + // test subquery with agg + checkAnswer( + sql( + "select sum(distinct(if(c > (select sum(distinct(a)) from t1)," + + " d, 0))) as csum from t2 group by c"), + Row(4) :: Nil) + + // test SortAggregateExec + // Converts to HashAggregateExecTransformer + var df = sql("select max(if(c > (select a from t1), 'str1', 'str2')) as csum from t2") + df.collect() + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[HashAggregateExecTransformer]).isDefined) + checkAnswer(df, Row("str1") :: Nil) + + // test ObjectHashAggregateExec + // Converts to HashAggregateExecTransformer + df = + sql("select collect_list(d), sum(if(c > (select a from t1), d, 0)) as csum from t2") + df.collect() + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[HashAggregateExecTransformer]).isDefined) + checkAnswer(df, Row(Array(4), 4) :: Nil) + } + } + } + } +} class GlutenTwoLevelAggregateHashMapSuite extends TwoLevelAggregateHashMapSuite From faf035be21d2c39527badacbba315e614b3286d9 Mon Sep 17 00:00:00 2001 From: Kapil Singh Date: Sat, 7 Feb 2026 07:46:03 +0530 Subject: [PATCH 4/7] restore --- .../utils/velox/VeloxTestSettings.scala | 6 - .../sql/GlutenAggregateHashMapSuite.scala | 153 +----------------- 2 files changed, 3 insertions(+), 156 deletions(-) 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 1c53edf75df..87b78fddc99 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 @@ -736,12 +736,6 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenSessionStateSuite] // TODO: 4.x enableSuite[GlutenSetCommandSuite] // 1 failure enableSuite[GlutenSingleLevelAggregateHashMapSuite] - .exclude("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") - .exclude( - "SPARK-19471: AggregationIterator does not initialize the generated result projection before using it") - // Checks Spark specific codegen code for HashMap - .exclude("SPARK-43876: Enable fast hashmap for distinct queries") - .excludeByPrefix("SPARK-31620: agg with subquery (whole-stage-codegen =") enableSuite[GlutenSparkSessionBuilderSuite] // TODO: 4.x enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite] // 1 failure enableSuite[GlutenTPCDSCollationQueryTestSuite] diff --git a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala index ed89f43f115..3f576c9c1a5 100644 --- a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala +++ b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala @@ -16,161 +16,14 @@ */ package org.apache.spark.sql -import org.apache.gluten.execution.{HashAggregateExecTransformer, WholeStageTransformer} - -import org.apache.spark.sql.execution.aggregate.{ObjectHashAggregateExec, SortAggregateExec} -import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec -import org.apache.spark.sql.functions.{col, collect_list, expr, monotonically_increasing_id, rand, randn, spark_partition_id, sum} -import org.apache.spark.sql.internal.SQLConf - -import scala.util.Random - class GlutenSingleLevelAggregateHashMapSuite extends SingleLevelAggregateHashMapSuite - with GlutenSQLTestsTrait { - import testImplicits._ - - private def assertNoExceptions(wholeStageBoundary: Boolean, c: Column): Unit = { - for ( - (wholeStage, useObjectHashAgg) <- - Seq((true, true), (true, false), (false, true), (false, false)) - ) { - withSQLConf( - (SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key, wholeStage.toString), - (SQLConf.USE_OBJECT_HASH_AGG.key, useObjectHashAgg.toString)) { - - val df = Seq(("1", 1), ("1", 2), ("2", 3), ("2", 4)).toDF("x", "y") - - // test case for HashAggregate - // Converts to HashAggregateExecTransformer - val hashAggDF = df.groupBy("x").agg(c, sum("y")) - hashAggDF.collect() - val hashAggPlan = hashAggDF.queryExecution.executedPlan - assert(find(hashAggPlan) { - case WholeStageTransformer(_: HashAggregateExecTransformer, _) if wholeStageBoundary => - true - case _: HashAggregateExecTransformer if !wholeStageBoundary => true - case _ => false - }.isDefined) - - // test case for ObjectHashAggregate and SortAggregate - // Both gets converted to HashAggregateExecTransformer - val objHashAggOrSortAggDF = df.groupBy("x").agg(c, collect_list("y")) - objHashAggOrSortAggDF.collect() - val objHashAggOrSortAggPlan = - stripAQEPlan(objHashAggOrSortAggDF.queryExecution.executedPlan) - assert(find(objHashAggOrSortAggPlan) { - case WholeStageTransformer(_: HashAggregateExecTransformer, _) if wholeStageBoundary => - true - case _: HashAggregateExecTransformer if !wholeStageBoundary => true - case _ => false - }.isDefined) - } - } - } - - testGluten( - "SPARK-19471: AggregationIterator does not initialize the generated result projection" + - " before using it") { - Seq( - (true, monotonically_increasing_id()), - (false, spark_partition_id()), - (false, rand(Random.nextLong())), - (false, randn(Random.nextLong())) - ).foreach(tup => assertNoExceptions(tup._1, tup._2)) - } - - testGluten("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") { - withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") { - val df = Seq(("1", "2", 1), ("1", "2", 2), ("2", "3", 3), ("2", "3", 4)) - .toDF("a", "b", "c") - .repartition(col("a")) - - val objHashAggDF = df - .withColumn("d", expr("(a, b, c)")) - .groupBy("a", "b") - .agg(collect_list("d").as("e")) - .withColumn("f", expr("(b, e)")) - .groupBy("a") - .agg(collect_list("f").as("g")) - val aggPlan = objHashAggDF.queryExecution.executedPlan - - val sortAggPlans = collect(aggPlan) { case sortAgg: SortAggregateExec => sortAgg } - // SortAggregate will be retained due velox_collect_list - assert(sortAggPlans.size == 4) - - val objHashAggPlans = collect(aggPlan) { - case objHashAgg: ObjectHashAggregateExec => objHashAgg - } - assert(objHashAggPlans.isEmpty) - - val exchangePlans = collect(aggPlan) { case shuffle: ShuffleExchangeExec => shuffle } - assert(exchangePlans.length == 1) - } - } - - Seq(true, false).foreach { - value => - testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") { - withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> value.toString) { - withTempView("t1", "t2") { - sql("create temporary view t1 as select * from values (1, 2) as t1(a, b)") - sql("create temporary view t2 as select * from values (3, 4) as t2(c, d)") - - // test without grouping keys - checkAnswer( - sql("select sum(if(c > (select a from t1), d, 0)) as csum from t2"), - Row(4) :: Nil) - - // test with grouping keys - checkAnswer( - sql( - "select c, sum(if(c > (select a from t1), d, 0)) as csum from " + - "t2 group by c"), - Row(3, 4) :: Nil) - - // test with distinct - checkAnswer( - sql( - "select avg(distinct(d)), sum(distinct(if(c > (select a from t1)," + - " d, 0))) as csum from t2 group by c"), - Row(4, 4) :: Nil) - - // test subquery with agg - checkAnswer( - sql( - "select sum(distinct(if(c > (select sum(distinct(a)) from t1)," + - " d, 0))) as csum from t2 group by c"), - Row(4) :: Nil) - - // test SortAggregateExec - // Converts to HashAggregateExecTransformer - var df = sql("select max(if(c > (select a from t1), 'str1', 'str2')) as csum from t2") - df.collect() - assert( - find(df.queryExecution.executedPlan)( - _.isInstanceOf[HashAggregateExecTransformer]).isDefined) - checkAnswer(df, Row("str1") :: Nil) - - // test ObjectHashAggregateExec - // Converts to HashAggregateExecTransformer - df = - sql("select collect_list(d), sum(if(c > (select a from t1), d, 0)) as csum from t2") - df.collect() - assert( - find(df.queryExecution.executedPlan)( - _.isInstanceOf[HashAggregateExecTransformer]).isDefined) - checkAnswer(df, Row(Array(4), 4) :: Nil) - } - } - } - } -} + with GlutenTestsCommonTrait {} class GlutenTwoLevelAggregateHashMapSuite extends TwoLevelAggregateHashMapSuite - with GlutenSQLTestsTrait {} + with GlutenTestsCommonTrait {} class GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite extends TwoLevelAggregateHashMapWithVectorizedMapSuite - with GlutenSQLTestsTrait {} + with GlutenTestsCommonTrait {} From cd916fd33a148f88a0a564cd449b6d0e46cc6cc0 Mon Sep 17 00:00:00 2001 From: Kapil Singh Date: Sat, 7 Feb 2026 07:49:16 +0530 Subject: [PATCH 5/7] remove suites --- .../utils/velox/VeloxTestSettings.scala | 3 -- .../sql/GlutenAggregateHashMapSuite.scala | 29 ------------------- .../utils/velox/VeloxTestSettings.scala | 3 -- .../sql/GlutenAggregateHashMapSuite.scala | 29 ------------------- 4 files changed, 64 deletions(-) delete mode 100644 gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala delete mode 100644 gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala 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 fcd150e599d..5067e322e7a 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 @@ -770,7 +770,6 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenSSBQuerySuite] enableSuite[GlutenSessionStateSuite] // TODO: 4.x enableSuite[GlutenSetCommandSuite] // 1 failure - enableSuite[GlutenSingleLevelAggregateHashMapSuite] enableSuite[GlutenSparkSessionBuilderSuite] // TODO: 4.x enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite] // 1 failure enableSuite[GlutenTPCDSCollationQueryTestSuite] @@ -786,8 +785,6 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenTPCDSV2_7_PlanStabilityWithStatsSuite] enableSuite[GlutenTPCHPlanStabilitySuite] enableSuite[GlutenTPCHQuerySuite] - enableSuite[GlutenTwoLevelAggregateHashMapSuite] - enableSuite[GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite] enableSuite[GlutenUDFSuite] enableSuite[GlutenUDTRegistrationSuite] enableSuite[GlutenUnsafeRowSuite] diff --git a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala deleted file mode 100644 index 3f576c9c1a5..00000000000 --- a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala +++ /dev/null @@ -1,29 +0,0 @@ -/* - * 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 - -class GlutenSingleLevelAggregateHashMapSuite - extends SingleLevelAggregateHashMapSuite - with GlutenTestsCommonTrait {} - -class GlutenTwoLevelAggregateHashMapSuite - extends TwoLevelAggregateHashMapSuite - with GlutenTestsCommonTrait {} - -class GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite - extends TwoLevelAggregateHashMapWithVectorizedMapSuite - with GlutenTestsCommonTrait {} 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 87b78fddc99..c74aecccc75 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 @@ -735,7 +735,6 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenSSBQuerySuite] enableSuite[GlutenSessionStateSuite] // TODO: 4.x enableSuite[GlutenSetCommandSuite] // 1 failure - enableSuite[GlutenSingleLevelAggregateHashMapSuite] enableSuite[GlutenSparkSessionBuilderSuite] // TODO: 4.x enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite] // 1 failure enableSuite[GlutenTPCDSCollationQueryTestSuite] @@ -751,8 +750,6 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenTPCDSV2_7_PlanStabilityWithStatsSuite] enableSuite[GlutenTPCHPlanStabilitySuite] enableSuite[GlutenTPCHQuerySuite] - enableSuite[GlutenTwoLevelAggregateHashMapSuite] - enableSuite[GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite] enableSuite[GlutenUDFSuite] enableSuite[GlutenUDTRegistrationSuite] enableSuite[GlutenUnsafeRowSuite] diff --git a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala deleted file mode 100644 index 3f576c9c1a5..00000000000 --- a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenAggregateHashMapSuite.scala +++ /dev/null @@ -1,29 +0,0 @@ -/* - * 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 - -class GlutenSingleLevelAggregateHashMapSuite - extends SingleLevelAggregateHashMapSuite - with GlutenTestsCommonTrait {} - -class GlutenTwoLevelAggregateHashMapSuite - extends TwoLevelAggregateHashMapSuite - with GlutenTestsCommonTrait {} - -class GlutenTwoLevelAggregateHashMapWithVectorizedMapSuite - extends TwoLevelAggregateHashMapWithVectorizedMapSuite - with GlutenTestsCommonTrait {} From aba76b2e2ad9c5813866a3f60e93aae7a27a0a12 Mon Sep 17 00:00:00 2001 From: Kapil Singh Date: Sat, 7 Feb 2026 08:06:51 +0530 Subject: [PATCH 6/7] update dataframeaggregationsuite --- .../utils/velox/VeloxTestSettings.scala | 2 +- .../sql/GlutenDataFrameAggregateSuite.scala | 124 +++++++++++++++--- .../utils/velox/VeloxTestSettings.scala | 2 +- .../sql/GlutenDataFrameAggregateSuite.scala | 123 ++++++++++++++--- 4 files changed, 214 insertions(+), 37 deletions(-) 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 5067e322e7a..01db2518524 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 @@ -829,7 +829,7 @@ class VeloxTestSettings extends BackendTestSettings { "SPARK-19471: AggregationIterator does not initialize the generated result projection" + " before using it", // Velox's collect_list / collect_set are by design declarative aggregate so plan check - // for ObjectHashAggregateExec will fail. + // for ObjectHashAggregateExec will fail. Overridden "SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle", "SPARK-31620: agg with subquery (whole-stage-codegen = true)", "SPARK-31620: agg with subquery (whole-stage-codegen = false)" diff --git a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala index 2f3777caa17..7e764631294 100644 --- a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala +++ b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala @@ -17,10 +17,11 @@ package org.apache.spark.sql import org.apache.gluten.config.GlutenConfig -import org.apache.gluten.execution.HashAggregateExecBaseTransformer +import org.apache.gluten.execution.{HashAggregateExecBaseTransformer, HashAggregateExecTransformer} import org.apache.spark.sql.execution.WholeStageCodegenExec -import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, SortAggregateExec} +import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec} +import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec import org.apache.spark.sql.expressions.Aggregator import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf @@ -187,23 +188,24 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS // This test is applicable to velox backend. For CH backend, the replacement is disabled. testGluten("use gluten hash agg to replace vanilla spark sort agg") { + withTempView("t1") { + withSQLConf((GlutenConfig.COLUMNAR_FORCE_HASHAGG_ENABLED.key, "false")) { + Seq("A", "B", "C", "D").toDF("col1").createOrReplaceTempView("t1") + // SortAggregateExec is expected to be used for string type input. + val df = spark.sql("select max(col1) from t1") + checkAnswer(df, Row("D") :: Nil) + assert(find(df.queryExecution.executedPlan)(_.isInstanceOf[SortAggregateExec]).isDefined) + } - withSQLConf((GlutenConfig.COLUMNAR_FORCE_HASHAGG_ENABLED.key, "false")) { - Seq("A", "B", "C", "D").toDF("col1").createOrReplaceTempView("t1") - // SortAggregateExec is expected to be used for string type input. - val df = spark.sql("select max(col1) from t1") - checkAnswer(df, Row("D") :: Nil) - assert(find(df.queryExecution.executedPlan)(_.isInstanceOf[SortAggregateExec]).isDefined) - } - - withSQLConf((GlutenConfig.COLUMNAR_FORCE_HASHAGG_ENABLED.key, "true")) { - Seq("A", "B", "C", "D").toDF("col1").createOrReplaceTempView("t1") - val df = spark.sql("select max(col1) from t1") - checkAnswer(df, Row("D") :: Nil) - // Sort agg is expected to be replaced by gluten's hash agg. - assert( - find(df.queryExecution.executedPlan)( - _.isInstanceOf[HashAggregateExecBaseTransformer]).isDefined) + withSQLConf((GlutenConfig.COLUMNAR_FORCE_HASHAGG_ENABLED.key, "true")) { + Seq("A", "B", "C", "D").toDF("col1").createOrReplaceTempView("t1") + val df = spark.sql("select max(col1) from t1") + checkAnswer(df, Row("D") :: Nil) + // Sort agg is expected to be replaced by gluten's hash agg. + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[HashAggregateExecBaseTransformer]).isDefined) + } } } @@ -280,4 +282,90 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS randn(Random.nextLong()) ).foreach(assertNoExceptions) } + + testGluten("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") { + withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") { + val df = Seq(("1", "2", 1), ("1", "2", 2), ("2", "3", 3), ("2", "3", 4)) + .toDF("a", "b", "c") + .repartition(col("a")) + + val objHashAggDF = df + .withColumn("d", expr("(a, b, c)")) + .groupBy("a", "b") + .agg(collect_list("d").as("e")) + .withColumn("f", expr("(b, e)")) + .groupBy("a") + .agg(collect_list("f").as("g")) + val aggPlan = objHashAggDF.queryExecution.executedPlan + + val sortAggPlans = collect(aggPlan) { case sortAgg: SortAggregateExec => sortAgg } + // SortAggregate will be retained due velox_collect_list + assert(sortAggPlans.size == 4) + + val objHashAggPlans = collect(aggPlan) { + case objHashAgg: ObjectHashAggregateExec => objHashAgg + } + assert(objHashAggPlans.isEmpty) + + val exchangePlans = collect(aggPlan) { case shuffle: ShuffleExchangeExec => shuffle } + assert(exchangePlans.length == 1) + } + } + + Seq(true, false).foreach { + value => + testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") { + withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> value.toString) { + withTempView("t1", "t2") { + sql("create temporary view t1 as select * from values (1, 2) as t1(a, b)") + sql("create temporary view t2 as select * from values (3, 4) as t2(c, d)") + + // test without grouping keys + checkAnswer( + sql("select sum(if(c > (select a from t1), d, 0)) as csum from t2"), + Row(4) :: Nil) + + // test with grouping keys + checkAnswer( + sql( + "select c, sum(if(c > (select a from t1), d, 0)) as csum from " + + "t2 group by c"), + Row(3, 4) :: Nil) + + // test with distinct + checkAnswer( + sql( + "select avg(distinct(d)), sum(distinct(if(c > (select a from t1)," + + " d, 0))) as csum from t2 group by c"), + Row(4, 4) :: Nil) + + // test subquery with agg + checkAnswer( + sql( + "select sum(distinct(if(c > (select sum(distinct(a)) from t1)," + + " d, 0))) as csum from t2 group by c"), + Row(4) :: Nil) + + // test SortAggregateExec + // Converts to HashAggregateExecTransformer + var df = sql("select max(if(c > (select a from t1), 'str1', 'str2')) as csum from t2") + df.collect() + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[HashAggregateExecTransformer]).isDefined) + checkAnswer(df, Row("str1") :: Nil) + + // test ObjectHashAggregateExec + // Converts to HashAggregateExecTransformer + df = + sql("select collect_list(d), sum(if(c > (select a from t1), d, 0)) as csum from t2") + df.collect() + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[HashAggregateExecTransformer]).isDefined) + checkAnswer(df, Row(Array(4), 4) :: Nil) + } + } + } + } } 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 c74aecccc75..df4f5766f69 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 @@ -798,7 +798,7 @@ class VeloxTestSettings extends BackendTestSettings { "SPARK-19471: AggregationIterator does not initialize the generated result projection" + " before using it", // Velox's collect_list / collect_set are by design declarative aggregate so plan check - // for ObjectHashAggregateExec will fail. + // for ObjectHashAggregateExec will fail. Overriden "SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle", "SPARK-31620: agg with subquery (whole-stage-codegen = true)", "SPARK-31620: agg with subquery (whole-stage-codegen = false)" diff --git a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala index 2f3777caa17..a0338f10bc9 100644 --- a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala +++ b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala @@ -17,10 +17,11 @@ package org.apache.spark.sql import org.apache.gluten.config.GlutenConfig -import org.apache.gluten.execution.HashAggregateExecBaseTransformer +import org.apache.gluten.execution.{HashAggregateExecBaseTransformer, HashAggregateExecTransformer} import org.apache.spark.sql.execution.WholeStageCodegenExec -import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, SortAggregateExec} +import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec} +import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec import org.apache.spark.sql.expressions.Aggregator import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf @@ -188,22 +189,24 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS // This test is applicable to velox backend. For CH backend, the replacement is disabled. testGluten("use gluten hash agg to replace vanilla spark sort agg") { - withSQLConf((GlutenConfig.COLUMNAR_FORCE_HASHAGG_ENABLED.key, "false")) { - Seq("A", "B", "C", "D").toDF("col1").createOrReplaceTempView("t1") - // SortAggregateExec is expected to be used for string type input. - val df = spark.sql("select max(col1) from t1") - checkAnswer(df, Row("D") :: Nil) - assert(find(df.queryExecution.executedPlan)(_.isInstanceOf[SortAggregateExec]).isDefined) - } + withTempView("t1") { + withSQLConf((GlutenConfig.COLUMNAR_FORCE_HASHAGG_ENABLED.key, "false")) { + Seq("A", "B", "C", "D").toDF("col1").createOrReplaceTempView("t1") + // SortAggregateExec is expected to be used for string type input. + val df = spark.sql("select max(col1) from t1") + checkAnswer(df, Row("D") :: Nil) + assert(find(df.queryExecution.executedPlan)(_.isInstanceOf[SortAggregateExec]).isDefined) + } - withSQLConf((GlutenConfig.COLUMNAR_FORCE_HASHAGG_ENABLED.key, "true")) { - Seq("A", "B", "C", "D").toDF("col1").createOrReplaceTempView("t1") - val df = spark.sql("select max(col1) from t1") - checkAnswer(df, Row("D") :: Nil) - // Sort agg is expected to be replaced by gluten's hash agg. - assert( - find(df.queryExecution.executedPlan)( - _.isInstanceOf[HashAggregateExecBaseTransformer]).isDefined) + withSQLConf((GlutenConfig.COLUMNAR_FORCE_HASHAGG_ENABLED.key, "true")) { + Seq("A", "B", "C", "D").toDF("col1").createOrReplaceTempView("t1") + val df = spark.sql("select max(col1) from t1") + checkAnswer(df, Row("D") :: Nil) + // Sort agg is expected to be replaced by gluten's hash agg. + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[HashAggregateExecBaseTransformer]).isDefined) + } } } @@ -280,4 +283,90 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS randn(Random.nextLong()) ).foreach(assertNoExceptions) } + + testGluten("SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle") { + withSQLConf(SQLConf.USE_OBJECT_HASH_AGG.key -> "true") { + val df = Seq(("1", "2", 1), ("1", "2", 2), ("2", "3", 3), ("2", "3", 4)) + .toDF("a", "b", "c") + .repartition(col("a")) + + val objHashAggDF = df + .withColumn("d", expr("(a, b, c)")) + .groupBy("a", "b") + .agg(collect_list("d").as("e")) + .withColumn("f", expr("(b, e)")) + .groupBy("a") + .agg(collect_list("f").as("g")) + val aggPlan = objHashAggDF.queryExecution.executedPlan + + val sortAggPlans = collect(aggPlan) { case sortAgg: SortAggregateExec => sortAgg } + // SortAggregate will be retained due velox_collect_list + assert(sortAggPlans.size == 4) + + val objHashAggPlans = collect(aggPlan) { + case objHashAgg: ObjectHashAggregateExec => objHashAgg + } + assert(objHashAggPlans.isEmpty) + + val exchangePlans = collect(aggPlan) { case shuffle: ShuffleExchangeExec => shuffle } + assert(exchangePlans.length == 1) + } + } + + Seq(true, false).foreach { + value => + testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") { + withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> value.toString) { + withTempView("t1", "t2") { + sql("create temporary view t1 as select * from values (1, 2) as t1(a, b)") + sql("create temporary view t2 as select * from values (3, 4) as t2(c, d)") + + // test without grouping keys + checkAnswer( + sql("select sum(if(c > (select a from t1), d, 0)) as csum from t2"), + Row(4) :: Nil) + + // test with grouping keys + checkAnswer( + sql( + "select c, sum(if(c > (select a from t1), d, 0)) as csum from " + + "t2 group by c"), + Row(3, 4) :: Nil) + + // test with distinct + checkAnswer( + sql( + "select avg(distinct(d)), sum(distinct(if(c > (select a from t1)," + + " d, 0))) as csum from t2 group by c"), + Row(4, 4) :: Nil) + + // test subquery with agg + checkAnswer( + sql( + "select sum(distinct(if(c > (select sum(distinct(a)) from t1)," + + " d, 0))) as csum from t2 group by c"), + Row(4) :: Nil) + + // test SortAggregateExec + // Converts to HashAggregateExecTransformer + var df = sql("select max(if(c > (select a from t1), 'str1', 'str2')) as csum from t2") + df.collect() + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[HashAggregateExecTransformer]).isDefined) + checkAnswer(df, Row("str1") :: Nil) + + // test ObjectHashAggregateExec + // Converts to HashAggregateExecTransformer + df = + sql("select collect_list(d), sum(if(c > (select a from t1), d, 0)) as csum from t2") + df.collect() + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[HashAggregateExecTransformer]).isDefined) + checkAnswer(df, Row(Array(4), 4) :: Nil) + } + } + } + } } From fb28bc8a23834270d30111c16e62d3150b5f4bdb Mon Sep 17 00:00:00 2001 From: Kapil Singh Date: Tue, 10 Feb 2026 21:39:56 +0530 Subject: [PATCH 7/7] Add docs --- gluten-ut/excluded-spark-uts.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 gluten-ut/excluded-spark-uts.md diff --git a/gluten-ut/excluded-spark-uts.md b/gluten-ut/excluded-spark-uts.md new file mode 100644 index 00000000000..598eca64932 --- /dev/null +++ b/gluten-ut/excluded-spark-uts.md @@ -0,0 +1,6 @@ +# List of excluded Spark UTs +## This doc. contains list of Spark UTs which are chosen not run with Gluten, because they checks for Spark specific behavior. + +| Suites | Versions | Reason to exclude | +|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| org.apache.spark.sql.SingleLevelAggregateHashMapSuite
org.apache.spark.sql.TwoLevelAggregateHashMapSuite
org.apache.spark.sql.TwoLevelAggregateHashMapWithVectorizedMapSuite | 4.0, 4.1 | This UTs is similar to `DataFrameAggregateSuite`.
The only difference being that it contains variation for Spark codegen enabled and disabled and enabling single level or two level aggregate hash-maps which is also specific to Spark Aggregate implementation.
We already run `GlutenDataFrameAggregateSuite` for `DataFrameAggregateSuite` so these additional suites doesn't need to run | \ No newline at end of file