[SPARK-20273] [SQL] Disallow Non-deterministic Filter push-down into Join Conditions

[gatorsmile]

What changes were proposed in this pull request?

sql("SELECT t1.b, rand(0) as r FROM cachedData, cachedData t1 GROUP BY t1.b having r > 0.5").show()

We will get the following error:

Job aborted due to stage failure: Task 1 in stage 4.0 failed 1 times, most recent failure: Lost task 1.0 in stage 4.0 (TID 8, localhost, executor driver): java.lang.NullPointerException
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificPredicate.eval(Unknown Source)
	at org.apache.spark.sql.execution.joins.BroadcastNestedLoopJoinExec$$anonfun$org$apache$spark$sql$execution$joins$BroadcastNestedLoopJoinExec$$boundCondition$1.apply(BroadcastNestedLoopJoinExec.scala:87)
	at org.apache.spark.sql.execution.joins.BroadcastNestedLoopJoinExec$$anonfun$org$apache$spark$sql$execution$joins$BroadcastNestedLoopJoinExec$$boundCondition$1.apply(BroadcastNestedLoopJoinExec.scala:87)
	at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:463)

Filters could be pushed down to the join conditions by the optimizer rule PushPredicateThroughJoin. However, Analyzer blocks users to add non-deterministics conditions (For details, see the PR #7535).

We should not push down non-deterministic conditions; otherwise, we need to explicitly initialize the non-deterministic expressions. This PR is to simply block it.

How was this patch tested?

Added a test case

[gatorsmile]

cc @cloud-fan

[SparkQA]

Test build #75644 has started for PR 17585 at commit be3fb64.

[cloud-fan]

retest this please

[cloud-fan]

LGTM

[SparkQA]

Test build #75650 has finished for PR 17585 at commit be3fb64.

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