spark git commit: [SPARK-17616][SQL] Support a single distinct aggregate combined with a non-partial aggregate

2016-09-22 Thread hvanhovell 
Repository: spark
Updated Branches:
  refs/heads/branch-2.0 47fc0b9f4 -> 0a593db36


[SPARK-17616][SQL] Support a single distinct aggregate combined with a 
non-partial aggregate

We currently cannot execute an aggregate that contains a single distinct 
aggregate function and an one or more non-partially plannable aggregate 
functions, for example:
```sql
select   grp,
 collect_list(col1),
 count(distinct col2)
from tbl_a
group by 1
```
This is a regression from Spark 1.6. This is caused by the fact that the single 
distinct aggregation code path assumes that all aggregates can be planned in 
two phases (is partially aggregatable). This PR works around this issue by 
triggering the `RewriteDistinctAggregates` in such cases (this is similar to 
the approach taken in 1.6).

Created `RewriteDistinctAggregatesSuite` which checks if the aggregates with 
distinct aggregate functions get rewritten into two `Aggregates` and an 
`Expand`. Added a regression test to `DataFrameAggregateSuite`.

Author: Herman van Hovell 

Closes #15187 from hvanhovell/SPARK-17616.

(cherry picked from commit 0d634875026ccf1eaf984996e9460d7673561f80)
Signed-off-by: Herman van Hovell 


Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0a593db3
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0a593db3
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0a593db3

Branch: refs/heads/branch-2.0
Commit: 0a593db360b3b7771f45f482cf45e8500f0faa76
Parents: 47fc0b9
Author: Herman van Hovell 
Authored: Thu Sep 22 14:29:27 2016 -0700
Committer: Herman van Hovell 
Committed: Thu Sep 22 16:22:31 2016 -0700

--
 .../optimizer/RewriteDistinctAggregates.scala   | 18 ++--
 .../RewriteDistinctAggregatesSuite.scala| 94 
 .../spark/sql/DataFrameAggregateSuite.scala |  8 ++
 3 files changed, 111 insertions(+), 9 deletions(-)
--


