[SPARK-2205] [SPARK-7871] [SPARK-9372] [SQL] [WIP] Improving SQL query planner

[yhuai]

This PR introduces three improvements to SQL planner..

First, it adds an optimization rule FilterNullsInJoinKey to add Filter before join operators to filter out rows having null values for join keys.

Second, it adds NullUnsafeClusteredDistribution and NullUnsafeHashPartitioning, which can be used to distribute rows having null values for join keys evenly. NullUnsafeClusteredDistribution is basically the same with ClusteredDistribution (now renamed to NullSafeClusteredDistribution) except that it does not require rows having null values for join keys be clustered.

Third, it adds PartitioningCollection, which is used to represent the outputPartitioning for SparkPlans with multiple children (e.g. ShuffledHashJoin). So, a SparkPlan can have multiple descriptions of its partitioning schemes. Taking ShuffledHashJoin as an example, it has two descriptions of its partitioning schemes, i.e. left.outputPartitioning and right.outputPartitioning. So when we have a query like select * from t1 join t2 on (t1.x = t2.x) join t3 on (t2.x = t3.x) will only have three Exchange operators (when shuffled joins are needed) instead of four.

Optimizations in the first and second improvement are guarded by spark.sql.advancedOptimization.

I will add more comments/doc and test later.

[SparkQA]

Test build #38507 has finished for PR 7685 at commit e66d5a9.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class AtLeastNNulls(n: Int, children: Seq[Expression]) extends Predicate
  • case class AtLeastNNonNullNans(n: Int, children: Seq[Expression]) extends Predicate
  • class DefaultOptimizer extends Optimizer
  • case class NullSafeClusteredDistribution(clustering: Seq[Expression]) extends Distribution
  • case class NullUnsafeClusteredDistribution(clustering: Seq[Expression]) extends Distribution
  • case class NullSafeHashPartitioning(expressions: Seq[Expression], numPartitions: Int)
  • case class NullUnsafeHashPartitioning(expressions: Seq[Expression], numPartitions: Int)
  • case class PartitioningCollection(partitionings: Seq[Partitioning])
  • case class FilterNullsInJoinKey(

[SparkQA]

Test build #38511 has finished for PR 7685 at commit 13d1c9e.

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

[SparkQA]

Test build #38588 has finished for PR 7685 at commit d3d2e64.

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

[SparkQA]

Test build #38606 has finished for PR 7685 at commit c57a954.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class AtLeastNNulls(n: Int, children: Seq[Expression]) extends Predicate
  • case class AtLeastNNonNullNans(n: Int, children: Seq[Expression]) extends Predicate
  • class DefaultOptimizer extends Optimizer
  • case class ClusteredDistribution(
  • case class HashPartitioning(expressions: Seq[Expression], numPartitions: Int, nullSafe: Boolean)
  • case class PartitioningCollection(partitionings: Seq[Partitioning])
  • case class FilterNullsInJoinKey(

[JoshRosen]

@yhuai, do you think that we should carve this up into multiple PRs to ease reviews? It looks like FilterNullsInJoinKey should be easy to split out.

[yhuai]

@JoshRosen yeah. Let me split it.

[yhuai]

btw, FilterNullsInJoinKeySuite is not deterministic. I will change it later.

[yhuai]

I am closing it.