Refactor SQLConf to display better error message

1. Add `SQLConfEntry` to store the information about a configuration. For those configurations that cannot be found in `sql-programming-guide.md`, I left the doc as `<TODO>`.
2. Verify the value when setting a configuration if this is in SQLConf.
3. Add a command `conf` to display all public configurations.

Author: zsxwing

docs/sql-programming-guide.md

@@ -1220,7 +1220,7 @@ Configuration of Parquet can be done using the `setConf` method on `SQLContext`
   <td>false</td>
   <td>
     Some other Parquet-producing systems, in particular Impala and older versions of Spark SQL, do
-    not differentiate between binary data and strings when writing out the Parquet schema.  This
+    not differentiate between binary data and strings when writing out the Parquet schema. This
     flag tells Spark SQL to interpret binary data as a string to provide compatibility with these systems.
   </td>
 </tr>
@@ -1237,7 +1237,7 @@ Configuration of Parquet can be done using the `setConf` method on `SQLContext`
   <td><code>spark.sql.parquet.cacheMetadata</code></td>
   <td>true</td>
   <td>
-    Turns on caching of Parquet schema metadata.  Can speed up querying of static data.
+    Turns on caching of Parquet schema metadata. Can speed up querying of static data.
   </td>
 </tr>
 <tr>

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

@@ -86,15 +86,15 @@ class SQLContext(@transient val sparkContext: SparkContext)
    * @group config
    * @since 1.0.0
    */
-  def setConf(key: String, value: String): Unit = conf.setConf(key, value)
+  def setConf(key: String, value: String): Unit = conf.setRawConf(key, value)
 
   /**
    * Return the value of Spark SQL configuration property for the given key.
    *
    * @group config
    * @since 1.0.0
    */
-  def getConf(key: String): String = conf.getConf(key)
+  def getConf(key: String): String = conf.getRawConf(key)
 
   /**
    * Return the value of Spark SQL configuration property for the given key. If the key is not set
@@ -103,7 +103,7 @@ class SQLContext(@transient val sparkContext: SparkContext)
    * @group config
    * @since 1.0.0
    */
-  def getConf(key: String, defaultValue: String): String = conf.getConf(key, defaultValue)
+  def getConf(key: String, defaultValue: String): String = conf.getRawConf(key, defaultValue)
 
   /**
    * Return all the configuration properties that have been set (i.e. not the default).
@@ -154,7 +154,7 @@ class SQLContext(@transient val sparkContext: SparkContext)
         // SET spark.sql.dialect=sql; Let's reset as default dialect automatically.
         val dialect = conf.dialect
         // reset the sql dialect
-        conf.unsetConf(SQLConf.DIALECT)
+        conf.unsetConf(SQLConf.DIALECT.key)
         // throw out the exception, and the default sql dialect will take effect for next query.
         throw new DialectException(
           s"""Instantiating dialect '$dialect' failed.

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

@@ -57,6 +57,7 @@ private[sql] class SparkSQLParser(fallback: String => LogicalPlan) extends Abstr
   protected val AS = Keyword("AS")
   protected val CACHE = Keyword("CACHE")
   protected val CLEAR = Keyword("CLEAR")
+  protected val CONF = Keyword("CONF")
   protected val IN = Keyword("IN")
   protected val LAZY = Keyword("LAZY")
   protected val SET = Keyword("SET")
@@ -65,7 +66,8 @@ private[sql] class SparkSQLParser(fallback: String => LogicalPlan) extends Abstr
   protected val TABLES = Keyword("TABLES")
   protected val UNCACHE = Keyword("UNCACHE")
 
-  override protected lazy val start: Parser[LogicalPlan] = cache | uncache | set | show | others
+  override protected lazy val start: Parser[LogicalPlan] =
+    cache | uncache | set | show | conf | others
 
   private lazy val cache: Parser[LogicalPlan] =
     CACHE ~> LAZY.? ~ (TABLE ~> ident) ~ (AS ~> restInput).? ^^ {
@@ -90,6 +92,9 @@ private[sql] class SparkSQLParser(fallback: String => LogicalPlan) extends Abstr
       case _ ~ dbName => ShowTablesCommand(dbName)
     }
 
+  private lazy val conf: Parser[LogicalPlan] =
+    CONF ^^^ ConfCommand
+
   private lazy val others: Parser[LogicalPlan] =
     wholeInput ^^ {
       case input => fallback(input)

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

@@ -91,7 +91,7 @@ case class SetCommand(
           "determining the number of reducers is not supported."
         throw new IllegalArgumentException(msg)
       } else {
-        sqlContext.setConf(SQLConf.SHUFFLE_PARTITIONS, value)
+        sqlContext.setConf(SQLConf.SHUFFLE_PARTITIONS.key, value)
         Seq(Row(s"${SQLConf.SHUFFLE_PARTITIONS}=$value"))
       }
 
@@ -254,3 +254,26 @@ case class ShowTablesCommand(databaseName: Option[String]) extends RunnableComma
     rows
   }
 }
+
+/**
+ * TODO
+ *
+ * :: DeveloperApi ::
+ */
+@DeveloperApi
+case object ConfCommand extends RunnableCommand {
+
+  // The result of SHOW TABLES has two columns, tableName and isTemporary.
+  override val output: Seq[Attribute] = {
+    val schema = StructType(
+      StructField("key", StringType, false) ::
+        StructField("default", StringType, false) ::
+        StructField("meaning", StringType, false) :: Nil)
+
+    schema.toAttributes
+  }
+
+  override def run(sqlContext: SQLContext): Seq[Row] = {
+    sqlContext.conf.getAllDefinedConfs.map { case (k, d, v) => Row(k, d, v) }
+  }
+}

