Ignores Thrift server ThriftServerPageSuite

wangyum · wangyum · commit 45109c71c3ef · 2019-11-12T18:32:36.000+08:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -349,6 +349,14 @@ object SQLConf {
     .checkValue(_ > 0, "The value of spark.sql.shuffle.partitions must be positive")
     .createWithDefault(200)
 
+  val SHUFFLE_WITHOUT_SHUFFLE_SIDE_RATIO =
+    buildConf("spark.sql.shuffle.withoutShuffleSideRatio")
+      .doc("The maximum number of without shuffle partition ratio lower than this config " +
+        "will not add shuffle exchange for it.")
+      .doubleConf
+      .checkValue(ratio => ratio > 0 && ratio <= 1, "The ratio value must be in [0, 1].")
+      .createWithDefault(1.0)
+
   val ADAPTIVE_EXECUTION_ENABLED = buildConf("spark.sql.adaptive.enabled")
     .doc("When true, enable adaptive query execution.")
     .booleanConf
@@ -2162,6 +2170,8 @@ class SQLConf extends Serializable with Logging {
 
   def numShufflePartitions: Int = getConf(SHUFFLE_PARTITIONS)
 
+  def withoutShuffleSideRatio: Double = getConf(SHUFFLE_WITHOUT_SHUFFLE_SIDE_RATIO)
+
   def adaptiveExecutionEnabled: Boolean = getConf(ADAPTIVE_EXECUTION_ENABLED)
 
   def targetPostShuffleInputSize: Long = getConf(SHUFFLE_TARGET_POSTSHUFFLE_INPUT_SIZE)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
@@ -83,22 +83,19 @@ case class EnsureRequirements(conf: SQLConf) extends Rule[SparkPlan] {
         numPartitionsSet.headOption
       }
 
-      // maxNumPostShufflePartitions is usually larger than numShufflePartitions,
-      // which causes some bucket map join lose efficacy after enabling adaptive execution.
-      // Please see SPARK-29655 for more details.
-      val expectedChildrenNumPartitions = if (conf.adaptiveExecutionEnabled) {
-        val withoutShuffleChildrenNumPartitions =
-          childrenIndexes.filterNot(children(_).isInstanceOf[ShuffleExchangeExec])
-            .map(children(_).outputPartitioning.numPartitions).toSet
-        if (withoutShuffleChildrenNumPartitions.nonEmpty &&
-          conf.maxNumPostShufflePartitions > conf.numShufflePartitions) {
-          math.min(math.max(withoutShuffleChildrenNumPartitions.max, conf.numShufflePartitions),
-            conf.maxNumPostShufflePartitions)
+      val maxNumPartition = childrenNumPartitions.max
+      val withoutShuffleChildrenNumPartitions =
+        childrenIndexes.filterNot(children(_).isInstanceOf[ShuffleExchangeExec])
+          .map(children(_).outputPartitioning.numPartitions).toSet
+      val expectedChildrenNumPartitions = if (withoutShuffleChildrenNumPartitions.nonEmpty) {
+        val withoutShuffleMaxNumPartition = withoutShuffleChildrenNumPartitions.max
+        if (withoutShuffleMaxNumPartition * 1.0 / maxNumPartition >= conf.withoutShuffleSideRatio) {
+          withoutShuffleMaxNumPartition
         } else {
-          childrenNumPartitions.max
+          maxNumPartition
         }
       } else {
-        childrenNumPartitions.max
+        maxNumPartition
       }
 
       val targetNumPartitions = requiredNumPartitions.getOrElse(expectedChildrenNumPartitions)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala
@@ -795,4 +795,43 @@ abstract class BucketedReadSuite extends QueryTest with SQLTestUtils {
     }
   }
 
+  test("Support spark.sql.shuffle.withoutShuffleSideRatio") {
+    // numBuckets >= spark.sql.shuffle.partitions
+    withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "5")  {
+      val bucketSpec = Some(BucketSpec(6, Seq("i", "j"), Nil))
+      val bucketedTableTestSpecLeft = BucketedTableTestSpec(bucketSpec, expectedShuffle = false)
+      val bucketedTableTestSpecRight = BucketedTableTestSpec(None, expectedShuffle = true)
+      testBucketing(
+        bucketedTableTestSpecLeft = bucketedTableTestSpecLeft,
+        bucketedTableTestSpecRight = bucketedTableTestSpecRight,
+        joinCondition = joinCondition(Seq("i", "j"))
+      )
+    }
+
+    // numBuckets < spark.sql.shuffle.partitions
+    withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "5")  {
+      val bucketSpec = Some(BucketSpec(4, Seq("i", "j"), Nil))
+      val bucketedTableTestSpecLeft = BucketedTableTestSpec(bucketSpec, expectedShuffle = true)
+      val bucketedTableTestSpecRight = BucketedTableTestSpec(None, expectedShuffle = true)
+      testBucketing(
+        bucketedTableTestSpecLeft = bucketedTableTestSpecLeft,
+        bucketedTableTestSpecRight = bucketedTableTestSpecRight,
+        joinCondition = joinCondition(Seq("i", "j"))
+      )
+    }
+
+    // numBuckets < spark.sql.shuffle.partitions and withoutShuffleSideRatio = 0.1
+    withSQLConf(
+      SQLConf.SHUFFLE_PARTITIONS.key -> "5",
+      SQLConf.SHUFFLE_WITHOUT_SHUFFLE_SIDE_RATIO.key -> "0.1")  {
+      val bucketSpec = Some(BucketSpec(4, Seq("i", "j"), Nil))
+      val bucketedTableTestSpecLeft = BucketedTableTestSpec(bucketSpec, expectedShuffle = false)
+      val bucketedTableTestSpecRight = BucketedTableTestSpec(None, expectedShuffle = true)
+      testBucketing(
+        bucketedTableTestSpecLeft = bucketedTableTestSpecLeft,
+        bucketedTableTestSpecRight = bucketedTableTestSpecRight,
+        joinCondition = joinCondition(Seq("i", "j"))
+      )
+    }
+  }
 }
