[CORE] Fix GlutenConfig runtime modifiability

backends-velox/src/test/scala/org/apache/gluten/config/AllVeloxConfiguration.scala

@@ -42,8 +42,8 @@ class AllVeloxConfiguration extends AnyFunSuite {
       s"""
          |## Gluten Velox backend configurations
          |
-         | Key | Default | Description
-         | --- | --- | ---
+         | Key | Status | Default | Description
+         | --- | --- | --- | ---
          |"""
 
     VeloxConfig.allEntries
@@ -53,16 +53,20 @@ class AllVeloxConfiguration extends AnyFunSuite {
       .foreach {
         entry =>
           val dft = entry.defaultValueString.replace("<", "&lt;").replace(">", "&gt;")
-          builder += Seq(s"${entry.key}", s"$dft", s"${entry.doc}")
+          builder += Seq(
+            s"${entry.key}",
+            AllGlutenConfiguration.configStatus(entry),
+            s"$dft",
+            s"${entry.doc}")
             .mkString("|")
       }
 
     builder ++=
       s"""
          |## Gluten Velox backend *experimental* configurations
          |
-         | Key | Default | Description
-         | --- | --- | ---
+         | Key | Status | Default | Description
+         | --- | --- | --- | ---
          |"""
 
     VeloxConfig.allEntries
@@ -72,7 +76,11 @@ class AllVeloxConfiguration extends AnyFunSuite {
       .foreach {
         entry =>
           val dft = entry.defaultValueString.replace("<", "&lt;").replace(">", "&gt;")
-          builder += Seq(s"${entry.key}", s"$dft", s"${entry.doc}")
+          builder += Seq(
+            s"${entry.key}",
+            AllGlutenConfiguration.configStatus(entry),
+            s"$dft",
+            s"${entry.doc}")
             .mkString("|")
       }
 

gluten-core/src/main/scala/org/apache/gluten/config/ConfigRegistry.scala

@@ -29,6 +29,35 @@ trait ConfigRegistry {
     require(existing.isEmpty, s"Config entry ${entry.key} already registered!")
   }
 
+  private def registerToSQLConf(entry: ConfigEntry[_], isStatic: Boolean): Unit = {
+    (entry.key :: entry.alternatives).foreach(registerToSQLConf(entry, _, isStatic))
+  }
+
+  private def registerToSQLConf(entry: ConfigEntry[_], key: String, isStatic: Boolean): Unit = {
+    var builder =
+      if (isStatic) SQLConf.buildStaticConf(key) else SQLConf.buildConf(key)
+    if (entry.doc.nonEmpty) {
+      builder = builder.doc(entry.doc)
+    }
+    if (entry.version.nonEmpty) {
+      builder = builder.version(entry.version)
+    }
+    if (!entry.isPublic) {
+      builder = builder.internal()
+    }
+
+    val sparkEntry = builder.stringConf.transform {
+      value =>
+        entry.valueConverter(value)
+        value
+    }
+    if (entry.defaultValue.isDefined) {
+      sparkEntry.createWithDefaultString(entry.defaultValueString)
+    } else {
+      sparkEntry.createOptional
+    }
+  }
+
   /** Visible for testing. */
   private[config] def allEntries: Seq[ConfigEntry[_]] = {
     configEntries.values.toSeq
@@ -38,15 +67,16 @@ trait ConfigRegistry {
     ConfigBuilder(key).onCreate {
       entry =>
         register(entry)
+        registerToSQLConf(entry, isStatic = false)
         ConfigRegistry.registerToAllEntries(entry)
     }
   }
 
   protected def buildStaticConf(key: String): ConfigBuilder = {
     ConfigBuilder(key).onCreate {
       entry =>
-        SQLConf.registerStaticConfigKey(key)
         register(entry)
+        registerToSQLConf(entry, isStatic = true)
         ConfigRegistry.registerToAllEntries(entry)
     }
   }

gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala

@@ -18,6 +18,7 @@ package org.apache.gluten.config
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.internal.{SQLConf, SQLConfProvider}
 
 class GlutenCoreConfig(conf: SQLConf) extends Logging {
@@ -62,7 +63,14 @@ class GlutenCoreConfig(conf: SQLConf) extends Logging {
  */
 object GlutenCoreConfig extends ConfigRegistry {
   override def get: GlutenCoreConfig = {
-    new GlutenCoreConfig(SQLConf.get)
+    new GlutenCoreConfig(activeSQLConf)
+  }
+
+  private[gluten] def activeSQLConf: SQLConf = {
+    SparkSession.getActiveSession
+      .filterNot(_.sparkContext.isStopped)
+      .map(_.sessionState.conf)
+      .getOrElse(SQLConf.get)
   }
 
   val SPARK_OFFHEAP_SIZE_KEY = "spark.memory.offHeap.size"

gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala

@@ -462,7 +462,7 @@ object GlutenConfig extends ConfigRegistry {
   val SPARK_MAX_BROADCAST_TABLE_SIZE = "spark.sql.maxBroadcastTableSize"
 
   def get: GlutenConfig = {
-    new GlutenConfig(SQLConf.get)
+    new GlutenConfig(GlutenCoreConfig.activeSQLConf)
   }
 
   def prefixOf(backendName: String): String = s"spark.gluten.sql.columnar.backend.$backendName"

gluten-substrait/src/test/scala/org/apache/gluten/config/AllGlutenConfiguration.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.gluten.config
 
+import org.apache.spark.sql.internal.SQLConf
+
 import org.scalactic.Prettifier
 import org.scalactic.source.Position
 import org.scalatest.Assertions._
@@ -67,44 +69,53 @@ class AllGlutenConfiguration extends AnyFunSuite {
       s"""
          |## Spark base configurations for Gluten plugin
          |
-         | Key | Recommend Setting | Description
-         | --- | --- | ---
+         | Key | Status | Recommend Setting | Description
+         | --- | --- | --- | ---
          |"""
 
     // scalastyle:off
     builder += Seq(
       "spark.plugins",
+      AllGlutenConfiguration.staticConfigStatus,
       "org.apache.gluten.GlutenPlugin",
       "To load Gluten's components by Spark's plug-in loader.").mkString("|")
     builder += Seq(
       "spark.memory.offHeap.enabled",
+      AllGlutenConfiguration.staticConfigStatus,
       "true",
       "Gluten use off-heap memory for certain operations.").mkString("|")
     builder += Seq(
       "spark.memory.offHeap.size",
+      AllGlutenConfiguration.staticConfigStatus,
       "30G",
       "The absolute amount of memory which can be used for off-heap allocation, in bytes unless otherwise specified.<br /> Note: Gluten Plugin will leverage this setting to allocate memory space for native usage even offHeap is disabled. <br /> The value is based on your system and it is recommended to set it larger if you are facing Out of Memory issue in Gluten Plugin."
     ).mkString("|")
     builder += Seq(
       "spark.shuffle.manager",
+      AllGlutenConfiguration.staticConfigStatus,
       "org.apache.spark.shuffle.sort.ColumnarShuffleManager",
-      "To turn on Gluten Columnar Shuffle Plugin.").mkString("|")
+      "To turn on Gluten Columnar Shuffle Plugin."
+    ).mkString("|")
     builder += Seq(
       "spark.driver.extraClassPath",
+      AllGlutenConfiguration.staticConfigStatus,
       "/path/to/gluten_jar_file",
-      "Gluten Plugin jar file to prepend to the classpath of the driver.").mkString("|")
+      "Gluten Plugin jar file to prepend to the classpath of the driver."
+    ).mkString("|")
     builder += Seq(
       "spark.executor.extraClassPath",
+      AllGlutenConfiguration.staticConfigStatus,
       "/path/to/gluten_jar_file",
-      "Gluten Plugin jar file to prepend to the classpath of executors.").mkString("|")
+      "Gluten Plugin jar file to prepend to the classpath of executors."
+    ).mkString("|")
     // scalastyle:on
 
     builder ++=
       s"""
          |## Gluten configurations
          |
-         | Key | Default | Description
-         | --- | --- | ---
+         | Key | Status | Default | Description
+         | --- | --- | --- | ---
          |"""
 
     val allEntries = GlutenConfig.allEntries ++ GlutenCoreConfig.allEntries
@@ -116,16 +127,20 @@ class AllGlutenConfiguration extends AnyFunSuite {
       .foreach {
         entry =>
           val dft = entry.defaultValueString.replace("<", "&lt;").replace(">", "&gt;")
-          builder += Seq(s"${entry.key}", s"$dft", s"${entry.doc}")
+          builder += Seq(
+            s"${entry.key}",
+            AllGlutenConfiguration.configStatus(entry),
+            s"$dft",
+            s"${entry.doc}")
             .mkString("|")
       }
 
     builder ++=
       s"""
          |## Gluten *experimental* configurations
          |
-         | Key | Default | Description
-         | --- | --- | ---
+         | Key | Status | Default | Description
+         | --- | --- | --- | ---
          |"""
 
     allEntries
@@ -135,7 +150,11 @@ class AllGlutenConfiguration extends AnyFunSuite {
       .foreach {
         entry =>
           val dft = entry.defaultValueString.replace("<", "&lt;").replace(">", "&gt;")
-          builder += Seq(s"${entry.key}", s"$dft", s"${entry.doc}")
+          builder += Seq(
+            s"${entry.key}",
+            AllGlutenConfiguration.configStatus(entry),
+            s"$dft",
+            s"${entry.doc}")
             .mkString("|")
       }
 
@@ -147,6 +166,16 @@ class AllGlutenConfiguration extends AnyFunSuite {
 }
 
 object AllGlutenConfiguration {
+  private val Anchor = Character.toString(0x2693)
+  private val CounterclockwiseArrows = new String(Character.toChars(0x1f504))
+
+  val staticConfigStatus: String = s"$Anchor Static"
+  val dynamicConfigStatus: String = s"$CounterclockwiseArrows Dynamic"
+
+  def configStatus(entry: ConfigEntry[_]): String = {
+    if (SQLConf.isStaticConfigKey(entry.key)) staticConfigStatus else dynamicConfigStatus
+  }
+
   def isRegenerateGoldenFiles: Boolean = sys.env.get("GLUTEN_UPDATE").contains("1")
 
   def getCodeSourceLocation[T](clazz: Class[T]): String = {

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenRuntimeConfigSuite.scala

@@ -16,4 +16,26 @@
  */
 package org.apache.spark.sql
 
-class GlutenRuntimeConfigSuite extends RuntimeConfigSuite with GlutenTestsTrait {}
+import org.apache.gluten.config.GlutenConfig
+
+class GlutenRuntimeConfigSuite extends RuntimeConfigSuite with GlutenTestsTrait {
+  test("Gluten configs report correct runtime modifiability") {
+    val conf = SparkSession.active.conf
+    assert(conf.isModifiable(GlutenConfig.COLUMNAR_FILESCAN_ENABLED.key))
+    assert(!conf.isModifiable(GlutenConfig.GLUTEN_UI_ENABLED.key))
+  }
+
+  test("GlutenConfig reads active SparkSession runtime configs") {
+    val conf = SparkSession.active.conf
+    val key = GlutenConfig.COLUMNAR_FILESCAN_ENABLED.key
+    val original = conf.get(key)
+    try {
+      conf.set(key, false)
+      assert(!GlutenConfig.get.enableColumnarFileScan)
+      conf.set(key, true)
+      assert(GlutenConfig.get.enableColumnarFileScan)
+    } finally {
+      conf.set(key, original)
+    }
+  }
+}

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenSparkSqlParserSuite.scala

@@ -18,4 +18,37 @@ package org.apache.spark.sql.execution
 
 import org.apache.spark.sql.GlutenSQLTestsBaseTrait
 
-class GlutenSparkSqlParserSuite extends SparkSqlParserSuite with GlutenSQLTestsBaseTrait {}
+import org.scalactic.source.Position
+import org.scalatest.Tag
+
+class GlutenSparkSqlParserSuite extends SparkSqlParserSuite with GlutenSQLTestsBaseTrait {
+  private var registerQuotedConfigParserTest = false
+
+  override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit
+      pos: Position): Unit = {
+    if (isConfigParserCoverage(testName) && !registerQuotedConfigParserTest) {
+      ()
+    } else {
+      super.test(testName, testTags: _*)(testFun)(pos)
+    }
+  }
+
+  registerQuotedConfigParserTest = true
+  test("Checks if SET/RESET can parse all the configurations") {
+    sqlConf.getAllDefinedConfs.map(_._1).foreach {
+      key: String =>
+        val quotedKey = quoteConfigKey(key)
+        spark.sessionState.sqlParser.parsePlan(s"SET $quotedKey")
+        spark.sessionState.sqlParser.parsePlan(s"RESET $quotedKey")
+    }
+  }
+  registerQuotedConfigParserTest = false
+
+  private def isConfigParserCoverage(testName: String): Boolean = {
+    testName == "Checks if SET/RESET can parse all the configurations"
+  }
+
+  private def quoteConfigKey(key: String): String = {
+    s"`${key.replace("`", "``")}`"
+  }
+}

gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenRuntimeConfigSuite.scala

@@ -16,6 +16,28 @@
  */
 package org.apache.spark.sql
 
+import org.apache.gluten.config.GlutenConfig
+
 import org.apache.spark.sql.shim.GlutenTestsTrait
 
-class GlutenRuntimeConfigSuite extends RuntimeConfigSuite with GlutenTestsTrait {}
+class GlutenRuntimeConfigSuite extends RuntimeConfigSuite with GlutenTestsTrait {
+  test("Gluten configs report correct runtime modifiability") {
+    val conf = SparkSession.active.conf
+    assert(conf.isModifiable(GlutenConfig.COLUMNAR_FILESCAN_ENABLED.key))
+    assert(!conf.isModifiable(GlutenConfig.GLUTEN_UI_ENABLED.key))
+  }
+
+  test("GlutenConfig reads active SparkSession runtime configs") {
+    val conf = SparkSession.active.conf
+    val key = GlutenConfig.COLUMNAR_FILESCAN_ENABLED.key
+    val original = conf.get(key)
+    try {
+      conf.set(key, false)
+      assert(!GlutenConfig.get.enableColumnarFileScan)
+      conf.set(key, true)
+      assert(GlutenConfig.get.enableColumnarFileScan)
+    } finally {
+      conf.set(key, original)
+    }
+  }
+}

gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/GlutenSparkSqlParserSuite.scala

@@ -18,4 +18,37 @@ package org.apache.spark.sql.execution
 
 import org.apache.spark.sql.GlutenSQLTestsBaseTrait
 
-class GlutenSparkSqlParserSuite extends SparkSqlParserSuite with GlutenSQLTestsBaseTrait {}
+import org.scalactic.source.Position
+import org.scalatest.Tag
+
+class GlutenSparkSqlParserSuite extends SparkSqlParserSuite with GlutenSQLTestsBaseTrait {
+  private var registerQuotedConfigParserTest = false
+
+  override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit
+      pos: Position): Unit = {
+    if (isConfigParserCoverage(testName) && !registerQuotedConfigParserTest) {
+      ()
+    } else {
+      super.test(testName, testTags: _*)(testFun)(pos)
+    }
+  }
+
+  registerQuotedConfigParserTest = true
+  test("Checks if SET/RESET can parse all the configurations") {
+    sqlConf.getAllDefinedConfs.map(_._1).foreach {
+      key: String =>
+        val quotedKey = quoteConfigKey(key)
+        spark.sessionState.sqlParser.parsePlan(s"SET $quotedKey")
+        spark.sessionState.sqlParser.parsePlan(s"RESET $quotedKey")
+    }
+  }
+  registerQuotedConfigParserTest = false
+
+  private def isConfigParserCoverage(testName: String): Boolean = {
+    testName == "Checks if SET/RESET can parse all the configurations"
+  }
+
+  private def quoteConfigKey(key: String): String = {
+    s"`${key.replace("`", "``")}`"
+  }
+}