sql/core/src/main/scala/org/apache/spark/sql/execution/commands.scala

@@ -46,7 +46,7 @@ package object debug {
    */
   implicit class DebugSQLContext(sqlContext: SQLContext) {
     def debug(): Unit = {
-      sqlContext.setConf(SQLConf.DATAFRAME_EAGER_ANALYSIS, "false")
+      sqlContext.setConf(SQLConf.DATAFRAME_EAGER_ANALYSIS.key, "false")
     }
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/debug/package.scala

@@ -114,8 +114,8 @@ private[sql] case class ParquetTableScan(
 
     // Tell FilteringParquetRowInputFormat whether it's okay to cache Parquet and FS metadata
     conf.set(
-      SQLConf.PARQUET_CACHE_METADATA,
-      sqlContext.getConf(SQLConf.PARQUET_CACHE_METADATA, "true"))
+      SQLConf.PARQUET_CACHE_METADATA.key,
+      sqlContext.getConf(SQLConf.PARQUET_CACHE_METADATA.key, "true"))
 
     val baseRDD =
       new org.apache.spark.rdd.NewHadoopRDD(
@@ -536,7 +536,7 @@ private[parquet] class FilteringParquetRowInputFormat
 
     import org.apache.spark.sql.parquet.FilteringParquetRowInputFormat.blockLocationCache
 
-    val cacheMetadata = configuration.getBoolean(SQLConf.PARQUET_CACHE_METADATA, true)
+    val cacheMetadata = configuration.getBoolean(SQLConf.PARQUET_CACHE_METADATA.key, true)
 
     val splits = mutable.ArrayBuffer.empty[ParquetInputSplit]
     val filter: Filter = ParquetInputFormat.getFilter(configuration)

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetTableOperations.scala

@@ -220,7 +220,7 @@ private[sql] class ParquetRelation2(
     }
 
     conf.setClass(
-      SQLConf.OUTPUT_COMMITTER_CLASS,
+      SQLConf.OUTPUT_COMMITTER_CLASS.key,
       committerClass,
       classOf[ParquetOutputCommitter])
 
@@ -259,7 +259,7 @@ private[sql] class ParquetRelation2(
       filters: Array[Filter],
       inputFiles: Array[FileStatus],
       broadcastedConf: Broadcast[SerializableWritable[Configuration]]): RDD[Row] = {
-    val useMetadataCache = sqlContext.getConf(SQLConf.PARQUET_CACHE_METADATA, "true").toBoolean
+    val useMetadataCache = sqlContext.getConf(SQLConf.PARQUET_CACHE_METADATA.key, "true").toBoolean
     val parquetFilterPushDown = sqlContext.conf.parquetFilterPushDown
     // Create the function to set variable Parquet confs at both driver and executor side.
     val initLocalJobFuncOpt =
@@ -498,7 +498,7 @@ private[sql] object ParquetRelation2 extends Logging {
       ParquetTypesConverter.convertToString(dataSchema.toAttributes))
 
     // Tell FilteringParquetRowInputFormat whether it's okay to cache Parquet and FS metadata
-    conf.set(SQLConf.PARQUET_CACHE_METADATA, useMetadataCache.toString)
+    conf.set(SQLConf.PARQUET_CACHE_METADATA.key, useMetadataCache.toString)
   }
 
   /** This closure sets input paths at the driver side. */

sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala

@@ -332,7 +332,7 @@ private[sql] abstract class BaseWriterContainer(
 
   private def newOutputCommitter(context: TaskAttemptContext): OutputCommitter = {
     val committerClass = context.getConfiguration.getClass(
-      SQLConf.OUTPUT_COMMITTER_CLASS, null, classOf[OutputCommitter])
+      SQLConf.OUTPUT_COMMITTER_CLASS.key, null, classOf[OutputCommitter])
 
     Option(committerClass).map { clazz =>
       logInfo(s"Using user defined output committer class ${clazz.getCanonicalName}")

sql/core/src/main/scala/org/apache/spark/sql/sources/commands.scala

@@ -38,7 +38,7 @@ class LocalSQLContext
   protected[sql] class SQLSession extends super.SQLSession {
     protected[sql] override lazy val conf: SQLConf = new SQLConf {
       /** Fewer partitions to speed up testing. */
-      override def numShufflePartitions: Int = this.getConf(SQLConf.SHUFFLE_PARTITIONS, "5").toInt
+      override def numShufflePartitions: Int = this.getConf(SQLConf.SHUFFLE_PARTITIONS, 5)
     }
   }