Fix the potential bug of referencing FileSystem

Author: chenghao-intel

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

@@ -68,8 +68,6 @@ class HadoopTableReader(
     math.max(sc.hiveconf.getInt("mapred.map.tasks", 1), sc.sparkContext.defaultMinPartitions)
   }
 
-  @transient private lazy val fs = FileSystem.get(sc.hiveconf)
-
   // TODO: set aws s3 credentials.
 
   private val _broadcastedHiveConf =
@@ -220,6 +218,7 @@ class HadoopTableReader(
    * returned in a single, comma-separated string.
    */
   private def applyFilterIfNeeded(path: Path, filterOpt: Option[PathFilter]): Option[String] = {
+    val fs = path.getFileSystem(sc.hiveconf)
     if (fs.exists(path)) {
       // if the file exists
       filterOpt match {

sql/hive/src/test/scala/org/apache/spark/sql/hive/QueryPartitionSuite.scala

@@ -19,8 +19,8 @@ package org.apache.spark.sql.hive
 
 import org.scalatest.BeforeAndAfter
 
-import org.apache.spark.sql.hive.test.TestHive._
 import org.apache.spark.util.Utils
+import org.apache.spark.sql.hive.test.TestHive._
 import org.apache.spark.sql.hive.execution.HiveComparisonTest
 
 abstract class QueryPartitionSuite extends HiveComparisonTest with BeforeAndAfter {
@@ -78,4 +78,4 @@ class QueryPartitionSuite2 extends QueryPartitionSuite {
 
   createQueryTest("SPARK-5068 scan table with non-existed path",
     "select key,value from table_with_partition", false)
-}
+}