diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala index 731aae882a7b..be597edecba9 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala @@ -895,7 +895,7 @@ object SparkSession extends Logging { * SparkSession exists, the method creates a new SparkSession and assigns the * newly created SparkSession as the global default. * - * In case an existing SparkSession is returned, the config options specified in + * In case an existing SparkSession is returned, the non-static config options specified in * this builder will be applied to the existing SparkSession. * * @since 2.0.0 @@ -905,10 +905,7 @@ object SparkSession extends Logging { // Get the session from current thread's active session. var session = activeThreadSession.get() if ((session ne null) && !session.sparkContext.isStopped) { - options.foreach { case (k, v) => session.sessionState.conf.setConfString(k, v) } - if (options.nonEmpty) { - logWarning("Using an existing SparkSession; some configuration may not take effect.") - } + applyModifiableSettings(session) return session } @@ -917,10 +914,7 @@ object SparkSession extends Logging { // If the current thread does not have an active session, get it from the global session. session = defaultSession.get() if ((session ne null) && !session.sparkContext.isStopped) { - options.foreach { case (k, v) => session.sessionState.conf.setConfString(k, v) } - if (options.nonEmpty) { - logWarning("Using an existing SparkSession; some configuration may not take effect.") - } + applyModifiableSettings(session) return session } @@ -959,6 +953,22 @@ object SparkSession extends Logging { return session } + + private def applyModifiableSettings(session: SparkSession): Unit = { + val (staticConfs, otherConfs) = + options.partition(kv => SQLConf.staticConfKeys.contains(kv._1)) + + otherConfs.foreach { case (k, v) => session.sessionState.conf.setConfString(k, v) } + + if (staticConfs.nonEmpty) { + logWarning("Using an existing SparkSession; the static sql configurations will not take" + + " effect.") + } + if (otherConfs.nonEmpty) { + logWarning("Using an existing SparkSession; some spark core configurations may not take" + + " effect.") + } + } } /** diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionBuilderSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionBuilderSuite.scala index f2386413bea1..7b76d0702d83 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionBuilderSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionBuilderSuite.scala @@ -22,7 +22,7 @@ import org.scalatest.BeforeAndAfterEach import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite} import org.apache.spark.internal.config.UI.UI_ENABLED import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.internal.StaticSQLConf.GLOBAL_TEMP_DATABASE +import org.apache.spark.sql.internal.StaticSQLConf._ /** * Test cases for the builder pattern of [[SparkSession]]. @@ -168,4 +168,51 @@ class SparkSessionBuilderSuite extends SparkFunSuite with BeforeAndAfterEach { assert(session.sessionState.conf.getConfString("spark.app.name") === "test-app-SPARK-31234") assert(session.sessionState.conf.getConf(GLOBAL_TEMP_DATABASE) === "globaltempdb-spark-31234") } + + test("SPARK-31532: should not propagate static sql configs to the existing" + + " active/default SparkSession") { + val session = SparkSession.builder() + .master("local") + .config(GLOBAL_TEMP_DATABASE.key, value = "globalTempDB-SPARK-31532") + .config("spark.app.name", "test-app-SPARK-31532") + .getOrCreate() + // do not propagate static sql configs to the existing active session + val session1 = SparkSession + .builder() + .config(GLOBAL_TEMP_DATABASE.key, "globalTempDB-SPARK-31532-1") + .getOrCreate() + assert(session.conf.get(GLOBAL_TEMP_DATABASE) === "globaltempdb-spark-31532") + assert(session1.conf.get(GLOBAL_TEMP_DATABASE) === "globaltempdb-spark-31532") + + // do not propagate static sql configs to the existing default session + SparkSession.clearActiveSession() + val session2 = SparkSession + .builder() + .config(WAREHOUSE_PATH.key, "SPARK-31532-db") + .config(GLOBAL_TEMP_DATABASE.key, value = "globalTempDB-SPARK-31532-2") + .getOrCreate() + + assert(!session.conf.get(WAREHOUSE_PATH).contains("SPARK-31532-db")) + assert(session.conf.get(WAREHOUSE_PATH) === session2.conf.get(WAREHOUSE_PATH)) + assert(session2.conf.get(GLOBAL_TEMP_DATABASE) === "globaltempdb-spark-31532") + } + + test("SPARK-31532: propagate static sql configs if no existing SparkSession") { + val conf = new SparkConf() + .setMaster("local") + .setAppName("test-app-SPARK-31532-2") + .set(GLOBAL_TEMP_DATABASE.key, "globaltempdb-spark-31532") + .set(WAREHOUSE_PATH.key, "SPARK-31532-db") + SparkContext.getOrCreate(conf) + + // propagate static sql configs if no existing session + val session = SparkSession + .builder() + .config(GLOBAL_TEMP_DATABASE.key, "globalTempDB-SPARK-31532-2") + .config(WAREHOUSE_PATH.key, "SPARK-31532-db-2") + .getOrCreate() + assert(session.conf.get("spark.app.name") === "test-app-SPARK-31532-2") + assert(session.conf.get(GLOBAL_TEMP_DATABASE) === "globaltempdb-spark-31532-2") + assert(session.conf.get(WAREHOUSE_PATH) === "SPARK-31532-db-2") + } }