[SPARK-20246][SQL] Don't pushdown non-deterministic expression through Aggregate

[viirya]

What changes were proposed in this pull request?

We inadvertently push down non-deterministic expression through Aggregate like this:

import org.apache.spark.sql.functions._
val df = spark.range(1,1000).distinct.withColumn("random", rand()).filter(col("random") > 0.3).orderBy("random")

The optimized plan before this:

Sort [random#4 ASC NULLS FIRST], true
+- Aggregate [id#0L], [id#0L, rand(-3812200341668550973) AS random#4]
   +- Filter (rand(-3812200341668550973) > 0.3)
      +- Range (1, 1000, step=1, splits=Some(2))

The optimized plan after this:

Sort [random#4 ASC NULLS FIRST], true
+- Filter (random#4 > 0.3)
   +- Aggregate [id#0L], [id#0L, rand(-6229677203478213930) AS random#4]
      +- Range (1, 1000, step=1, splits=Some(2))

How was this patch tested?

Will add test later.

Please review http://spark.apache.org/contributing.html before opening a pull request.

[SparkQA]

Test build #75592 has started for PR 17559 at commit 77896e9.

[viirya]

retest this please.

[viirya]

close this as there is a duplicate pr #17562.

[SparkQA]

Test build #75595 has finished for PR 17559 at commit 77896e9.

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