[SPARK-24761][SQL] Adding of isModifiable() to RuntimeConfig

python/pyspark/sql/conf.py

@@ -63,6 +63,14 @@ def _checkType(self, obj, identifier):
             raise TypeError("expected %s '%s' to be a string (was '%s')" %
                             (identifier, obj, type(obj).__name__))
 
+    @ignore_unicode_prefix
+    @since(2.4)
+    def isModifiable(self, key):
+        """Indicates whether the configuration property with the given key
+        is modifiable in the current session.
+        """
+        return self._jconf.isModifiable(key)
+
 
 def _test():
     import os

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

@@ -1892,4 +1892,8 @@ class SQLConf extends Serializable with Logging {
     }
     cloned
   }
+
+  def isModifiable(key: String): Boolean = {
+    sqlConfEntries.containsKey(key) && !staticConfKeys.contains(key)
+  }
 }

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

@@ -132,6 +132,17 @@ class RuntimeConfig private[sql](sqlConf: SQLConf = new SQLConf) {
     sqlConf.unsetConf(key)
   }
 
+  /**
+   * Indicates whether the configuration property with the given key
+   * is modifiable in the current session.
+   *
+   * @return `true` if the configuration property is modifiable. For static SQL, Spark Core,
+   *         invalid (not existing) and other non-modifiable configuration properties,
+   *         the returned value is `false`.
+   * @since 2.4.0
+   */
+  def isModifiable(key: String): Boolean = sqlConf.isModifiable(key)
+
   /**
    * Returns whether a particular key is set.
    */

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

@@ -54,4 +54,18 @@ class RuntimeConfigSuite extends SparkFunSuite {
       conf.get("k1")
     }
   }
+
+  test("SPARK-24761: is a config parameter modifiable") {
+    val conf = newConf()
+
+    // SQL configs
+    assert(!conf.isModifiable("spark.sql.sources.schemaStringLengthThreshold"))
+    assert(conf.isModifiable("spark.sql.streaming.checkpointLocation"))
+    // Core configs
+    assert(!conf.isModifiable("spark.task.cpus"))
+    assert(!conf.isModifiable("spark.executor.cores"))
+    // Invalid config parameters
+    assert(!conf.isModifiable(""))
+    assert(!conf.isModifiable("invalid config parameter"))
+  }
 }