[SPARK-32991][SQL] [FOLLOWUP] Reset command relies on session initials first

### What changes were proposed in this pull request?

As a follow-up of #30045, we modify the RESET command here to respect the session initial configs per session first then fall back to the `SharedState` conf, which makes each session could maintain a different copy of initial configs for resetting.

### Why are the changes needed?

to make reset command saner.
### Does this PR introduce _any_ user-facing change?

yes, RESET will respect session initials first not always go to the system defaults

### How was this patch tested?

add new tests

Closes #30642 from yaooqinn/SPARK-32991-F.

Authored-by: Kent Yao <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 205d8e4)
Signed-off-by: Wenchen Fan <[email protected]>

Author: yaooqinn

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

@@ -29,7 +29,7 @@ import scala.util.matching.Regex
 
 import org.apache.hadoop.fs.Path
 
-import org.apache.spark.{SparkContext, TaskContext}
+import org.apache.spark.{SparkConf, SparkContext, TaskContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config._
 import org.apache.spark.internal.config.{IGNORE_MISSING_FILES => SPARK_IGNORE_MISSING_FILES}
@@ -77,6 +77,29 @@ object SQLConf {
     }
   }
 
+  /**
+   * Merge all non-static configs to the SQLConf. For example, when the 1st [[SparkSession]] and
+   * the global [[SharedState]] have been initialized, all static configs have taken affect and
+   * should not be set to other values. Other later created sessions should respect all static
+   * configs and only be able to change non-static configs.
+   */
+  private[sql] def mergeNonStaticSQLConfigs(
+      sqlConf: SQLConf,
+      configs: Map[String, String]): Unit = {
+    for ((k, v) <- configs if !staticConfKeys.contains(k)) {
+      sqlConf.setConfString(k, v)
+    }
+  }
+
+  /**
+   * Extract entries from `SparkConf` and put them in the `SQLConf`
+   */
+  private[sql] def mergeSparkConf(sqlConf: SQLConf, sparkConf: SparkConf): Unit = {
+    sparkConf.getAll.foreach { case (k, v) =>
+      sqlConf.setConfString(k, v)
+    }
+  }
+
   /**
    * Default config. Only used when there is no active SparkSession for the thread.
    * See [[get]] for more information.

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

@@ -83,7 +83,7 @@ class SparkSession private(
     @transient private val existingSharedState: Option[SharedState],
     @transient private val parentSessionState: Option[SessionState],
     @transient private[sql] val extensions: SparkSessionExtensions,
-    @transient private val initialSessionOptions: Map[String, String])
+    @transient private[sql] val initialSessionOptions: Map[String, String])
   extends Serializable with Closeable with Logging { self =>
 
   // The call site where this SparkSession was constructed.

sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala

@@ -172,16 +172,18 @@ object SetCommand {
 case class ResetCommand(config: Option[String]) extends RunnableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
-    val defaults = sparkSession.sharedState.conf
+    val globalInitialConfigs = sparkSession.sharedState.conf
     config match {
       case Some(key) =>
         sparkSession.conf.unset(key)
-        defaults.getOption(key).foreach(sparkSession.conf.set(key, _))
+        sparkSession.initialSessionOptions.get(key)
+          .orElse(globalInitialConfigs.getOption(key))
+          .foreach(sparkSession.conf.set(key, _))
       case None =>
         sparkSession.sessionState.conf.clear()
-        defaults.getAll.foreach { case (k, v) =>
-          sparkSession.sessionState.conf.setConfString(k, v)
-        }
+        SQLConf.mergeSparkConf(sparkSession.sessionState.conf, globalInitialConfigs)
+        SQLConf.mergeNonStaticSQLConfigs(sparkSession.sessionState.conf,
+          sparkSession.initialSessionOptions)
     }
     Seq.empty[Row]
   }

sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala

@@ -16,7 +16,6 @@
  */
 package org.apache.spark.sql.internal
 
-import org.apache.spark.SparkConf
 import org.apache.spark.annotation.Unstable
 import org.apache.spark.sql.{ExperimentalMethods, SparkSession, UDFRegistration, _}
 import org.apache.spark.sql.catalyst.analysis.{Analyzer, FunctionRegistry, ResolveSessionCatalog}
