fix changed configuration

Author: adrian-wang

sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala

@@ -177,6 +177,7 @@ private[sql] case class AddExchange(sqlContext: SQLContext) extends Rule[SparkPl
           rowOrdering: Option[Ordering[Row]] = None): SparkPlan = {
         val needSort = child.outputOrdering != rowOrdering
         if (child.outputPartitioning != partitioning || needSort) {
+          // TODO: if only needSort, we need only sort each partition instead of an Exchange
           Exchange(partitioning, child, sort = needSort)
         } else {
           child

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

@@ -65,7 +65,6 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
     cacheManager.clearCache()
 
     val SORTMERGEJOIN_ENABLED: Boolean = conf.sortMergeJoinEnabled
-    conf.setConf("spark.sql.autoSortMergeJoin", "false")
     Seq(
       ("SELECT * FROM testData LEFT SEMI JOIN testData2 ON key = a", classOf[LeftSemiJoinHash]),
       ("SELECT * FROM testData LEFT SEMI JOIN testData2", classOf[LeftSemiJoinBNL]),
@@ -96,14 +95,14 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
         classOf[BroadcastNestedLoopJoin])
     ).foreach { case (query, joinClass) => assertJoin(query, joinClass) }
     try {
-      conf.setConf("spark.sql.autoSortMergeJoin", "true")
+      conf.setConf("spark.sql.planner.sortMergeJoin", "true")
       Seq(
         ("SELECT * FROM testData JOIN testData2 ON key = a", classOf[SortMergeJoin]),
         ("SELECT * FROM testData JOIN testData2 ON key = a and key = 2", classOf[SortMergeJoin]),
         ("SELECT * FROM testData JOIN testData2 ON key = a where key = 2", classOf[SortMergeJoin])
       ).foreach { case (query, joinClass) => assertJoin(query, joinClass) }
     } finally {
-      conf.setConf("spark.sql.autoSortMergeJoin", SORTMERGEJOIN_ENABLED.toString)
+      conf.setConf("spark.sql.planner.sortMergeJoin", SORTMERGEJOIN_ENABLED.toString)
     }
   }