showcase, DO NOT MERGE

[cloud-fan]

What changes were proposed in this pull request?

this is another approach to fix SPARK-12978, which was done in #10896

How was this patch tested?

new test in PlannerSuite

[cloud-fan]

cc @maropu @hvanhovell @yhuai @liancheng @clockfly

[SparkQA]

Test build #64660 has finished for PR 14876 at commit a5b6d71.

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

[maropu]

Thank you for your concrete example! I'll check in hours.

[yhuai]

I think we need to have more strict conditions to make sure these two operators are for the same group by clause (Although I do not have a case showing this will break, it is better to list the condition in a more specific way).

[maropu]

I found that we need to push-down partial aggregation below exchange operators instead of merging them? For example, in the spark v2.0 branch,

(0 to 1000).map(x => (x % 2, x.toString)).toDF("a", "b").repartition($"a").createOrReplaceTempView("t")
spark.sql("select max(b) from t group by a").explain

This prints like:

== Physical Plan ==
SortAggregate(key=[a#5], functions=[max(b#6)], output=[max(b)#20])
+- SortAggregate(key=[a#5], functions=[partial_max(b#6)], output=[a#5,max#22])
   +- *Sort [a#5 ASC], false, 0
      +- Exchange hashpartitioning(a#5, 4)
         +- LocalTableScan [a#5, b#6]

In this case, I think it is more natural to push-down the partial aggregation below the exchange.

[maropu]

On the other hand, when caching the already-partitioned input table, we cannot push-down them;

(0 to 1000).map(x => (x % 2, x.toString)).toDF("a", "b").repartition($"a").cache.createOrReplaceTempView("t")
spark.sql("select max(b) from t group by a").explain

== Physical Plan ==
SortAggregate(key=[a#40], functions=[max(b#41)])
+- SortAggregate(key=[a#40], functions=[partial_max(b#41)])
   +- *Sort [a#40 ASC], false, 0
      +- InMemoryTableScan [a#40, b#41]
         :  +- InMemoryRelation [a#40, b#41], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas)
         :     :  +- Exchange hashpartitioning(a#40, 4)
         :     :     +- LocalTableScan [a#40, b#41]

ISTM all we can do is merge the aggregations into one in this case.

[cloud-fan]

yea, pushing down partial aggregate below exchange is a good idea, but I think it's out of the scope of SPARK-12978, which is aim to remove unnecessary partial aggregate right?

[maropu]

yea, I thinks so. I like the approach in this pr.

[cloud-fan]

closing, @maropu will take over