@@ -73,15 +72,6 @@ abstract class BaseSessionStateBuilder(
    */
   protected def extensions: SparkSessionExtensions = session.extensions
 
-  /**
-   * Extract entries from `SparkConf` and put them in the `SQLConf`
-   */
-  protected def mergeSparkConf(sqlConf: SQLConf, sparkConf: SparkConf): Unit = {
-    sparkConf.getAll.foreach { case (k, v) =>
-      sqlConf.setConfString(k, v)
-    }
-  }
-
   /**
    * SQL-specific key-value configurations.
    *
@@ -92,15 +82,15 @@ abstract class BaseSessionStateBuilder(
     parentState.map { s =>
       val cloned = s.conf.clone()
       if (session.sparkContext.conf.get(StaticSQLConf.SQL_LEGACY_SESSION_INIT_WITH_DEFAULTS)) {
-        mergeSparkConf(cloned, session.sparkContext.conf)
+        SQLConf.mergeSparkConf(cloned, session.sparkContext.conf)
       }
       cloned
     }.getOrElse {
       val conf = new SQLConf
-      mergeSparkConf(conf, session.sparkContext.conf)
-      options.foreach {
-        case (k, v) => conf.setConfString(k, v)
-      }
+      SQLConf.mergeSparkConf(conf, session.sharedState.conf)
+      // the later added configs to spark conf shall be respected too
+      SQLConf.mergeNonStaticSQLConfigs(conf, session.sparkContext.conf.getAll.toMap)
+      SQLConf.mergeNonStaticSQLConfigs(conf, session.initialSessionOptions)
       conf
     }
   }
@@ -372,7 +362,7 @@ private[sql] trait WithTestConf { self: BaseSessionStateBuilder =>
     parentState.map { s =>
       val cloned = s.conf.clone()
       if (session.sparkContext.conf.get(StaticSQLConf.SQL_LEGACY_SESSION_INIT_WITH_DEFAULTS)) {
-        mergeSparkConf(conf, session.sparkContext.conf)
+        SQLConf.mergeSparkConf(conf, session.sparkContext.conf)
       }
       cloned
     }.getOrElse {
@@ -384,7 +374,7 @@ private[sql] trait WithTestConf { self: BaseSessionStateBuilder =>
           overrideConfigurations.foreach { case (key, value) => setConfString(key, value) }
         }
       }
-      mergeSparkConf(conf, session.sparkContext.conf)
+      SQLConf.mergeSparkConf(conf, session.sparkContext.conf)
       conf
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala

@@ -67,11 +67,12 @@ private[sql] class SharedState(
       case (k, _)  if k == "hive.metastore.warehouse.dir" || k == WAREHOUSE_PATH.key =>
         logWarning(s"Not allowing to set ${WAREHOUSE_PATH.key} or hive.metastore.warehouse.dir " +
           s"in SparkSession's options, it should be set statically for cross-session usages")
-      case (k, v) =>
-        logDebug(s"Applying initial SparkSession options to SparkConf/HadoopConf: $k -> $v")
+      case (k, v) if SQLConf.staticConfKeys.contains(k) =>
+        logDebug(s"Applying static initial session options to SparkConf: $k -> $v")
         confClone.set(k, v)
+      case (k, v) =>
+        logDebug(s"Applying other initial session options to HadoopConf: $k -> $v")
         hadoopConfClone.set(k, v)
-
     }
     (confClone, hadoopConfClone)
   }

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

@@ -24,6 +24,7 @@ import org.apache.spark.internal.config.EXECUTOR_ALLOW_SPARK_CONTEXT
 import org.apache.spark.internal.config.UI.UI_ENABLED
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.StaticSQLConf._
+import org.apache.spark.util.ThreadUtils
 
 /**
  * Test cases for the builder pattern of [[SparkSession]].
@@ -305,34 +306,84 @@ class SparkSessionBuilderSuite extends SparkFunSuite with BeforeAndAfterEach {
     // newly specified values
     val sharedWH = spark.sharedState.conf.get(wh)
     val sharedTD = spark.sharedState.conf.get(td)
-    val sharedCustom = spark.sharedState.conf.get(custom)
     assert(sharedWH === "./data2",
       "The warehouse dir in shared state should be determined by the 1st created spark session")
     assert(sharedTD === "alice",
       "Static sql configs in shared state should be determined by the 1st created spark session")
-    assert(sharedCustom === "kyao",
-      "Dynamic sql configs in shared state should be determined by the 1st created spark session")
+    assert(spark.sharedState.conf.getOption(custom).isEmpty,
+      "Dynamic sql configs is session specific")
 
     assert(spark.conf.get(wh) === sharedWH,
       "The warehouse dir in session conf and shared state conf should be consistent")
     assert(spark.conf.get(td) === sharedTD,
       "Static sql configs in session conf and shared state conf should be consistent")
-    assert(spark.conf.get(custom) === sharedCustom,
-      "Dynamic sql configs in session conf and shared state conf should be consistent before" +
-        " setting to new ones")
+    assert(spark.conf.get(custom) === "kyao", "Dynamic sql configs is session specific")
 
     spark.sql("RESET")
 
     assert(spark.conf.get(wh) === sharedWH,
       "The warehouse dir in shared state should be respect after RESET")
     assert(spark.conf.get(td) === sharedTD,
       "Static sql configs in shared state should be respect after RESET")
-    assert(spark.conf.get(custom) === sharedCustom,
-      "Dynamic sql configs in shared state should be respect after RESET")
+    assert(spark.conf.get(custom) === "kyao",
+      "Dynamic sql configs in session initial map should be respect after RESET")
 
-    val spark2 = SparkSession.builder().getOrCreate()
+    val spark2 = SparkSession.builder()
+      .config(wh, "./data3")
+      .config(custom, "kyaoo").getOrCreate()
     assert(spark2.conf.get(wh) === sharedWH)
     assert(spark2.conf.get(td) === sharedTD)
-    assert(spark2.conf.get(custom) === sharedCustom)
+    assert(spark2.conf.get(custom) === "kyaoo")
+  }
+
+  test("SPARK-32991: RESET should work properly with multi threads") {
+    val wh = "spark.sql.warehouse.dir"
+    val td = "spark.sql.globalTempDatabase"
+    val custom = "spark.sql.custom"
+    val spark = ThreadUtils.runInNewThread("new session 0", false) {
+      SparkSession.builder()
+        .master("local")
+        .config(wh, "./data0")
+        .config(td, "bob")
+        .config(custom, "c0")
+        .getOrCreate()
+    }
+
+    spark.sql(s"SET $custom=c1")
+    assert(spark.conf.get(custom) === "c1")
+    spark.sql("RESET")
+    assert(spark.conf.get(wh) === "./data0",
+      "The warehouse dir in shared state should be respect after RESET")
+    assert(spark.conf.get(td) === "bob",
+      "Static sql configs in shared state should be respect after RESET")
+    assert(spark.conf.get(custom) === "c0",
+      "Dynamic sql configs in shared state should be respect after RESET")
+
+    val spark1 = ThreadUtils.runInNewThread("new session 1", false) {
+      SparkSession.builder().getOrCreate()
+    }
+
+    assert(spark === spark1)
+
+    // TODO: SPARK-33718: After clear sessions, the SharedState will be unreachable, then all
+    // the new static will take effect.
+    SparkSession.clearDefaultSession()
+    val spark2 = ThreadUtils.runInNewThread("new session 2", false) {
+      SparkSession.builder()
+        .master("local")
+        .config(wh, "./data1")
+        .config(td, "alice")
+        .config(custom, "c2")
+        .getOrCreate()
+    }
+
+    assert(spark2 !== spark)
+    spark2.sql(s"SET $custom=c1")
+    assert(spark2.conf.get(custom) === "c1")
+    spark2.sql("RESET")
+    assert(spark2.conf.get(wh) === "./data1")
+    assert(spark2.conf.get(td) === "alice")
+    assert(spark2.conf.get(custom) === "c2")
+
   }
 }