In a HiveContext, make SQLConf a subset of HiveConf.

- Any settings that go to SQLConf will go to HiveConf as well.
- Add tests for the Hive case.

Author: concretevitamin

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

@@ -26,7 +26,7 @@ import scala.collection.mutable
  */
 class SQLConf {
 
-  private val settings = new mutable.HashMap[String, String]()
+  protected val settings = new mutable.HashMap[String, String]()
 
   private[spark] def clear() {
     settings.clear()

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

@@ -33,8 +33,7 @@ case class SetCommandPhysical(
   extends LeafNode {
 
    def execute(): RDD[Row] = (key, value) match {
-     case (Some(k), Some(v)) =>
-       context.emptyResult
+     case (Some(k), Some(v)) => context.emptyResult
      case (Some(k), None) =>
        val resultString = context.sqlConf.getOption(k) match {
          case Some(v) => s"$k=$v"
@@ -48,8 +47,7 @@ case class SetCommandPhysical(
        }.toSeq
        // Assume config parameters can fit into one split (machine) ;)
        context.sparkContext.parallelize(rows, 1)
-     case _ =>
-       context.emptyResult
+     case _ => context.emptyResult
    }
 
    def output: Seq[Attribute] = Seq.empty // TODO: right thing?

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

@@ -336,7 +336,7 @@ class SQLQuerySuite extends QueryTest {
         (6, "f")))
   }
 
-  test("SET commands using sql()") {
+  test("SET commands semantics using sql()") {
     sqlConf.clear()
     val testKey = "test.key.0"
     val testVal = "test.val.0"
@@ -345,6 +345,7 @@ class SQLQuerySuite extends QueryTest {
     // "set" itself returns all config variables currently specified in SQLConf.
     assert(sql("set").collect().size == 0)
 
+    // "set key=val"
     sql(s"SET $testKey=$testVal")
     checkAnswer(
       sql("SET"),

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

@@ -59,10 +59,8 @@ class LocalHiveContext(sc: SparkContext) extends HiveContext(sc) {
 
   /** Sets up the system initially or after a RESET command */
   protected def configure() {
-    // TODO: refactor this so we can work with other databases.
-    runSqlHive(
-      s"set javax.jdo.option.ConnectionURL=jdbc:derby:;databaseName=$metastorePath;create=true")
-    runSqlHive("set hive.metastore.warehouse.dir=" + warehousePath)
+    sqlConf.set("javax.jdo.option.ConnectionURL", s"jdbc:derby:;databaseName=$metastorePath;create=true")
+    sqlConf.set("hive.metastore.warehouse.dir", warehousePath)
   }
 
   configure() // Must be called before initializing the catalog below.
@@ -134,12 +132,19 @@ 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`.
+   * Any properties set by sqlConf.set() or a SET command inside hql() or sql()
+   * will be set in the SQLConf, *as well as* getting set in the HiveConf.  In
+   * other words, the SQLConf properties will be a subset of the HiveConf properties
+   * throughout the life time of this session.
    */
   @transient protected[hive] lazy val hiveconf = new HiveConf(classOf[SessionState])
-  @transient override lazy val sqlConf: SQLConf = new SQLConf(hiveconf.getAllProperties)
-
+  @transient override lazy val sqlConf: SQLConf = new SQLConf(hiveconf.getAllProperties) {
+    override def set(key: String, value: String): SQLConf = {
+      hiveconf.set(key, value)
+      settings(key) = value
+      this
+    }
+  }
   @transient protected[hive] lazy val sessionState = new SessionState(hiveconf)
 
   sessionState.err = new PrintStream(outputBuffer, true, "UTF-8")

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.hive.execution
 
 import org.apache.spark.sql.hive.test.TestHive._
+import org.apache.spark.sql.Row
 
 /**
  * A set of test cases expressed in Hive QL that are not covered by the tests included in the hive distribution.
@@ -158,10 +159,17 @@ class HiveQuerySuite extends HiveComparisonTest {
     hql("SELECT * FROM src").toString
   }
 
+  test("HiveContext initializes its SQLConf to pick up its HiveConf's current params") {
+    import scala.collection.JavaConversions._ // implicits for java.util.Properties
+    val hiveconfSet: Set[(String, String)] = hiveconf.getAllProperties.toSeq.toSet
+    val sqlconfSet: Set[(String, String)] = sqlConf.getAll.toSet
+    assert(sqlconfSet.intersect(hiveconfSet) === sqlconfSet)
+    assert(sqlConf.getAll.size > 0, "The two confs should not be empty")
+  }
+
   test("parse HQL set commands") {
-    // Adapted from SQLConfSuite.
-    sqlConf.clear()
-    val testKey = "spark.sql.key"
+    // Adapted from its SQL counterpart.
+    val testKey = "spark.sql.key.usedfortestonly"
     val testVal = "val0,val_1,val2.3,my_table"
 
     hql(s"set $testKey=$testVal")
@@ -177,8 +185,60 @@ class HiveQuerySuite extends HiveComparisonTest {
 
     hql(s"set $testKey=")
     assert(sqlConf.get(testKey, "0") == "")
+  }
+
+  test("SET commands semantics for a HiveContext") {
+    // Adapted from its SQL counterpart.
+    val testKey = "spark.sql.key.usedfortestonly"
+    var testVal = "test.val.0"
+    val nonexistentKey = "nonexistent"
+    def fromRows(row: Array[Row]): Array[String] = row.map(_.getString(0))
+
+    sqlConf.clear()
+
+    // "set" itself returns all config variables currently specified in SQLConf.
+    assert(hql("set").collect().size == 0)
+
+    // "set key=val"
+    hql(s"SET $testKey=$testVal")
+    assert(fromRows(hql("SET").collect()) sameElements Array(s"$testKey=$testVal"))
+    assert(hiveconf.get(testKey, "") == testVal)
 
+    hql(s"SET ${testKey + testKey}=${testVal + testVal}")
+    assert(fromRows(hql("SET").collect()) sameElements
+      Array(
+        s"$testKey=$testVal",
+        s"${testKey + testKey}=${testVal + testVal}"))
+    assert(hiveconf.get(testKey + testKey, "") == testVal + testVal)
+
+    // "set key"
+    assert(fromRows(hql(s"SET $testKey").collect()) sameElements
+      Array(s"$testKey=$testVal"))
+    assert(fromRows(hql(s"SET $nonexistentKey").collect()) sameElements
+      Array(s"$nonexistentKey is undefined"))
+
+    // Assert that sql() should have the same effects as hql() by repeating the above using sql().
     sqlConf.clear()
+    assert(sql("set").collect().size == 0)
+
+    sql(s"SET $testKey=$testVal")
+    assert(fromRows(sql("SET").collect()) sameElements Array(s"$testKey=$testVal"))
+    assert(hiveconf.get(testKey, "") == testVal)
+
+    sql(s"SET ${testKey + testKey}=${testVal + testVal}")
+    assert(fromRows(sql("SET").collect()) sameElements
+      Array(
+        s"$testKey=$testVal",
+        s"${testKey + testKey}=${testVal + testVal}"))
+    assert(hiveconf.get(testKey + testKey, "") == testVal + testVal)
+
+    assert(fromRows(sql(s"SET $testKey").collect()) sameElements
+      Array(s"$testKey=$testVal"))
+    assert(fromRows(sql(s"SET $nonexistentKey").collect()) sameElements
+      Array(s"$nonexistentKey is undefined"))
   }
 
+  // Put tests that depend on specific Hive settings before these last two test,
+  // since they modify /clear stuff.
+
 }