You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by hv...@apache.org on 2016/09/22 21:29:31 UTC
spark git commit: [SPARK-17616][SQL] Support a single distinct
aggregate combined with a non-partial aggregate
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 <hv...@databricks.com>
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 <hv...@databricks.com>
Authored: Thu Sep 22 14:29:27 2016 -0700
Committer: Herman van Hovell <hv...@databricks.com>
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 -> AttributeReference(e.sql, e.dataType, nullable = true)()
}
http://git-wip-us.apache.org/repos/asf/spark/blob/0d634875/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala
----------------------------------------------------------------------
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala
new file mode 100644
index 0000000..0b973c3
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala
@@ -0,0 +1,94 @@
+/*
+ * 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.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.SimpleCatalystConf
+import org.apache.spark.sql.catalyst.analysis.{Analyzer, EmptyFunctionRegistry}
+import org.apache.spark.sql.catalyst.catalog.{InMemoryCatalog, SessionCatalog}
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions.{If, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectSet, Count}
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Expand, LocalRelation, LogicalPlan}
+import org.apache.spark.sql.types.{IntegerType, StringType}
+
+class RewriteDistinctAggregatesSuite extends PlanTest {
+ val conf = SimpleCatalystConf(caseSensitiveAnalysis = false, groupByOrdinal = false)
+ val catalog = new SessionCatalog(new InMemoryCatalog, EmptyFunctionRegistry, conf)
+ val analyzer = new Analyzer(catalog, conf)
+
+ val nullInt = Literal(null, IntegerType)
+ val nullString = Literal(null, StringType)
+ val testRelation = LocalRelation('a.string, 'b.string, 'c.string, 'd.string, 'e.int)
+
+ private def checkRewrite(rewrite: LogicalPlan): Unit = rewrite match {
+ case Aggregate(_, _, Aggregate(_, _, _: Expand)) =>
+ case _ => fail(s"Plan is not rewritten:\n$rewrite")
+ }
+
+ test("single distinct group") {
+ val input = testRelation
+ .groupBy('a)(countDistinct('e))
+ .analyze
+ val rewrite = RewriteDistinctAggregates(input)
+ comparePlans(input, rewrite)
+ }
+
+ test("single distinct group with partial aggregates") {
+ val input = testRelation
+ .groupBy('a, 'd)(
+ countDistinct('e, 'c).as('agg1),
+ max('b).as('agg2))
+ .analyze
+ val rewrite = RewriteDistinctAggregates(input)
+ comparePlans(input, rewrite)
+ }
+
+ test("single distinct group with non-partial aggregates") {
+ val input = testRelation
+ .groupBy('a, 'd)(
+ countDistinct('e, 'c).as('agg1),
+ CollectSet('b).toAggregateExpression().as('agg2))
+ .analyze
+ checkRewrite(RewriteDistinctAggregates(input))
+ }
+
+ test("multiple distinct groups") {
+ val input = testRelation
+ .groupBy('a)(countDistinct('b, 'c), countDistinct('d))
+ .analyze
+ checkRewrite(RewriteDistinctAggregates(input))
+ }
+
+ test("multiple distinct groups with partial aggregates") {
+ val input = testRelation
+ .groupBy('a)(countDistinct('b, 'c), countDistinct('d), sum('e))
+ .analyze
+ checkRewrite(RewriteDistinctAggregates(input))
+ }
+
+ test("multiple distinct groups with non-partial aggregates") {
+ val input = testRelation
+ .groupBy('a)(
+ countDistinct('b, 'c),
+ countDistinct('d),
+ CollectSet('b).toAggregateExpression())
+ .analyze
+ checkRewrite(RewriteDistinctAggregates(input))
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/0d634875/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
----------------------------------------------------------------------
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 427390a..0e172be 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
@@ -493,4 +493,12 @@ class DataFrameAggregateSuite extends QueryTest with SharedSQLContext {
Row(new java.math.BigDecimal(2.0), new java.math.BigDecimal(1.5)),
Row(new java.math.BigDecimal(3.0), new java.math.BigDecimal(1.5))))
}
+
+ test("SPARK-17616: distinct aggregate combined with a non-partial aggregate") {
+ val df = Seq((1, 3, "a"), (1, 2, "b"), (3, 4, "c"), (3, 4, "c"), (3, 5, "d"))
+ .toDF("x", "y", "z")
+ checkAnswer(
+ df.groupBy($"x").agg(countDistinct($"y"), sort_array(collect_list($"z"))),
+ Seq(Row(1, 2, Seq("a", "b")), Row(3, 2, Seq("c", "c", "d"))))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org