[CARMEL-7557][CARMEL-4814] Use proper file system (apache#217)

Author: fenzhu

sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala

@@ -21,7 +21,8 @@ import java.util.concurrent.TimeUnit._
 import java.util.regex.Pattern
 
 import org.apache.commons.lang3.StringUtils
-import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileStatus, Path}
 
 import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
@@ -327,9 +328,10 @@ trait FileSourceScanLike extends DataSourceScanExec {
     if (sourcePaths.forall(p => !pattern.matcher(p).find())) return partitions
 
     val localPaths = sourcePaths.map(_.replaceFirst(remoteHdfsRegex, localHdfsPrefix))
-    def listFilesIfExists(paths: Seq[String], fileSystem: FileSystem): Seq[FileStatus] = {
+    def listFilesIfExists(paths: Seq[String], conf: Configuration): Seq[FileStatus] = {
       paths.flatMap { path =>
-        val p = new Path(path).makeQualified(fileSystem.getUri, fileSystem.getWorkingDirectory)
+        val p = new Path(path)
+        val fileSystem = p.getFileSystem(conf)
         if (fileSystem.exists(p)) {
           fileSystem.listStatus(p)
         } else {
@@ -350,14 +352,13 @@ trait FileSourceScanLike extends DataSourceScanExec {
         sparkContext.setJobDescription("List local cache directories.")
         val numParallelism = Math.min(localPaths.size, parallelPartitionDiscoveryParallelism)
         sparkContext.parallelize(localPaths, numParallelism).mapPartitions { lps =>
-          val fs = new Path(lps.toSeq.head).getFileSystem(serializableConfiguration.value)
-          listFilesIfExists(lps.toSeq, fs).iterator
+          listFilesIfExists(lps.toSeq, serializableConfiguration.value).iterator
         }.collect().toSeq
       } finally {
         sparkContext.setJobDescription(previousJobDescription)
       }
     } else {
-      listFilesIfExists(localPaths, new Path(localPaths.head).getFileSystem(hadoopConf))
+      listFilesIfExists(localPaths, hadoopConf)
     }
 
     val finishList = System.currentTimeMillis()