Finish up SQLConf patch.

Author: concretevitamin

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SqlParser.scala

@@ -131,6 +131,7 @@ class SqlParser extends StandardTokenParsers with PackratParsers {
   protected val OUTER = Keyword("OUTER")
   protected val RIGHT = Keyword("RIGHT")
   protected val SELECT = Keyword("SELECT")
+  protected val SET = Keyword("SET")
   protected val STRING = Keyword("STRING")
   protected val SUM = Keyword("SUM")
   protected val TRUE = Keyword("TRUE")
@@ -168,11 +169,21 @@ class SqlParser extends StandardTokenParsers with PackratParsers {
     }
   }
 
-  protected lazy val query: Parser[LogicalPlan] =
-    select * (
-      UNION ~ ALL ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Union(q1, q2) } |
-      UNION ~ opt(DISTINCT) ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Distinct(Union(q1, q2)) }
-    ) | insert
+  protected lazy val query: Parser[LogicalPlan] = (
+      setCommand
+    | select * (
+        UNION ~ ALL ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Union(q1, q2) } |
+        UNION ~ opt(DISTINCT) ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Distinct(Union(q1, q2)) }
+      )
+    | insert
+  )
+
+  protected lazy val setCommand: Parser[LogicalPlan] = {
+    val keyVal = ident | numericLit | stringLit
+    (SET ~> keyVal <~ "=") ~ keyVal <~ opt(";") ^^ {
+      case key ~ value => SetCommand(key, value)
+    }
+  }
 
   protected lazy val select: Parser[LogicalPlan] =
     SELECT ~> opt(DISTINCT) ~ projections ~

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala

@@ -110,6 +110,11 @@ abstract class Command extends LeafNode {
  */
 case class NativeCommand(cmd: String) extends Command
 
+/**
+ * Commands of the form "set key=value".
+ */
+case class SetCommand(key: String, value: String) extends Command
+
 /**
  * Returned by a parser when the users only wants to see what query plan would be executed, without
  * actually performing the execution.

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

@@ -26,6 +26,10 @@ class SQLConf {
 
   private val settings = new mutable.HashMap[String, String]()
 
+  private[spark] def clear() {
+    settings.clear()
+  }
+
   def set(key: String, value: String): SQLConf = {
     if (key == null) {
       throw new NullPointerException("null key")

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

@@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.{ScalaReflection, dsl}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.types._
 import org.apache.spark.sql.catalyst.optimizer.Optimizer
-import org.apache.spark.sql.catalyst.plans.logical.{Subquery, LogicalPlan}
+import org.apache.spark.sql.catalyst.plans.logical.{SetCommand, Subquery, LogicalPlan}
 import org.apache.spark.sql.catalyst.rules.RuleExecutor
 
 import org.apache.spark.sql.columnar.InMemoryColumnarTableScan
@@ -40,7 +40,6 @@ import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.SparkStrategies
 
 import org.apache.spark.sql.parquet.ParquetRelation
-import org.apache.spark.sql.SQLConf
 
 /**
  * :: AlphaComponent ::
@@ -250,6 +249,10 @@ class SQLContext(@transient val sparkContext: SparkContext)
   @transient
   protected[sql] val planner = new SparkPlanner
 
+  @transient
+  protected lazy val emptyResult =
+    sparkContext.parallelize(Seq(new GenericRow(Array[Any]()): Row), 1)
+
   /**
    * Prepares a planned SparkPlan for execution by binding references to specific ordinals, and
    * inserting shuffle operations as needed.
@@ -275,7 +278,14 @@ class SQLContext(@transient val sparkContext: SparkContext)
     lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan)
 
     /** Internal version of the RDD. Avoids copies and has no schema */
-    lazy val toRdd: RDD[Row] = executedPlan.execute()
+    lazy val toRdd: RDD[Row] = {
+      logical match {
+        case SetCommand(key, value) =>
+          sqlConf.set(key, value)
+          emptyResult
+        case _ => executedPlan.execute()
+      }
+    }
 
     protected def stringOrError[A](f: => A): String =
       try f.toString catch { case e: Throwable => e.toString }

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

@@ -20,19 +20,15 @@ package org.apache.spark.sql
 import org.apache.spark.sql.test._
 
 /* Implicits */
-//import TestSQLContext._
-//import TestData._
+import TestSQLContext._
 
 class SQLConfSuite extends QueryTest {
 
-  // Make sure the tables are loaded.
-//  TestData
-
-  test("basic setting and getting in SQLConf") {
-    val conf = TestSQLContext.sqlConf
-    val testKey = "test.key"
-    val testVal = "test.val"
+  val conf = TestSQLContext.sqlConf
+  val testKey = "test.key.0"
+  val testVal = "test.val.0"
 
+  test("programmatic ways of basic setting and getting") {
     assert(conf != null)
     assert(conf.getOption(testKey).isEmpty)
     assert(conf.getAll.toSet === Set())
@@ -49,6 +45,19 @@ class SQLConfSuite extends QueryTest {
     assert(TestSQLContext.sqlConf.get(testKey, testVal + "_") == testVal)
     assert(TestSQLContext.sqlConf.getOption(testKey) == Some(testVal))
     assert(TestSQLContext.sqlConf.contains(testKey))
+
+    conf.clear()
+  }
+
+  test("SQLConf picks up SQL set commands") {
+    sql(s"set $testKey=$testVal")
+    assert(conf.get(testKey, testVal + "_") == testVal)
+    assert(TestSQLContext.sqlConf.get(testKey, testVal + "_") == testVal)
+
+    sql("set mapred.reduce.tasks=20")
+    assert(conf.get("mapred.reduce.tasks", "0") == "20")
+    sql("set mapred.reduce.tasks = 40")
+    assert(conf.get("mapred.reduce.tasks", "0") == "40")
   }
 
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala

@@ -234,10 +234,6 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
   @transient
   override protected[sql] val planner = hivePlanner
 
-  @transient
-  protected lazy val emptyResult =
-    sparkContext.parallelize(Seq(new GenericRow(Array[Any]()): Row), 1)
-
   /** Extends QueryExecution with hive specific features. */
   protected[sql] abstract class QueryExecution extends super.QueryExecution {
     // TODO: Create mixin for the analyzer instead of overriding things here.