update

Author: lianhuiwang

sql/hive/src/test/scala/org/apache/spark/sql/hive/StatisticsSuite.scala

@@ -762,35 +762,37 @@ class StatisticsSuite extends StatisticsCollectionTestBase with TestHiveSingleto
 
   test("auto converts to broadcast join by size estimate of scanned partitions " +
     "for partitioned table") {
-    withTempView("tempTbl", "largeTbl", "partTbl") {
-      spark.range(0, 1000, 1, 2).selectExpr("id as col1", "id as col2")
-        .createOrReplaceTempView("tempTbl")
-      spark.range(0, 100000, 1, 2).selectExpr("id as col1", "id as col2").
-        createOrReplaceTempView("largeTbl")
-      sql("CREATE TABLE partTbl (col1 INT, col2 STRING) " +
-        "PARTITIONED BY (part1 STRING, part2 INT) STORED AS textfile")
-      for (part1 <- Seq("a", "b", "c", "d"); part2 <- Seq(1, 2)) {
-        sql(
-          s"""
-             |INSERT OVERWRITE TABLE partTbl PARTITION (part1='$part1',part2='$part2')
-             |select col1, col2 from tempTbl
+    withTempView("tempTbl", "largeTbl") {
+      withTable("partTbl") {
+        spark.range(0, 1000, 1, 2).selectExpr("id as col1", "id as col2")
+          .createOrReplaceTempView("tempTbl")
+        spark.range(0, 100000, 1, 2).selectExpr("id as col1", "id as col2")
+          .createOrReplaceTempView("largeTbl")
+        sql("CREATE TABLE partTbl (col1 INT, col2 STRING) " +
+          "PARTITIONED BY (part1 STRING, part2 INT) STORED AS textfile")
+        for (part1 <- Seq("a", "b", "c", "d"); part2 <- Seq(1, 2)) {
+          sql(
+            s"""
+               |INSERT OVERWRITE TABLE partTbl PARTITION (part1='$part1',part2='$part2')
+               |select col1, col2 from tempTbl
            """.stripMargin)
-      }
-      val query = "select * from largeTbl join partTbl on (largeTbl.col1 = partTbl.col1 " +
-        "and partTbl.part1 = 'a' and partTbl.part2 = 1)"
-      withSQLConf(SQLConf.ENABLE_FALL_BACK_TO_HDFS_FOR_STATS.key -> "true",
-        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "8001") {
-
-        withSQLConf(SQLConf.HIVE_METASTORE_PARTITION_PRUNING.key -> "true") {
-          val broadcastJoins =
-            sql(query).queryExecution.sparkPlan.collect { case j: BroadcastHashJoinExec => j }
-          assert(broadcastJoins.nonEmpty)
         }
-
-        withSQLConf(SQLConf.HIVE_METASTORE_PARTITION_PRUNING.key -> "false") {
-          val broadcastJoins =
-            sql(query).queryExecution.sparkPlan.collect { case j: BroadcastHashJoinExec => j }
-          assert(broadcastJoins.isEmpty)
+        val query = "select * from largeTbl join partTbl on (largeTbl.col1 = partTbl.col1 " +
+          "and partTbl.part1 = 'a' and partTbl.part2 = 1)"
+        withSQLConf(SQLConf.ENABLE_FALL_BACK_TO_HDFS_FOR_STATS.key -> "true",
+          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "8001") {
+
+          withSQLConf(SQLConf.HIVE_METASTORE_PARTITION_PRUNING.key -> "true") {
+            val broadcastJoins =
+              sql(query).queryExecution.sparkPlan.collect { case j: BroadcastHashJoinExec => j }
+            assert(broadcastJoins.nonEmpty)
+          }
+
+          withSQLConf(SQLConf.HIVE_METASTORE_PARTITION_PRUNING.key -> "false") {
+            val broadcastJoins =
+              sql(query).queryExecution.sparkPlan.collect { case j: BroadcastHashJoinExec => j }
+            assert(broadcastJoins.isEmpty)
+          }
         }
       }
     }