This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new aed551631307 [SPARK-53155][SQL] Global lower agggregation should not be replaced with a project
aed551631307 is described below
commit aed55163130743e52d694f1e22a6d7bc83abd8cd
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Thu Aug 7 08:11:51 2025 -0700
[SPARK-53155][SQL] Global lower agggregation should not be replaced with a project
This patch fixes the optimization rule `RemoveRedundantAggregates`.
The optimizer rule `RemoveRedundantAggregates` removes redundant lower aggregation from a query plan and replace it with a project of referred non-aggregate expressions. However, if the removed aggregation is a global one, that is not correct because a project is different with a global aggregation in semantics.
For example, if the input relation is empty, a project might be optimized to an empty relation, while a global aggregation will return a single row.
Yes, this fixes a user-facing bug. Previously, a global aggregation under another aggregation might be treated as redundant and replaced as a project with non-aggregation expressions. If the input relation is empty, the replacement is incorrect and might produce incorrect result. This patch adds a new unit test to show the difference.
Unit test, manual test.
No
Closes #51884 from viirya/fix_remove_redundant_agg.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
(cherry picked from commit 3aa8c9dbc1c0d4622cd62a65db510d6feac31ba3)
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
.../optimizer/RemoveRedundantAggregates.scala | 8 +++++++-
.../optimizer/RemoveRedundantAggregatesSuite.scala | 21 ++++++++++++++++++++-
.../apache/spark/sql/DataFrameAggregateSuite.scala | 7 +++++++
3 files changed, 34 insertions(+), 2 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RemoveRedundantAggregates.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RemoveRedundantAggregates.scala
index d6a4bd030c9d..b4602d0ddcc9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RemoveRedundantAggregates.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RemoveRedundantAggregates.scala
@@ -54,7 +54,13 @@ object RemoveRedundantAggregates extends Rule[LogicalPlan] with AliasHelper {
.map(_.toAttribute)
))
- upperHasNoDuplicateSensitiveAgg && upperRefsOnlyDeterministicNonAgg
+ // If the lower aggregation is global, it is not redundant because a project with
+ // non-aggregate expressions is different with global aggregation in semantics.
+ // E.g., if the input relation is empty, a project might be optimized to an empty
+ // relation, while a global aggregation will return a single row.
+ lazy val lowerIsGlobalAgg = lower.groupingExpressions.isEmpty
+
+ upperHasNoDuplicateSensitiveAgg && upperRefsOnlyDeterministicNonAgg && !lowerIsGlobalAgg
}
private def isDuplicateSensitive(ae: AggregateExpression): Boolean = {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RemoveRedundantAggregatesSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RemoveRedundantAggregatesSuite.scala
index 2af3057c0b85..40b3d36d4bfc 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RemoveRedundantAggregatesSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RemoveRedundantAggregatesSuite.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.catalyst.optimizer
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
-import org.apache.spark.sql.catalyst.expressions.{Expression, PythonUDAF}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Literal, PythonUDAF}
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi, PlanTest}
import org.apache.spark.sql.catalyst.plans.logical.{Distinct, LocalRelation, LogicalPlan}
@@ -289,4 +289,23 @@ class RemoveRedundantAggregatesSuite extends PlanTest {
val originalQuery = Distinct(x.groupBy($"a", $"b")($"a", TrueLiteral)).analyze
comparePlans(Optimize.execute(originalQuery), originalQuery)
}
+
+ test("SPARK-53155: global lower aggregation should not be removed") {
+ object OptimizeNonRemovedRedundantAgg extends RuleExecutor[LogicalPlan] {
+ val batches = Batch("RemoveRedundantAggregates", FixedPoint(10),
+ PropagateEmptyRelation,
+ RemoveRedundantAggregates) :: Nil
+ }
+
+ val query = relation
+ .groupBy()(Literal(1).as("col1"), Literal(2).as("col2"), Literal(3).as("col3"))
+ .groupBy($"col1")(max($"col1"))
+ .analyze
+ val expected = relation
+ .groupBy()(Literal(1).as("col1"), Literal(2).as("col2"), Literal(3).as("col3"))
+ .groupBy($"col1")(max($"col1"))
+ .analyze
+ val optimized = OptimizeNonRemovedRedundantAgg.execute(query)
+ comparePlans(optimized, expected)
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
index 6ce0a657d5b9..9db406ff12fe 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -2567,6 +2567,13 @@ class DataFrameAggregateSuite extends QueryTest
checkAnswer(df, Row(1.001d, 1, 1) :: Row(6.002d, 1, 1) :: Nil)
}
}
+
+ test("SPARK-53155: global lower aggregation should not be removed") {
+ val df = emptyTestData
+ .groupBy().agg(lit(1).as("col1"), lit(2).as("col2"), lit(3).as("col3"))
+ .groupBy($"col1").agg(max("col1"))
+ checkAnswer(df, Seq(Row(1, 1)))
+ }
}
case class B(c: Option[Double])
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org