http://git-wip-us.apache.org/repos/asf/spark/blob/0a593db3/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
--
diff --git 
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
 
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
index 0f43e7b..d6a39ec 100644
--- 
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
+++ 
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
@@ -119,14 +119,16 @@ object RewriteDistinctAggregates extends 
Rule[LogicalPlan] {
   .filter(_.isDistinct)
   .groupBy(_.aggregateFunction.children.toSet)
 
-// Aggregation strategy can handle the query with single distinct
-if (distinctAggGroups.size > 1) {
+// Check if the aggregates contains functions that do not support partial 
aggregation.
+val existsNonPartial = 
aggExpressions.exists(!_.aggregateFunction.supportsPartial)
+
+// Aggregation strategy can handle queries with a single distinct group 
and partial aggregates.
+if (distinctAggGroups.size > 1 || (distinctAggGroups.size == 1 && 
existsNonPartial)) {
   // Create the attributes for the grouping id and the group by clause.
-  val gid =
-new AttributeReference("gid", IntegerType, false)(isGenerated = true)
+  val gid = AttributeReference("gid", IntegerType, nullable = 
false)(isGenerated = true)
   val groupByMap = a.groupingExpressions.collect {
 case ne: NamedExpression => ne -> ne.toAttribute
-case e => e -> new AttributeReference(e.sql, e.dataType, e.nullable)()
+case e => e -> AttributeReference(e.sql, e.dataType, e.nullable)()
   }
   val groupByAttrs = groupByMap.map(_._2)
 
@@ -135,9 +137,7 @@ object RewriteDistinctAggregates extends Rule[LogicalPlan] {
   def patchAggregateFunctionChildren(
   af: AggregateFunction)(
   attrs: Expression => Expression): AggregateFunction = {
-af.withNewChildren(af.children.map {
-  case afc => attrs(afc)
-}).asInstanceOf[AggregateFunction]
+
af.withNewChildren(af.children.map(attrs)).asInstanceOf[AggregateFunction]
   }
 
   // Setup unique distinct aggregate children.
@@ -265,5 +265,5 @@ object RewriteDistinctAggregates extends Rule[LogicalPlan] {
 // NamedExpression. This is done to prevent collisions between distinct 
and regular aggregate
 // children, in this case attribute reuse causes the input of the regular 
aggregate to bound to
 // the (nulled out) input of the distinct aggregate.
-e -> new

spark git commit: [SPARK-17616][SQL] Support a single distinct aggregate combined with a non-partial aggregate

2016-09-22 Thread hvanhovell 
Repository: spark
Updated Branches:
  refs/heads/master 3cdae0ff2 -> 0d6348750


[SPARK-17616][SQL] Support a single distinct aggregate combined with a 
non-partial aggregate

## What changes were proposed in this pull request?
We currently cannot execute an aggregate that contains a single distinct 
aggregate function and an one or more non-partially plannable aggregate 
functions, for example:
```sql
select   grp,
 collect_list(col1),
 count(distinct col2)
from tbl_a
group by 1
```
This is a regression from Spark 1.6. This is caused by the fact that the single 
distinct aggregation code path assumes that all aggregates can be planned in 
two phases (is partially aggregatable). This PR works around this issue by 
triggering the `RewriteDistinctAggregates` in such cases (this is similar to 
the approach taken in 1.6).

## How was this patch tested?
Created `RewriteDistinctAggregatesSuite` which checks if the aggregates with 
distinct aggregate functions get rewritten into two `Aggregates` and an 
`Expand`. Added a regression test to `DataFrameAggregateSuite`.

Author: Herman van Hovell 

Closes #15187 from hvanhovell/SPARK-17616.


Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0d634875
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0d634875
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0d634875

Branch: refs/heads/master
Commit: 0d634875026ccf1eaf984996e9460d7673561f80
Parents: 3cdae0f
Author: Herman van Hovell 
Authored: Thu Sep 22 14:29:27 2016 -0700
Committer: Herman van Hovell 
Committed: Thu Sep 22 14:29:27 2016 -0700

--
 .../optimizer/RewriteDistinctAggregates.scala   | 18 ++--
 .../RewriteDistinctAggregatesSuite.scala| 94 
 .../spark/sql/DataFrameAggregateSuite.scala |  8 ++
 3 files changed, 111 insertions(+), 9 deletions(-)
--


http://git-wip-us.apache.org/repos/asf/spark/blob/0d634875/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
--
diff --git 
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
 
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
index 0f43e7b..d6a39ec 100644
--- 
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
+++ 
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala
@@ -119,14 +119,16 @@ object RewriteDistinctAggregates extends 
Rule[LogicalPlan] {
   .filter(_.isDistinct)
   .groupBy(_.aggregateFunction.children.toSet)
 
-// Aggregation strategy can handle the query with single distinct
-if (distinctAggGroups.size > 1) {
+// Check if the aggregates contains functions that do not support partial 
aggregation.
+val existsNonPartial = 
aggExpressions.exists(!_.aggregateFunction.supportsPartial)
+
+// Aggregation strategy can handle queries with a single distinct group 
and partial aggregates.
+if (distinctAggGroups.size > 1 || (distinctAggGroups.size == 1 && 
existsNonPartial)) {
   // Create the attributes for the grouping id and the group by clause.
-  val gid =
-new AttributeReference("gid", IntegerType, false)(isGenerated = true)
+  val gid = AttributeReference("gid", IntegerType, nullable = 
false)(isGenerated = true)
   val groupByMap = a.groupingExpressions.collect {
 case ne: NamedExpression => ne -> ne.toAttribute
-case e => e -> new AttributeReference(e.sql, e.dataType, e.nullable)()
+case e => e -> AttributeReference(e.sql, e.dataType, e.nullable)()
   }
   val groupByAttrs = groupByMap.map(_._2)
 
@@ -135,9 +137,7 @@ object RewriteDistinctAggregates extends Rule[LogicalPlan] {
   def patchAggregateFunctionChildren(
   af: AggregateFunction)(
   attrs: Expression => Expression): AggregateFunction = {
-af.withNewChildren(af.children.map {
-  case afc => attrs(afc)
-}).asInstanceOf[AggregateFunction]
+
af.withNewChildren(af.children.map(attrs)).asInstanceOf[AggregateFunction]
   }
 
   // Setup unique distinct aggregate children.
@@ -265,5 +265,5 @@ object RewriteDistinctAggregates extends Rule[LogicalPlan] {
 // NamedExpression. This is done to prevent collisions between distinct 
and regular aggregate
 // children, in this case attribute reuse causes the input of the regular 
aggregate to bound to
 // the (nulled out) input of the distinct aggregate.
-e -> new AttributeReference(e.sql, e.dataType, true)()
+e ->