Skip to content

Commit 37d6b3c

Browse files
linhongliu-dbcloud-fan
authored andcommitted
[SPARK-32761][SQL][3.0] Allow aggregating multiple foldable distinct expressions
### What changes were proposed in this pull request? For queries with multiple foldable distinct columns, since they will be eliminated during execution, it's not mandatory to let `RewriteDistinctAggregates` handle this case. And in the current code, `RewriteDistinctAggregates` *dose* miss some "aggregating with multiple foldable distinct expressions" cases. For example: `select count(distinct 2), count(distinct 2, 3)` will be missed. But in the planner, this will trigger an error that "multiple distinct expressions" are not allowed. As the foldable distinct columns can be eliminated finally, we can allow this in the aggregation planner check. ### Why are the changes needed? bug fix ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? added test case Authored-by: Linhong Liu <linhong.liudatabricks.com> Signed-off-by: Wenchen Fan <wenchendatabricks.com> (cherry picked from commit a410658) Closes #30052 from linhongliu-db/SPARK-32761-3.0. Authored-by: Linhong Liu <linhong.liu@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
1 parent 160f458 commit 37d6b3c

2 files changed

Lines changed: 8 additions & 2 deletions

File tree

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -517,7 +517,8 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
517517

518518
val (functionsWithDistinct, functionsWithoutDistinct) =
519519
aggregateExpressions.partition(_.isDistinct)
520-
if (functionsWithDistinct.map(_.aggregateFunction.children.toSet).distinct.length > 1) {
520+
if (functionsWithDistinct.map(
521+
_.aggregateFunction.children.filterNot(_.foldable).toSet).distinct.length > 1) {
521522
// This is a sanity check. We should not reach here when we have multiple distinct
522523
// column sets. Our `RewriteDistinctAggregates` should take care this case.
523524
sys.error("You hit a query analyzer bug. Please report your query to " +
@@ -548,7 +549,8 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
548549
// to be [COUNT(DISTINCT foo), MAX(DISTINCT foo)], but
549550
// [COUNT(DISTINCT bar), COUNT(DISTINCT foo)] is disallowed because those two distinct
550551
// aggregates have different column expressions.
551-
val distinctExpressions = functionsWithDistinct.head.aggregateFunction.children
552+
val distinctExpressions =
553+
functionsWithDistinct.head.aggregateFunction.children.filterNot(_.foldable)
552554
val normalizedNamedDistinctExpressions = distinctExpressions.map { e =>
553555
// Ideally this should be done in `NormalizeFloatingNumbers`, but we do it here
554556
// because `distinctExpressions` is not extracted during logical phase.

sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2467,6 +2467,10 @@ class DataFrameSuite extends QueryTest
24672467
val df = l.join(r, $"col2" === $"col4", "LeftOuter")
24682468
checkAnswer(df, Row("2", "2"))
24692469
}
2470+
2471+
test("SPARK-32761: aggregating multiple distinct CONSTANT columns") {
2472+
checkAnswer(sql("select count(distinct 2), count(distinct 2,3)"), Row(1, 1))
2473+
}
24702474
}
24712475

24722476
case class GroupByKey(a: Int, b: Int)

0 commit comments

Comments
 (0)