[SPARK-32777][SQL] Aggregation support aggregate function with multiple foldable expressions. #29626

beliefer · 2020-09-02T07:37:19Z

What changes were proposed in this pull request?

Spark SQL exists a bug show below:

spark.sql(
  " SELECT COUNT(DISTINCT 2), COUNT(DISTINCT 2, 3)")
  .show()
+-----------------+--------------------+
|count(DISTINCT 2)|count(DISTINCT 2, 3)|
+-----------------+--------------------+
|                1|                   1|
+-----------------+--------------------+

spark.sql(
  " SELECT COUNT(DISTINCT 2), COUNT(DISTINCT 3, 2)")
  .show()
+-----------------+--------------------+
|count(DISTINCT 2)|count(DISTINCT 3, 2)|
+-----------------+--------------------+
|                1|                   0|
+-----------------+--------------------+

The first query is correct, but the second query is not.
The root reason is the second query rewrited by RewriteDistinctAggregates who expand the output but lost the 2.

Why are the changes needed?

Fix a bug.
SELECT COUNT(DISTINCT 2), COUNT(DISTINCT 3, 2) should return 1, 1

Does this PR introduce any user-facing change?

Yes

How was this patch tested?

New UT

…ions

SparkQA · 2020-09-02T07:55:31Z

Test build #128196 has finished for PR 29626 at commit 0f355ad.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

beliefer · 2020-09-02T07:59:55Z

retest this please

HyukjinKwon · 2020-09-02T08:11:26Z

cc @cloud-fan and @linhongliu-db

cloud-fan · 2020-09-02T08:43:59Z

sql/core/src/test/resources/sql-tests/results/group-by-filter.sql.out

+
+
+-- !query
+SELECT COUNT(DISTINCT id), COUNT(DISTINCT 2,3) FILTER (WHERE dept_id = 30) FROM emp


can we test WHERE dept_id > 0? to make sure that we get correct result even if there are more than one inputs after filter.

SparkQA · 2020-09-02T08:59:24Z

