First cut at SQLConf inside SQLContext.

Author: concretevitamin

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

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import scala.collection.mutable
+
+/**
+ * SQLConf holds potentially query-dependent, mutable config parameters and hints.
+ */
+class SQLConf {
+
+  private val settings = new mutable.HashMap[String, String]()
+
+  def set(key: String, value: String): SQLConf = {
+    if (key == null) {
+      throw new NullPointerException("null key")
+    }
+    if (value == null) {
+      throw new NullPointerException("null value")
+    }
+    settings(key) = value
+    this
+  }
+
+  def get(key: String): String = {
+    settings.getOrElse(key, throw new NoSuchElementException(key))
+  }
+
+  def get(key: String, defaultValue: String): String = {
+    settings.getOrElse(key, defaultValue)
+  }
+
+  def getAll: Array[(String, String)] = settings.clone().toArray
+
+  def getOption(key: String): Option[String] = {
+    settings.get(key)
+  }
+
+  def contains(key: String): Boolean = settings.contains(key)
+
+  def toDebugString: String = {
+    settings.toArray.sorted.map{ case (k, v) => s"$k=$v" }.mkString("\n")
+  }
+
+}

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

@@ -40,6 +40,7 @@ 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 ::
@@ -57,6 +58,9 @@ class SQLContext(@transient val sparkContext: SparkContext)
 
   self =>
 
+  @transient
+  val sqlConf: SQLConf = new SQLConf
+
   @transient
   protected[sql] lazy val catalog: Catalog = new SimpleCatalog
   @transient
@@ -190,6 +194,8 @@ class SQLContext(@transient val sparkContext: SparkContext)
   protected[sql] class SparkPlanner extends SparkStrategies {
     val sparkContext = self.sparkContext
 
+    def sqlConf = self.sqlConf
+
     val strategies: Seq[Strategy] =
       TakeOrdered ::
       PartialAggregation ::

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

@@ -29,6 +29,10 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
   self: SQLContext#SparkPlanner =>
 
   object HashJoin extends Strategy with PredicateHelper {
+
+//    var broadCastTables =
+//    sqlConf.get("spark.sql.hints.broadcastTables", "").split(",")
+
     def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
       // Find inner joins where at least some predicates can be evaluated by matching hash keys
       // using the HashFilteredJoin pattern.

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

@@ -0,0 +1,54 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements.  See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License.  You may obtain a copy of the License at
+*
+*    http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.test._
+
+/* Implicits */
+//import TestSQLContext._
+//import TestData._
+
+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"
+
+    assert(conf != null)
+    assert(conf.getOption(testKey).isEmpty)
+    assert(conf.getAll.toSet === Set())
+
+    conf.set(testKey, testVal)
+    assert(conf.get(testKey) == testVal)
+    assert(conf.get(testKey, testVal + "_") == testVal)
+    assert(conf.getOption(testKey) == Some(testVal))
+    assert(conf.contains(testKey))
+
+    // Tests SQLConf as accessed from a SQLContext is mutable after
+    // the latter is initialized, unlike SparkConf inside a SparkContext.
+    assert(TestSQLContext.sqlConf.get(testKey) == testVal)
+    assert(TestSQLContext.sqlConf.get(testKey, testVal + "_") == testVal)
+    assert(TestSQLContext.sqlConf.getOption(testKey) == Some(testVal))
+    assert(TestSQLContext.sqlConf.contains(testKey))
+  }
+
+}

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

@@ -30,8 +30,8 @@ class PlannerSuite extends FunSuite {
   test("unions are collapsed") {
     val query = testData.unionAll(testData).unionAll(testData).logicalPlan
     val planned = BasicOperators(query).head
-    val logicalUnions = query collect { case u: logical.Union => u}
-    val physicalUnions = planned collect { case u: execution.Union => u}
+    val logicalUnions = query collect { case u: logical.Union => u }
+    val physicalUnions = planned collect { case u: execution.Union => u }
 
     assert(logicalUnions.size === 2)
     assert(physicalUnions.size === 1)