[SPARK-30416][SQL] Log a warning for deprecated SQL config in set() and unset()

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

@@ -200,6 +200,38 @@ object SQLConf {
     Map(configs.map { cfg => cfg.key -> cfg } : _*)
   }
 
+  /**
+   * Holds information about keys that have been deprecated.
+   *
+   * @param key The deprecated key.
+   * @param version Version of Spark where key was deprecated.
+   * @param comment Additional info regarding to the removed config. For example,
+   *                reasons of config deprecation, what users should use instead of it.
+   */
+  case class DeprecatedConfig(key: String, version: String, comment: String)
+
+  /**
+   * Maps deprecated SQL config keys to information about the deprecation.
+   *
+   * The extra information is logged as a warning when the SQL config is present
+   * in the user's configuration.
+   */
+  val deprecatedSQLConfigs: Map[String, DeprecatedConfig] = {
+    val configs = Seq(
+      DeprecatedConfig("spark.sql.hive.verifyPartitionPath", "3.0.0",
+        "This config will be replaced by spark.files.ignoreMissingFiles."),
+      DeprecatedConfig("spark.sql.execution.pandas.respectSessionTimeZone", "2.3",
+        "Behavior for `false` config value is considered as a bug, and " +
+        "it will be prohibited in the future releases."),
+      DeprecatedConfig(
+        "spark.sql.legacy.execution.pandas.groupedMap.assignColumnsByName", "2.4",
+        "The config allows to switch to the behaviour before Spark 2.4 " +
+        "and will be removed in the future releases.")
+    )
+
+    Map(configs.map { cfg => cfg.key -> cfg } : _*)
+  }
+
   val ANALYZER_MAX_ITERATIONS = buildConf("spark.sql.analyzer.maxIterations")
     .internal()
     .doc("The max number of iterations the analyzer runs.")

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

@@ -18,9 +18,10 @@
 package org.apache.spark.sql
 
 import org.apache.spark.annotation.Stable
+import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.{ConfigEntry, OptionalConfigEntry}
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.RemovedConfig
+import org.apache.spark.sql.internal.SQLConf.{DeprecatedConfig, RemovedConfig}
 
 /**
  * Runtime configuration interface for Spark. To access this, use `SparkSession.conf`.
@@ -30,7 +31,7 @@ import org.apache.spark.sql.internal.SQLConf.RemovedConfig
  * @since 2.0.0
  */
 @Stable
-class RuntimeConfig private[sql](sqlConf: SQLConf = new SQLConf) {
+class RuntimeConfig private[sql](sqlConf: SQLConf = new SQLConf) extends Logging {
 
   /**
    * Sets the given Spark runtime configuration property.
@@ -40,6 +41,7 @@ class RuntimeConfig private[sql](sqlConf: SQLConf = new SQLConf) {
   def set(key: String, value: String): Unit = {
     requireNonStaticConf(key)
     requireDefaultValueOfRemovedConf(key, value)
+    logDeprecationWarning(key)
     sqlConf.setConfString(key, value)
   }
 
@@ -128,6 +130,7 @@ class RuntimeConfig private[sql](sqlConf: SQLConf = new SQLConf) {
    */
   def unset(key: String): Unit = {
     requireNonStaticConf(key)
+    logDeprecationWarning(key)
     sqlConf.unsetConf(key)
   }
 
@@ -168,4 +171,16 @@ class RuntimeConfig private[sql](sqlConf: SQLConf = new SQLConf) {
         }
     }
   }
+
+  /**
+   * Logs a warning message if the given config key is deprecated.
+   */
+  private def logDeprecationWarning(key: String): Unit = {
+    SQLConf.deprecatedSQLConfigs.get(key).foreach {
+      case DeprecatedConfig(configName, version, comment) =>
+        logWarning(
+          s"The SQL config '$configName' has been deprecated in Spark v$version " +
+          s"and may be removed in the future. $comment")
+    }
+  }
 }

sql/core/src/test/scala/org/apache/spark/sql/internal/SQLConfSuite.scala

@@ -17,7 +17,12 @@
 
 package org.apache.spark.sql.internal
 
+import scala.collection.mutable.ArrayBuffer
+import scala.language.reflectiveCalls
+
 import org.apache.hadoop.fs.Path
+import org.apache.log4j.{AppenderSkeleton, Level}
+import org.apache.log4j.spi.LoggingEvent
 
 import org.apache.spark.sql._
 import org.apache.spark.sql.internal.StaticSQLConf._
@@ -330,4 +335,31 @@ class SQLConfSuite extends QueryTest with SharedSparkSession {
     }
     assert(e.getMessage.contains(config))
   }
+
+  test("log deprecation warnings") {
+    val logAppender = new AppenderSkeleton {
+      val loggingEvents = new ArrayBuffer[LoggingEvent]()
+
+      override def append(loggingEvent: LoggingEvent): Unit = loggingEvents.append(loggingEvent)
+      override def close(): Unit = {}
+      override def requiresLayout(): Boolean = false
+    }
+    def check(config: String): Unit = {
+      assert(logAppender.loggingEvents.exists(
+        e => e.getLevel == Level.WARN &&
+        e.getRenderedMessage.contains(config)))
+    }
+
+    val config1 = "spark.sql.hive.verifyPartitionPath"
+    withLogAppender(logAppender) {
+      spark.conf.set(config1, true)
+    }
+    check(config1)
+
+    val config2 = "spark.sql.execution.pandas.respectSessionTimeZone"
+    withLogAppender(logAppender) {
+      spark.conf.unset(config2)
+    }
+    check(config2)
+  }
 }