Changes for HiveQl and HiveContext.

Author: concretevitamin

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

@@ -113,7 +113,9 @@ case class NativeCommand(cmd: String) extends Command
 /**
  * Commands of the form "set key=value".
  */
-case class SetCommand(key: String, value: String) extends Command
+case class SetCommand(key: String, value: String) extends Command {
+  override def toString = s"<command> set $key=$value"
+}
 
 /**
  * Returned by a parser when the users only wants to see what query plan would be executed, without

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

@@ -30,16 +30,20 @@ import org.apache.hadoop.hive.ql.Driver
 import org.apache.hadoop.hive.ql.processors._
 import org.apache.hadoop.hive.ql.session.SessionState
 
-import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.analysis.{Analyzer, OverrideCatalog}
 import org.apache.spark.sql.catalyst.expressions.GenericRow
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, LowerCaseSchema}
-import org.apache.spark.sql.catalyst.plans.logical.{NativeCommand, ExplainCommand}
+import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.catalyst.types._
 import org.apache.spark.sql.execution._
+import org.apache.spark.sql.catalyst.types.StructType
+import org.apache.spark.sql.catalyst.plans.logical.NativeCommand
+import org.apache.spark.sql.catalyst.plans.logical.ExplainCommand
+import org.apache.spark.sql.catalyst.types.ArrayType
+import org.apache.spark.sql.catalyst.plans.logical.LowerCaseSchema
+import org.apache.spark.sql.catalyst.types.MapType
 
 /* Implicit conversions */
 import scala.collection.JavaConversions._
@@ -168,7 +172,7 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
   /**
    * Execute the command using Hive and return the results as a sequence. Each element
    * in the sequence is one row.
-   */
+   */                                                      `````
   protected def runHive(cmd: String, maxRows: Int = 1000): Seq[String] = {
     try {
       val cmd_trimmed: String = cmd.trim()
@@ -240,20 +244,30 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
     override lazy val optimizedPlan =
       optimizer(catalog.PreInsertionCasts(catalog.CreateTables(analyzed)))
 
-    override lazy val toRdd: RDD[Row] =
+    override lazy val toRdd: RDD[Row] = {
+
+      def processCmd(cmd: String): RDD[Row] = {
+        val output = runSqlHive(cmd)
+        if (output.size == 0) {
+          emptyResult
+        } else {
+          val asRows = output.map(r => new GenericRow(r.split("\t").asInstanceOf[Array[Any]]))
+          sparkContext.parallelize(asRows, 1)
+        }
+      }
+
       analyzed match {
+        case SetCommand(key, value) =>
+          logger.debug("inside Hive's toRdd -- matched SetCommand")
+          // Record the set command inside SQLConf, as well as have Hive execute it.
+          sqlConf.set(key, value)
+          processCmd(s"set $key=$value")
         case NativeCommand(cmd) =>
-          val output = runSqlHive(cmd)
-
-          if (output.size == 0) {
-            emptyResult
-          } else {
-            val asRows = output.map(r => new GenericRow(r.split("\t").asInstanceOf[Array[Any]]))
-            sparkContext.parallelize(asRows, 1)
-          }
+          processCmd(cmd)
         case _ =>
           executedPlan.execute().map(_.copy())
       }
+    }
 
     protected val primitiveTypes =
       Seq(StringType, IntegerType, LongType, DoubleType, FloatType, BooleanType, ByteType,
@@ -312,6 +326,7 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
     override def simpleString: String =
       logical match {
         case _: NativeCommand => "<Executed by Hive>"
+        case _: SetCommand => "<Set Command: Executed by Hive, and noted by SQLContext>"
         case _ => executedPlan.toString
       }
   }

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

@@ -208,7 +208,8 @@ private[hive] object HiveQl {
   def parseSql(sql: String): LogicalPlan = {
     try {
       if (sql.toLowerCase.startsWith("set")) {
-        NativeCommand(sql)
+        val kvPair = sql.drop(3).split("=")
+        SetCommand(kvPair(0).trim, kvPair(1).trim)
       } else if (sql.toLowerCase.startsWith("add jar")) {
         AddJar(sql.drop(8))
       } else if (sql.toLowerCase.startsWith("add file")) {