Clean up semantics of several cases of SET.

- Add planner & physical plan for SET so that the pipeline propagates it
  cleanly.
- For HiveContext: make SQLConf and HiveConf equal after initialization
  of the context.

Author: concretevitamin

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

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql
 
+import java.util.Properties
+
 import scala.collection.mutable
 
 /**
@@ -30,6 +32,12 @@ class SQLConf {
     settings.clear()
   }
 
+  def this(props: Properties) = {
+    this()
+    import scala.collection.JavaConversions._ // implicits for java.util.Properties
+    props.foreach { case (k, v) => this.settings(k) = v }
+  }
+
   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

@@ -58,7 +58,7 @@ class SQLContext(@transient val sparkContext: SparkContext)
   self =>
 
   @transient
-  val sqlConf: SQLConf = new SQLConf
+  lazy val sqlConf: SQLConf = new SQLConf
 
   @transient
   protected[sql] lazy val catalog: Catalog = new SimpleCatalog
@@ -193,9 +193,10 @@ class SQLContext(@transient val sparkContext: SparkContext)
   protected[sql] class SparkPlanner extends SparkStrategies {
     val sparkContext = self.sparkContext
 
-    def sqlConf = self.sqlConf
+    val sqlConf = self.sqlConf
 
     val strategies: Seq[Strategy] =
+      SetCommandStrategy(self) ::
       TakeOrdered ::
       PartialAggregation ::
       HashJoin ::
@@ -250,7 +251,7 @@ class SQLContext(@transient val sparkContext: SparkContext)
   protected[sql] val planner = new SparkPlanner
 
   @transient
-  protected lazy val emptyResult =
+  protected[sql] lazy val emptyResult =
     sparkContext.parallelize(Seq(new GenericRow(Array[Any]()): Row), 1)
 
   /**
@@ -281,8 +282,13 @@ class SQLContext(@transient val sparkContext: SparkContext)
     lazy val toRdd: RDD[Row] = {
       logical match {
         case SetCommand(key, value) =>
-          sqlConf.set(key, value)
-          emptyResult // TODO: should this return something else? Single row consisting of one 0?
+          if (key.isDefined && value.isDefined) {
+            sqlConf.set(key.get, value.get)
+          }
+          // It doesn't matter what we return here, since toRdd is used
+          // to force the evaluation happen eagerly.  To query the results,
+          // one must use SchemaRDD operations to extract them.
+          emptyResult
         case _ => executedPlan.execute()
       }
     }

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

@@ -217,4 +217,13 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
       case _ => Nil
     }
   }
+
+  case class SetCommandStrategy(context: SQLContext) extends Strategy {
+    def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+      case logical.SetCommand(key, value) =>
+        Seq(execution.SetCommandPhysical(key, value, context))
+      case _ => Nil
+    }
+  }
+
 }

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

@@ -49,6 +49,8 @@ class SQLConfSuite extends QueryTest {
     conf.clear()
   }
 
+  // TODO: this is more suitable for a "SetCommandSuite"? Or put them inside the query suite?
+
   test("parse SQL set commands") {
     sql(s"set $testKey=$testVal")
     assert(conf.get(testKey, testVal + "_") == testVal)
@@ -64,22 +66,26 @@ class SQLConfSuite extends QueryTest {
     sql(s"set $key=$vs")
     assert(conf.get(key, "0") == vs)
 
-    sql(s"set$key=")
+    sql(s"set $key=")
     assert(conf.get(key, "0") == "")
 
     conf.clear()
   }
 
   test("set itself returns all config variables currently specified in Hive or SQLConf") {
-    assert(sql("set").collect() === Seq(""))
+    def fromRows(row: Array[Row]): Array[String] = row.map(_.getString(0))
+
+    assert(sql("set").collect().size == 0)
 
     sql(s"SET $testKey=$testVal")
-    assert(sql("set").collect() === Seq(s"$testKey=$testVal"))
+    assert(fromRows(sql("set").collect()) sameElements Array(s"$testKey=$testVal"))
 
     sql(s"SET ${testKey + testKey}=${testVal + testVal}")
-    // TODO: should this be 1-elem Seq or 2-elem?
-    assert(sql("set").collect() ===
-      Seq(s"$testKey=$testVal\n${testKey + testKey}=${testVal + testVal}"))
+    assert(fromRows(sql("set").collect()) sameElements
+      Array(
+        s"$testKey=$testVal",
+        s"${testKey + testKey}=${testVal + testVal}"
+      ))
   }
 
 }

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

@@ -133,7 +133,13 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
     }
   }
 
+  /**
+   * Contract: after initialization of this HiveContext, these two confs should
+   * contain exactly the same key-value pairs throughout the life time of `this`.
+   */
   @transient protected[hive] lazy val hiveconf = new HiveConf(classOf[SessionState])
+  @transient override lazy val sqlConf: SQLConf = new SQLConf(hiveconf.getAllProperties)
+
   @transient protected[hive] lazy val sessionState = new SessionState(hiveconf)
 
   sessionState.err = new PrintStream(outputBuffer, true, "UTF-8")
@@ -222,6 +228,7 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
     val hiveContext = self
 
     override val strategies: Seq[Strategy] = Seq(
+      SetCommandStrategy(self),
       TakeOrdered,
       ParquetOperations,
       HiveTableScans,
@@ -259,8 +266,14 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
       analyzed match {
         case SetCommand(key, value) =>
           // Record the set command inside SQLConf, as well as have Hive execute it.
-          sqlConf.set(key, value)
-          processCmd(s"set $key=$value")
+          if (key.isDefined && value.isDefined) {
+            sqlConf.set(key.get, value.get)
+            processCmd(s"SET $key=$value")
+          }
+          // Only the above case needs to be executed in Hive eagerly (i.e. now).
+          // The other cases will be taken care of when the actual results are
+          // being extracted.
+          emptyResult
         case NativeCommand(cmd) =>
           processCmd(cmd)
         case _ =>