Add failing test illustrating bad exchange planning.

Author: JoshRosen

sql/core/src/test/scala/org/apache/spark/sql/execution/ExchangeSuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.plans.physical.SinglePartition
+import org.apache.spark.sql.execution.joins.SortMergeJoin
 
 class ExchangeSuite extends SparkPlanTest {
   test("shuffling UnsafeRows in exchange") {
@@ -29,4 +30,14 @@ class ExchangeSuite extends SparkPlanTest {
       input.map(Row.fromTuple)
     )
   }
+
+  test("EnsureRequirements shouldn't add exchange to SMJ inputs if both are SinglePartition") {
+    val df = (1 to 10).map(Tuple1.apply).toDF("a").repartition(1)
+    val keys = Seq(df.col("a").expr)
+    val smj = SortMergeJoin(keys, keys, df.queryExecution.sparkPlan, df.queryExecution.sparkPlan)
+    val afterEnsureRequirements = EnsureRequirements(df.sqlContext).apply(smj)
+    if (afterEnsureRequirements.collect { case Exchange(_, _) => true }.nonEmpty) {
+      fail(s"No Exchanges should have been added:\n$afterEnsureRequirements")
+    }
+  }
 }