[SPARK-26601][SQL] Make broadcast-exchange thread pool configurable

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

@@ -126,4 +126,13 @@ object StaticSQLConf {
       .intConf
       .createWithDefault(1000)
 
+  val MAX_BROADCAST_EXCHANGE_THREADNUMBER =
+    buildStaticConf("spark.sql.broadcastExchange.maxThreadNumber")
+      .doc("The maximum degree of parallelism to fetch and broadcast the table.If we " +
+        "encounter memory issue when broadcast table we can decrease this number." +
+        "Notice the number should be carefully chosen since decrease parallelism will " +
+        "cause longer waiting for other broadcasting.And increase parallelism may " +
+        "cause memory problem.")
+      .intConf
+      .createWithDefault(128)
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala

@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.plans.physical.{BroadcastMode, BroadcastPar
 import org.apache.spark.sql.execution.{SparkPlan, SQLExecution}
 import org.apache.spark.sql.execution.joins.HashedRelation
 import org.apache.spark.sql.execution.metric.SQLMetrics
-import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.util.{SparkFatalException, ThreadUtils}
 
 /**
@@ -157,5 +157,6 @@ case class BroadcastExchangeExec(
 
 object BroadcastExchangeExec {
   private[execution] val executionContext = ExecutionContext.fromExecutorService(
-    ThreadUtils.newDaemonCachedThreadPool("broadcast-exchange", 128))
+    ThreadUtils.newDaemonCachedThreadPool("broadcast-exchange",
+      SQLConf.get.getConf(StaticSQLConf.MAX_BROADCAST_EXCHANGE_THREADNUMBER)))
 }

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

@@ -17,15 +17,18 @@
 
 package org.apache.spark.sql.execution
 
+import scala.concurrent.{Future, TimeoutException}
+import scala.concurrent.duration._
 import scala.util.Random
 
-import org.apache.spark.sql.{Dataset, Row}
+import org.apache.spark.sql.{Dataset, Row, SparkSession}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Literal}
 import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, IdentityBroadcastMode, SinglePartition}
 import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ReusedExchangeExec, ShuffleExchangeExec}
 import org.apache.spark.sql.execution.joins.HashedRelationBroadcastMode
-import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.sql.test.SharedSQLContext
+import org.apache.spark.util.ThreadUtils
 
 class ExchangeSuite extends SparkPlanTest with SharedSQLContext {
   import testImplicits._
@@ -132,4 +135,33 @@ class ExchangeSuite extends SparkPlanTest with SharedSQLContext {
     val projection2 = cached.select("_1", "_3").queryExecution.executedPlan
     assert(!projection1.sameResult(projection2))
   }
+
+  test("SPARK-26601: Make broadcast-exchange thread pool configurable") {
+    val previousNumber = SparkSession.getActiveSession.get.sparkContext.conf
+      .get(StaticSQLConf.MAX_BROADCAST_EXCHANGE_THREADNUMBER)
+
+    SparkSession.getActiveSession.get.sparkContext.conf.
+      set(StaticSQLConf.MAX_BROADCAST_EXCHANGE_THREADNUMBER, 1)
+    assert(SQLConf.get.getConf(StaticSQLConf.MAX_BROADCAST_EXCHANGE_THREADNUMBER) === 1)
+
+    Future {
+      Thread.sleep(5*1000)
+    } (BroadcastExchangeExec.executionContext)
+
+    val f = Future {} (BroadcastExchangeExec.executionContext)
+    intercept[TimeoutException] {
+      ThreadUtils.awaitResult(f, 3 seconds)
+    }
+
+    var executed = false
+    val ef = Future {
+      executed = true
+    } (BroadcastExchangeExec.executionContext)
+    ThreadUtils.awaitResult(ef, 3 seconds)
+    assert(executed)
+
+    // for other test
+    SparkSession.getActiveSession.get.sparkContext.conf.
+      set(StaticSQLConf.MAX_BROADCAST_EXCHANGE_THREADNUMBER, previousNumber)
+  }
 }