Test build #128198 has finished for PR 29626 at commit e15519e.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class JDBCScan(
case class JDBCScanBuilder(
case class JDBCWriteBuilder(schema: StructType, options: JdbcOptionsInWrite) extends V1WriteBuilder

SparkQA · 2020-09-02T09:18:53Z

Test build #128197 has finished for PR 29626 at commit e15519e.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class JDBCScan(
case class JDBCScanBuilder(
case class JDBCWriteBuilder(schema: StructType, options: JdbcOptionsInWrite) extends V1WriteBuilder

SparkQA · 2020-09-02T12:14:13Z

Test build #128201 has finished for PR 29626 at commit 9ad3469.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-09-02T23:07:24Z

Test build #128208 has finished for PR 29626 at commit 81eb9e4.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-09-03T07:05:02Z

Test build #128219 has finished for PR 29626 at commit 52896fb.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

beliefer · 2020-09-03T07:42:12Z

retest this please

beliefer · 2020-09-03T14:58:57Z

cc @cloud-fan

SparkQA · 2020-09-04T12:19:54Z

Test build #128285 has finished for PR 29626 at commit 8622411.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-09-04T15:31:52Z

Test build #128294 has finished for PR 29626 at commit 43cb6a0.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-09-07T05:02:30Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala

+      }
+    }
+    val distinctAggGroupLookup = distinctAggGroupMap.toMap
+    val distinctAggGroups = distinctAggGroupMap.groupBy(_._2).map{ kv =>


nit: mapValues

cloud-fan · 2020-09-07T05:04:12Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala

            val naf = patchAggregateFunctionChildren(af) { x =>
-              val condition = if (e.filter.isDefined) {
-                e.filter.map(distinctAggFilterAttrLookup.get(_)).get
+              val condition = e.filter.map(distinctAggFilterAttrLookup.get(_)).getOrElse(None)


nit: is .getOrElse(None) needed?

SparkQA · 2020-09-07T07:05:02Z

Test build #128341 has finished for PR 29626 at commit 64834f8.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-09-07T11:42:55Z

Test build #128345 has finished for PR 29626 at commit fa7d578.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-09-08T14:36:51Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala

-              val condition = if (e.filter.isDefined) {
-                e.filter.map(distinctAggFilterAttrLookup.get(_)).get
+              val condition = e.filter.map(distinctAggFilterAttrLookup.get(_)).flatten
+              if (distinctAggGroupLookup(e).contains(x)) {


I'm wondering if we can simplify the logic a bit. The goal is to only do the replacement for the first child if all the children are foldable. How about

val af = e.aggregateFunction val condition = e.filter.map(distinctAggFilterAttrLookup.get(_)).flatten val naf = if (af.children.forall(_.foldable)) { val firstChild = evalWithinGroup(id, af.children.head, condition) af.withNewChildren(firstChild +: af.children.drop(1)).asInstanceOf[AggregateFunction] } else { patchAggregateFunctionChildren(af) { x => distinctAggChildAttrLookup.get(x).map(evalWithinGroup(id, _, condition)) } }

Then we don't need to change a lot of code to create distinctAggGroupLookup

SparkQA · 2020-09-09T07:05:02Z

Test build #128429 has finished for PR 29626 at commit 9a44bec.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

beliefer · 2020-09-09T07:08:44Z

retest this please

SparkQA · 2020-09-09T11:17:28Z

Test build #128438 has finished for PR 29626 at commit 9a44bec.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

beliefer · 2020-09-10T01:55:30Z

retest this please

cloud-fan · 2020-09-10T02:10:29Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala

-                None
+            val condition = e.filter.map(distinctAggFilterAttrLookup.get(_)).flatten
+            val naf = if (af.children.forall(_.foldable)) {
+              val firstChild = evalWithinGroup(id, af.children.head, condition)


can we add some comments to explain why we are doing it?

SparkQA · 2020-09-10T05:51:03Z

Test build #128479 has finished for PR 29626 at commit 16588f5.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

beliefer · 2020-09-10T05:54:30Z

retest this please

SparkQA · 2020-09-10T07:05:02Z

Test build #128478 has finished for PR 29626 at commit 9a44bec.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-09-10T07:05:03Z

Test build #128489 has finished for PR 29626 at commit 16588f5.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

beliefer · 2020-09-10T07:12:33Z

retest this please

SparkQA · 2020-09-10T11:17:31Z

Test build #128498 has finished for PR 29626 at commit 16588f5.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-09-10T11:25:30Z

github action passed, merging to master, thanks!

beliefer · 2020-09-11T01:56:35Z

@cloud-fan Thanks

beliefer and others added 14 commits June 19, 2020 10:36

Reuse completeNextStageWithFetchFailure

4a6f903

Merge remote-tracking branch 'upstream/master'

96456e2

Merge remote-tracking branch 'upstream/master'

4314005

Merge remote-tracking branch 'upstream/master'

d6af4a7

Merge remote-tracking branch 'upstream/master'

f69094f

Merge remote-tracking branch 'upstream/master'

b86a42d

Merge branch 'master' of github.com:beliefer/spark

2ac5159

Merge remote-tracking branch 'upstream/master'

9021d6c

Merge branch 'master' of github.com:beliefer/spark

74a2ef4

Merge remote-tracking branch 'upstream/master'

9828158

Merge remote-tracking branch 'upstream/master'

9cd1aaf

Merge remote-tracking branch 'upstream/master'

abfcbb9

Merge remote-tracking branch 'upstream/master'

07c6c81

Support multiple foldable distinct expressions.

0f355ad

probot-autolabeler bot added the SQL label Sep 2, 2020

beliefer added 2 commits September 2, 2020 15:48

Merge remote-tracking branch 'upstream/master'

580130b

Merge branch 'master' into support-multiple-foldable-distinct-express…

e15519e

…ions

cloud-fan reviewed Sep 2, 2020

View reviewed changes

Update test case.

9ad3469

Fix bug.

81eb9e4

Update golden files.

52896fb

beliefer changed the title ~~[WIP][SPARK-32777][SQL] Aggregation support aggregate function with multiple foldable expressions.~~ [SPARK-32777][SQL] Aggregation support aggregate function with multiple foldable expressions. Sep 3, 2020

beliefer added 2 commits September 4, 2020 15:17

Fix bug

8622411

Optimize code

43cb6a0

cloud-fan reviewed Sep 7, 2020

View reviewed changes

Optimize code

64834f8

Optimize code

fa7d578

cloud-fan reviewed Sep 8, 2020

View reviewed changes

beliefer added 2 commits September 9, 2020 10:47

Optimize code.

5f14d54

Optimize code.

9a44bec

cloud-fan reviewed Sep 10, 2020

View reviewed changes

Add comments.

16588f5

cloud-fan closed this in a22871f Sep 10, 2020



		-- !query
		SELECT COUNT(DISTINCT id), COUNT(DISTINCT 2,3) FILTER (WHERE dept_id = 30) FROM emp

[SPARK-32777][SQL] Aggregation support aggregate function with multiple foldable expressions. #29626

[SPARK-32777][SQL] Aggregation support aggregate function with multiple foldable expressions. #29626

Uh oh!

Conversation

beliefer commented Sep 2, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

SparkQA commented Sep 2, 2020

Uh oh!

beliefer commented Sep 2, 2020

Uh oh!

HyukjinKwon commented Sep 2, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 2, 2020

Uh oh!

SparkQA commented Sep 2, 2020

Uh oh!

SparkQA commented Sep 2, 2020

Uh oh!

SparkQA commented Sep 2, 2020

Uh oh!

SparkQA commented Sep 3, 2020

Uh oh!

beliefer commented Sep 3, 2020

Uh oh!

beliefer commented Sep 3, 2020

Uh oh!

SparkQA commented Sep 4, 2020

Uh oh!

SparkQA commented Sep 4, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 7, 2020

Uh oh!

SparkQA commented Sep 7, 2020

Uh oh!

cloud-fan Sep 8, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 9, 2020

Uh oh!

beliefer commented Sep 9, 2020

Uh oh!

SparkQA commented Sep 9, 2020

Uh oh!

beliefer commented Sep 10, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 10, 2020

Uh oh!

beliefer commented Sep 10, 2020

Uh oh!

SparkQA commented Sep 10, 2020

Uh oh!

SparkQA commented Sep 10, 2020

Uh oh!

beliefer commented Sep 10, 2020

Uh oh!

SparkQA commented Sep 10, 2020

beliefer commented Sep 2, 2020 •

edited

Loading

cloud-fan Sep 8, 2020 •

edited

Loading