use sessionstate classloader in makeRDDforTable

Author: adrian-wang

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -103,17 +103,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
     }
   }
 
-  private var classLoader = Thread.currentThread().getContextClassLoader();
-  if (classLoader == null) {
-    classLoader = this.getClass.getClassLoader()
-  }
-
-  def getClassLoader: ClassLoader = classLoader
-
-  def setClassLoader(cl: ClassLoader) = {
-    classLoader = cl
-  }
-
   /**
    * Create a SparkContext that loads settings from system properties (for instance, when
    * launching with ./bin/spark-submit).

sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala

@@ -620,7 +620,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "load_file_with_space_in_the_name",
     "loadpart1",
     "louter_join_ppr",
-    "mapjoin_addjar",
     "mapjoin_distinct",
     "mapjoin_filter_on_outerjoin",
     "mapjoin_mapjoin",

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

@@ -77,7 +77,7 @@ class HadoopTableReader(
   override def makeRDDForTable(hiveTable: HiveTable): RDD[Row] =
     makeRDDForTable(
       hiveTable,
-      Class.forName(relation.tableDesc.getSerdeClassName, true, sc.sparkContext.getClassLoader)
+      Class.forName(relation.tableDesc.getSerdeClassName, true, sc.sessionState.getConf.getClassLoader)
         .asInstanceOf[Class[Deserializer]],
       filterOpt = None)
 

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala

@@ -17,10 +17,6 @@
 
 package org.apache.spark.sql.hive.execution
 
-import org.apache.commons.lang.StringUtils
-import org.apache.hadoop.hive.ql.exec.Utilities
-import org.apache.hadoop.hive.ql.parse.VariableSubstitution
-
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.EliminateSubQueries
 import org.apache.spark.sql.catalyst.util._
@@ -83,20 +79,7 @@ case class AddJar(path: String) extends RunnableCommand {
     val hiveContext = sqlContext.asInstanceOf[HiveContext]
     hiveContext.runSqlHive(s"ADD JAR $path")
     hiveContext.sparkContext.addJar(path)
-    try {
-      val loader = Thread.currentThread().getContextClassLoader()
-      val newLoader = Utilities.addToClassPath(loader, StringUtils.split(path, ","))
-      Thread.currentThread().setContextClassLoader(newLoader)
-      sqlContext.sparkContext.setClassLoader(newLoader)
-      logInfo("Added " + path + " to class path")
-      Seq(Row(0))
-    } catch {
-      case e: Throwable =>
-        logError("Unable to register " + path + "\nException: "
-          + e.getMessage() + "\n"
-          + org.apache.hadoop.util.StringUtils.stringifyException(e), e)
-        Seq(Row(1))
-    }
+    Seq(Row(0))
   }
 }
 

sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala

@@ -66,10 +66,6 @@ class TestHiveContext(sc: SparkContext) extends HiveContext(sc) {
   // without restarting the JVM.
   System.clearProperty("spark.hostPort")
   CommandProcessorFactory.clean(hiveconf)
-  System.setProperty("hive.version", HiveShim.version)
-  System.setProperty(
-    "maven.local.repository",
-    System.getProperty("user.home") + File.separator + ".m2" + File.separator + "repository")
 
   hiveconf.set("hive.plan.serialization.format", "javaXML")