[SPARK-5846] Correctly set job description and pool for SQL jobs

This is apache#4630 but modified for the 1.2 branch, because I'm guessing it makes sense to fix this issue in that branch (again, unless I missed something obvious here...)

Author: Kay Ousterhout <[email protected]>

Closes apache#4631 from kayousterhout/SPARK-5846_1.2.1 and squashes the following commits:

ffe8ff2 [Kay Ousterhout] [SPARK-5846] Correctly set job description and pool for SQL jobs

Author: kayousterhout

sql/hive-thriftserver/v0.12.0/src/main/scala/org/apache/spark/sql/hive/thriftserver/Shim12.scala

@@ -186,6 +186,11 @@ private[hive] class SparkExecuteStatementOperation(
   def run(): Unit = {
     logInfo(s"Running query '$statement'")
     setState(OperationState.RUNNING)
+    val groupId = round(random * 1000000).toString
+    hiveContext.sparkContext.setJobGroup(groupId, statement)
+    sessionToActivePool.get(parentSession).foreach { pool =>
+      hiveContext.sparkContext.setLocalProperty("spark.scheduler.pool", pool)
+    }
     try {
       result = hiveContext.sql(statement)
       logDebug(result.queryExecution.toString())
@@ -196,11 +201,6 @@ private[hive] class SparkExecuteStatementOperation(
         case _ =>
       }
 
-      val groupId = round(random * 1000000).toString
-      hiveContext.sparkContext.setJobGroup(groupId, statement)
-      sessionToActivePool.get(parentSession).foreach { pool =>
-        hiveContext.sparkContext.setLocalProperty("spark.scheduler.pool", pool)
-      }
       iter = {
         val useIncrementalCollect =
           hiveContext.getConf("spark.sql.thriftServer.incrementalCollect", "false").toBoolean

sql/hive-thriftserver/v0.13.1/src/main/scala/org/apache/spark/sql/hive/thriftserver/Shim13.scala

@@ -157,6 +157,11 @@ private[hive] class SparkExecuteStatementOperation(
   def run(): Unit = {
     logInfo(s"Running query '$statement'")
     setState(OperationState.RUNNING)
+    val groupId = round(random * 1000000).toString
+    hiveContext.sparkContext.setJobGroup(groupId, statement)
+    sessionToActivePool.get(parentSession).foreach { pool =>
+      hiveContext.sparkContext.setLocalProperty("spark.scheduler.pool", pool)
+    }
     try {
       result = hiveContext.sql(statement)
       logDebug(result.queryExecution.toString())
@@ -167,11 +172,6 @@ private[hive] class SparkExecuteStatementOperation(
         case _ =>
       }
 
-      val groupId = round(random * 1000000).toString
-      hiveContext.sparkContext.setJobGroup(groupId, statement)
-      sessionToActivePool.get(parentSession).foreach { pool =>
-        hiveContext.sparkContext.setLocalProperty("spark.scheduler.pool", pool)
-      }
       iter = {
         val useIncrementalCollect =
           hiveContext.getConf("spark.sql.thriftServer.incrementalCollect", "false").toBoolean