From ba14f283d8a68cc90ee828eee5dcc9a95e6dd5a4 Mon Sep 17 00:00:00 2001
From: Zhan Zhang <zhazhan@gmail.com>
Date: Fri, 8 Aug 2014 10:47:18 -0700
Subject: [PATCH 1/3] test

---
 pom.xml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pom.xml b/pom.xml
index 556b9da3d6d90..d482b84511da9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1,4 +1,4 @@
-<?xml version="1.0" encoding="UTF-8"?>
+test<?xml version="1.0" encoding="UTF-8"?>
 <!--
   ~ Licensed to the Apache Software Foundation (ASF) under one or more
   ~ contributor license agreements.  See the NOTICE file distributed with

From f6a8a40baa588539727d2cd450f6b638756b9b29 Mon Sep 17 00:00:00 2001
From: Zhan Zhang <zhazhan@gmail.com>
Date: Fri, 8 Aug 2014 10:57:06 -0700
Subject: [PATCH 2/3] revert

---
 pom.xml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pom.xml b/pom.xml
index d482b84511da9..556b9da3d6d90 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1,4 +1,4 @@
-test<?xml version="1.0" encoding="UTF-8"?>
+<?xml version="1.0" encoding="UTF-8"?>
 <!--
   ~ Licensed to the Apache Software Foundation (ASF) under one or more
   ~ contributor license agreements.  See the NOTICE file distributed with

From 53a26ae4c3f671014e5d020a1fc6d3502003f598 Mon Sep 17 00:00:00 2001
From: Zhan Zhang <zhazhan@gmail.com>
Date: Mon, 27 Oct 2014 15:09:52 -0700
Subject: [PATCH 3/3] clean up session state in HiveContext. Currently the
 sessionState in HiveContext is lazy evaluated, and to start the session,
 SessionState.start() has to be invoked to initialize it. Using ThreadLocal
 variable natively in hive to track the availability of SessionState and
 initialize it if required. It avoid multiple thread sharing the same
 SessionState or single thread invoking SessionState.start with the same
 SessionState multiple times, which seems to cause unexpected behavior.

---
 .../sql/hive/thriftserver/SparkSQLEnv.scala   |  4 ++--
 .../apache/spark/sql/hive/HiveContext.scala   | 22 ++++++++++++++-----
 .../spark/sql/hive/HiveMetastoreCatalog.scala |  5 +++--
 3 files changed, 21 insertions(+), 10 deletions(-)

diff --git a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLEnv.scala b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLEnv.scala
index 2136a2ea63543..41fa9168d81f9 100644
--- a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLEnv.scala
+++ b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLEnv.scala
@@ -39,8 +39,8 @@ private[hive] object SparkSQLEnv extends Logging {
       sparkContext.addSparkListener(new StatsReportListener())
 
       hiveContext = new HiveContext(sparkContext) {
-        @transient override lazy val sessionState = {
-          val state = SessionState.get()
+        @transient lazy val sessionState = {
+          val state = getSessionState()
           setConf(state.getConf.getAllProperties)
           state
         }
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
index fad4091d48a89..005b8a0649b83 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
@@ -230,13 +230,21 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
    * in the HiveConf.
    */
   @transient lazy val hiveconf = new HiveConf(classOf[SessionState])
-  @transient protected[hive] lazy val sessionState = {
-    val ss = new SessionState(hiveconf)
-    setConf(hiveconf.getAllProperties)  // Have SQLConf pick up the initial set of HiveConf.
-    SessionState.start(ss)
-    ss.err = new PrintStream(outputBuffer, true, "UTF-8")
-    ss.out = new PrintStream(outputBuffer, true, "UTF-8")
 
+  /**
+   * If the thread local sessionstate is not set, start a new SessionState
+   * SessionState.start will put ss to thread local
+   * @return
+   */
+  def getSessionState() = {
+    var ss = SessionState.get
+    if (ss == null) {
+      ss = new SessionState(hiveconf)
+      setConf(hiveconf.getAllProperties) // Have SQLConf pick up the initial set of HiveConf.
+      SessionState.start(ss)
+      ss.err = new PrintStream(outputBuffer, true, "UTF-8")
+      ss.out = new PrintStream(outputBuffer, true, "UTF-8")
+    }
     ss
   }
 
@@ -283,6 +291,8 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
    */
   protected def runHive(cmd: String, maxRows: Int = 1000): Seq[String] = {
     try {
+      // invoke getSessionState to initialize session state if not already done
+      val sessionState = getSessionState()
       val cmd_trimmed: String = cmd.trim()
       val tokens: Array[String] = cmd_trimmed.split("\\s+")
       val cmd_1: String = cmd_trimmed.substring(tokens(0).length()).trim()
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
index 04c48c385966e..239364270d3ed 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
@@ -51,7 +51,8 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
       tableName: String,
       alias: Option[String]): LogicalPlan = synchronized {
     val (databaseName, tblName) = processDatabaseAndTableName(
-                                    db.getOrElse(hive.sessionState.getCurrentDatabase), tableName)
+                                    db.getOrElse(hive.getSessionState
+                                      .getCurrentDatabase), tableName)
     val table = client.getTable(databaseName, tblName)
     val partitions: Seq[Partition] =
       if (table.isPartitioned) {
@@ -112,7 +113,7 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
 
       case CreateTableAsSelect(db, tableName, child) =>
         val (dbName, tblName) = processDatabaseAndTableName(db, tableName)
-        val databaseName = dbName.getOrElse(hive.sessionState.getCurrentDatabase)
+        val databaseName = dbName.getOrElse(hive.getSessionState.getCurrentDatabase)
 
         CreateTableAsSelect(Some(databaseName), tableName, child)
     }