spark git commit: [SPARK-17124][SQL] RelationalGroupedDataset.agg should preserve order and allow multiple aggregates per column

[we...@apache.org]

Repository: spark
Updated Branches:
  refs/heads/master 31a015572 -> 9560c8d29


[SPARK-17124][SQL] RelationalGroupedDataset.agg should preserve order and allow multiple aggregates per column

## What changes were proposed in this pull request?
This patch fixes a longstanding issue with one of the RelationalGroupedDataset.agg function. Even though the signature accepts vararg of pairs, the underlying implementation turns the seq into a map, and thus not order preserving nor allowing multiple aggregates per column.

This change also allows users to use this function to run multiple different aggregations for a single column, e.g.
```
agg("age" -> "max", "age" -> "count")
```

## How was this patch tested?
Added a test case in DataFrameAggregateSuite.

Author: petermaxlee <pe...@gmail.com>

Closes #14697 from petermaxlee/SPARK-17124.


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

Branch: refs/heads/master
Commit: 9560c8d29542a5dcaaa07b7af9ef5ddcdbb5d14d
Parents: 31a0155
Author: petermaxlee <pe...@gmail.com>
Authored: Sun Aug 21 00:25:55 2016 +0800
Committer: Wenchen Fan <we...@databricks.com>
Committed: Sun Aug 21 00:25:55 2016 +0800

----------------------------------------------------------------------
 .../org/apache/spark/sql/RelationalGroupedDataset.scala   |  6 ++++--
 .../org/apache/spark/sql/DataFrameAggregateSuite.scala    | 10 ++++++++++
 2 files changed, 14 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/spark/blob/9560c8d2/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala
index 7cfd1cd..53d7324 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala
@@ -128,7 +128,7 @@ class RelationalGroupedDataset protected[sql](
   }
 
   /**
-   * (Scala-specific) Compute aggregates by specifying a map from column name to
+   * (Scala-specific) Compute aggregates by specifying the column names and
    * aggregate methods. The resulting [[DataFrame]] will also contain the grouping columns.
    *
    * The available aggregate methods are `avg`, `max`, `min`, `sum`, `count`.
@@ -143,7 +143,9 @@ class RelationalGroupedDataset protected[sql](
    * @since 1.3.0
    */
   def agg(aggExpr: (String, String), aggExprs: (String, String)*): DataFrame = {
-    agg((aggExpr +: aggExprs).toMap)
+    toDF((aggExpr +: aggExprs).map { case (colName, expr) =>
+      strToExpr(expr)(df(colName).expr)
+    })
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/spark/blob/9560c8d2/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 92aa7b9..69a3b5f 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
@@ -87,6 +87,16 @@ class DataFrameAggregateSuite extends QueryTest with SharedSQLContext {
     )
   }
 
+  test("SPARK-17124 agg should be ordering preserving") {
+    val df = spark.range(2)
+    val ret = df.groupBy("id").agg("id" -> "sum", "id" -> "count", "id" -> "min")
+    assert(ret.schema.map(_.name) == Seq("id", "sum(id)", "count(id)", "min(id)"))
+    checkAnswer(
+      ret,
+      Row(0, 0, 1, 0) :: Row(1, 1, 1, 1) :: Nil
+    )
+  }
+
   test("rollup") {
     checkAnswer(
       courseSales.rollup("course", "year").sum("earnings"),


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org