pass the user parameters to metaDataHive

Author: zhichao-li

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala

@@ -263,6 +263,7 @@ private[hive] object SparkSQLCLIDriver extends Logging {
 
 }
 
+
 private[hive] class SparkSQLCLIDriver extends CliDriver with Logging {
   private val sessionState = SessionState.get().asInstanceOf[CliSessionState]
 

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLDriver.scala

@@ -18,7 +18,6 @@
 package org.apache.spark.sql.hive.thriftserver
 
 import java.util.{Arrays, ArrayList => JArrayList, List => JList}
-import org.apache.log4j.LogManager
 import org.apache.spark.sql.AnalysisException
 
 import scala.collection.JavaConverters._

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLEnv.scala

@@ -21,8 +21,10 @@ import java.io.PrintStream
 
 import scala.collection.JavaConverters._
 
+import org.apache.hadoop.hive.ql.session.SessionState
+
 import org.apache.spark.scheduler.StatsReportListener
-import org.apache.spark.sql.hive.HiveContext
+import org.apache.spark.sql.hive.{UserInput, HiveContext}
 import org.apache.spark.{Logging, SparkConf, SparkContext}
 import org.apache.spark.util.Utils
 
@@ -55,7 +57,10 @@ private[hive] object SparkSQLEnv extends Logging {
 
       sparkContext = new SparkContext(sparkConf)
       sparkContext.addSparkListener(new StatsReportListener())
-      hiveContext = new HiveContext(sparkContext)
+
+      val sessionState = SessionState.get()
+      hiveContext = new HiveContext(sparkContext,
+        Some(UserInput(sessionState.getIsSilent, sessionState.getIsVerbose)))
 
       hiveContext.metadataHive.setOut(new PrintStream(System.out, true, "UTF-8"))
       hiveContext.metadataHive.setInfo(new PrintStream(System.err, true, "UTF-8"))

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

@@ -56,6 +56,10 @@ import org.apache.spark.unsafe.types.UTF8String
 import org.apache.spark.util.Utils
 import org.apache.spark.{Logging, SparkContext}
 
+/**
+ * Use to encapsulate the user input parameters from spark-sql CLI.
+ */
+private[hive] case class UserInput (isSilent: Boolean, isVerbose: Boolean)
 
 /**
  * This is the HiveQL Dialect, this dialect is strongly bind with HiveContext
@@ -93,13 +97,19 @@ class HiveContext private[hive](
     listener: SQLListener,
     @transient private val execHive: ClientWrapper,
     @transient private val metaHive: ClientInterface,
-    isRootContext: Boolean)
+    isRootContext: Boolean,
+    userInput: Option[UserInput] = None)
   extends SQLContext(sc, cacheManager, listener, isRootContext) with Logging {
   self =>
 
   def this(sc: SparkContext) = {
-    this(sc, new CacheManager, SQLContext.createListenerAndUI(sc), null, null, true)
+    this(sc, new CacheManager, SQLContext.createListenerAndUI(sc), null, null, true, None)
+  }
+
+  def this(sc: SparkContext, userInput: Option[UserInput]) = {
+    this(sc, new CacheManager, SQLContext.createListenerAndUI(sc), null, null, true, userInput)
   }
+
   def this(sc: JavaSparkContext) = this(sc.sc)
 
   import org.apache.spark.sql.hive.HiveContext._
@@ -215,7 +225,7 @@ class HiveContext private[hive](
       config = newTemporaryConfiguration(useInMemoryDerby = true),
       isolationOn = false,
       baseClassLoader = Utils.getContextOrSparkClassLoader)
-    loader.createClient().asInstanceOf[ClientWrapper]
+    loader.createClient(userInput).asInstanceOf[ClientWrapper]
   }
 
   /**
@@ -324,7 +334,7 @@ class HiveContext private[hive](
         barrierPrefixes = hiveMetastoreBarrierPrefixes,
         sharedPrefixes = hiveMetastoreSharedPrefixes)
     }
-    isolatedLoader.createClient()
+    isolatedLoader.createClient(userInput)
   }
 
   protected[sql] override def parseSql(sql: String): LogicalPlan = {
@@ -656,7 +666,6 @@ class HiveContext private[hive](
   }
 }
 
-
 private[hive] object HiveContext {
   /** The version of hive used internally by Spark SQL. */
   val hiveExecutionVersion: String = "1.2.1"

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/ClientWrapper.scala

@@ -36,6 +36,7 @@ import org.apache.hadoop.security.UserGroupInformation
 import org.apache.hadoop.util.VersionInfo
 
 import org.apache.spark.{SparkConf, SparkException, Logging}
+import org.apache.spark.sql.hive.UserInput
 import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.execution.QueryExecutionException
 import org.apache.spark.util.{CircularBuffer, Utils}
@@ -61,10 +62,10 @@ private[hive] class ClientWrapper(
     override val version: HiveVersion,
     config: Map[String, String],
     initClassLoader: ClassLoader,
-    val clientLoader: IsolatedClientLoader)
+    val clientLoader: IsolatedClientLoader,
+    userInput: Option[UserInput])
   extends ClientInterface
   with Logging {
-
   overrideHadoopShims()
 
   // !! HACK ALERT !!
@@ -194,6 +195,10 @@ private[hive] class ClientWrapper(
       SessionState.start(state)
       state.out = new PrintStream(outputBuffer, true, "UTF-8")
       state.err = new PrintStream(outputBuffer, true, "UTF-8")
+      userInput.foreach { input =>
+        state.setIsSilent(input.isSilent)
+        state.setIsVerbose(input.isVerbose)
+      }
       state
     } finally {
       Thread.currentThread().setContextClassLoader(original)
@@ -494,7 +499,7 @@ private[hive] class ClientWrapper(
           results
 
         case _ =>
-          if (state.out != null) {
+          if (state.out != null && !state.getIsSilent) {
             // scalastyle:off println
             state.out.println(tokens(0) + " " + cmd_1)
             // scalastyle:on println
@@ -582,7 +587,7 @@ private[hive] class ClientWrapper(
   }
 
   def newSession(): ClientWrapper = {
-    clientLoader.createClient().asInstanceOf[ClientWrapper]
+    clientLoader.createClient(userInput).asInstanceOf[ClientWrapper]
   }
 
   def reset(): Unit = withHiveState {

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala

@@ -30,7 +30,7 @@ import org.apache.commons.io.{FileUtils, IOUtils}
 import org.apache.spark.Logging
 import org.apache.spark.deploy.SparkSubmitUtils
 import org.apache.spark.sql.catalyst.util.quietly
-import org.apache.spark.sql.hive.HiveContext
+import org.apache.spark.sql.hive.{UserInput, HiveContext}
 import org.apache.spark.util.{MutableURLClassLoader, Utils}
 
 /** Factory for `IsolatedClientLoader` with specific versions of hive. */
@@ -233,9 +233,9 @@ private[hive] class IsolatedClientLoader(
   }
 
   /** The isolated client interface to Hive. */
-  private[hive] def createClient(): ClientInterface = {
+  private[hive] def createClient(userInput: Option[UserInput] = None): ClientInterface = {
     if (!isolationOn) {
-      return new ClientWrapper(version, config, baseClassLoader, this)
+      return new ClientWrapper(version, config, baseClassLoader, this, userInput)
     }
     // Pre-reflective instantiation setup.
     logDebug("Initializing the logger to avoid disaster...")
@@ -246,7 +246,7 @@ private[hive] class IsolatedClientLoader(
       classLoader
         .loadClass(classOf[ClientWrapper].getName)
         .getConstructors.head
-        .newInstance(version, config, classLoader, this)
+        .newInstance(version, config, classLoader, this, userInput)
         .asInstanceOf[ClientInterface]
     } catch {
       case e: InvocationTargetException =>