[SPARK-24195][Core] Ignore the files with "local" scheme in SparkContext.addFile

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

@@ -1517,9 +1517,19 @@ class SparkContext(config: SparkConf) extends Logging {
    * only supported for Hadoop-supported filesystems.
    */
   def addFile(path: String, recursive: Boolean): Unit = {
-    val uri = new Path(path).toUri
+    var uri = new Path(path).toUri
+    // mark the original path's scheme is local or not, there is no need to add the local file
+    // in file server.
+    var localFile = false
     val schemeCorrectedPath = uri.getScheme match {
-      case null | "local" => new File(path).getCanonicalFile.toURI.toString
+      case null =>
+        new File(path).getCanonicalFile.toURI.toString
+      case "local" =>
+        localFile = true
+        val tmpPath = new File(uri.getPath).getCanonicalFile.toURI.toString
+        // SPARK-24195: Local is not a valid scheme for FileSystem, we should only keep path here.
+        uri = new Path(uri.getPath).toUri
+        tmpPath
       case _ => path
     }
 
@@ -1541,7 +1551,7 @@ class SparkContext(config: SparkConf) extends Logging {
       Utils.validateURL(uri)
     }
 
-    val key = if (!isLocal && scheme == "file") {
+    val key = if (!isLocal && scheme == "file" && !localFile) {
       env.rpcEnv.fileServer.addFile(new File(uri.getPath))
     } else {
       schemeCorrectedPath

core/src/test/scala/org/apache/spark/SparkContextSuite.scala

@@ -116,51 +116,54 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
   test("basic case for addFile and listFiles") {
     val dir = Utils.createTempDir()
 
+    // file and absolute path for normal path
     val file1 = File.createTempFile("someprefix1", "somesuffix1", dir)
     val absolutePath1 = file1.getAbsolutePath
 
+    // file and absolute path for relative path
     val file2 = File.createTempFile("someprefix2", "somesuffix2", dir)
     val relativePath = file2.getParent + "/../" + file2.getParentFile.getName + "/" + file2.getName
     val absolutePath2 = file2.getAbsolutePath
 
+    // file and absolute path for path with local scheme
+    val file3 = File.createTempFile("someprefix3", "somesuffix3", dir)
+    val localPath = s"local://${file3.getParent}/../${file3.getParentFile.getName}" +
+      s"/${file3.getName}"
+    val absolutePath3 = file3.getAbsolutePath
+
     try {
       Files.write("somewords1", file1, StandardCharsets.UTF_8)
       Files.write("somewords2", file2, StandardCharsets.UTF_8)
-      val length1 = file1.length()
-      val length2 = file2.length()
+      Files.write("somewords3", file3, StandardCharsets.UTF_8)
 
-      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
-      sc.addFile(file1.getAbsolutePath)
-      sc.addFile(relativePath)
-      sc.parallelize(Array(1), 1).map(x => {
-        val gotten1 = new File(SparkFiles.get(file1.getName))
-        val gotten2 = new File(SparkFiles.get(file2.getName))
-        if (!gotten1.exists()) {
+      def checkGottenFile(file: File, absolutePath: String): Unit = {
+        val length = file.length()
+        val gotten = new File(SparkFiles.get(file.getName))
+        if (!gotten.exists()) {
           throw new SparkException("file doesn't exist : " + absolutePath1)
         }
-        if (!gotten2.exists()) {
-          throw new SparkException("file doesn't exist : " + absolutePath2)
-        }
 
-        if (length1 != gotten1.length()) {
+        if (file.length() != gotten.length()) {
           throw new SparkException(
-            s"file has different length $length1 than added file ${gotten1.length()} : " +
+            s"file has different length $length than added file ${gotten.length()} : " +
               absolutePath1)
         }
-        if (length2 != gotten2.length()) {
-          throw new SparkException(
-            s"file has different length $length2 than added file ${gotten2.length()} : " +
-              absolutePath2)
-        }
 
-        if (absolutePath1 == gotten1.getAbsolutePath) {
+        if (absolutePath == gotten.getAbsolutePath) {
           throw new SparkException("file should have been copied :" + absolutePath1)
         }
-        if (absolutePath2 == gotten2.getAbsolutePath) {
-          throw new SparkException("file should have been copied : " + absolutePath2)
-        }
+      }
+
+      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
+      sc.addFile(file1.getAbsolutePath)
+      sc.addFile(relativePath)
+      sc.addFile(localPath)
+      sc.parallelize(Array(1), 1).map { x =>
+        checkGottenFile(file1, absolutePath1)
+        checkGottenFile(file2, absolutePath2)
+        checkGottenFile(file3, absolutePath3)
         x
-      }).count()
+      }.count()
       assert(sc.listFiles().filter(_.contains("somesuffix1")).size == 1)
     } finally {
       sc.stop()