removed testWindows from Utils.resolveURI and Utils.resolveURIs.

replaced SystemUtils.IS_OS_WINDOWS to Utils.isWindows.
removed Utils.formatPath from PythonRunner.scala.

Author: tsudukim

core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala

@@ -18,6 +18,8 @@
 package org.apache.spark.deploy
 
 import java.net.URI
+import java.io.File
+import scala.util.Try
 
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.JavaConversions._
@@ -81,16 +83,13 @@ object PythonRunner {
       throw new IllegalArgumentException("Launching Python applications through " +
         s"spark-submit is currently only supported for local files: $path")
     }
-    val windows = Utils.isWindows || testWindows
-    var formattedPath = Utils.formatPath(path, windows)
-
-    // Strip the URI scheme from the path
-    formattedPath =
-      new URI(formattedPath).getScheme match {
-        case null => formattedPath
-        case Utils.windowsDrive(d) if windows => formattedPath
-        case _ => new URI(formattedPath).getPath
-      }
+    // get path when scheme is file.
+    val uri = Try(new URI(path)).getOrElse(new File(path).toURI)
+    var formattedPath = uri.getScheme match {
+      case null => path
+      case "file" | "local" => uri.getPath
+      case _ => null
+    }
 
     // Guard against malformed paths potentially throwing NPE
     if (formattedPath == null) {
@@ -99,7 +98,9 @@ object PythonRunner {
 
     // In Windows, the drive should not be prefixed with "/"
     // For instance, python does not understand "/C:/path/to/sheep.py"
-    formattedPath = if (windows) formattedPath.stripPrefix("/") else formattedPath
+    if (formattedPath.matches("/[a-zA-Z]:/.*")) {
+      formattedPath = formattedPath.stripPrefix("/")
+    }
     formattedPath
   }
 

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -1765,7 +1765,7 @@ private[spark] object Utils extends Logging {
    * If the supplied path does not contain a scheme, or is a relative path, it will be
    * converted into an absolute path with a file:// scheme.
    */
-  def resolveURI(path: String, testWindows: Boolean = false): URI = {
+  def resolveURI(path: String): URI = {
     try {
       val uri = new URI(path)
       if (uri.getScheme() != null) {
@@ -1778,11 +1778,11 @@ private[spark] object Utils extends Logging {
   }
 
   /** Resolve a comma-separated list of paths. */
-  def resolveURIs(paths: String, testWindows: Boolean = false): String = {
+  def resolveURIs(paths: String): String = {
     if (paths == null || paths.trim.isEmpty) {
       ""
     } else {
-      paths.split(",").map { p => Utils.resolveURI(p, testWindows) }.mkString(",")
+      paths.split(",").map { p => Utils.resolveURI(p) }.mkString(",")
     }
   }
 
@@ -1793,7 +1793,7 @@ private[spark] object Utils extends Logging {
       Array.empty
     } else {
       paths.split(",").filter { p =>
-        val uri = resolveURI(p, windows)
+        val uri = resolveURI(p)
         Option(uri.getScheme).getOrElse("file") match {
           case windowsDrive(d) if windows => false
           case "local" | "file" => false

core/src/test/scala/org/apache/spark/deploy/PythonRunnerSuite.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.deploy
 
 import org.scalatest.FunSuite
+import org.apache.spark.util.Utils
 
 class PythonRunnerSuite extends FunSuite {
 
@@ -28,10 +29,13 @@ class PythonRunnerSuite extends FunSuite {
     assert(PythonRunner.formatPath("file:///spark.py") === "/spark.py")
     assert(PythonRunner.formatPath("local:/spark.py") === "/spark.py")
     assert(PythonRunner.formatPath("local:///spark.py") === "/spark.py")
-    assert(PythonRunner.formatPath("C:/a/b/spark.py", testWindows = true) === "C:/a/b/spark.py")
-    assert(PythonRunner.formatPath("/C:/a/b/spark.py", testWindows = true) === "C:/a/b/spark.py")
     assert(PythonRunner.formatPath("file:/C:/a/b/spark.py", testWindows = true) ===
       "C:/a/b/spark.py")
+    if (Utils.isWindows) {
+      assert(PythonRunner.formatPath("C:\\a\\b\\spark.py", testWindows = true) === "C:/a/b/spark.py")
+      assert(PythonRunner.formatPath("C:\\a b\\spark.py", testWindows = true) ===
+        "C:/a b/spark.py")
+    }
     intercept[IllegalArgumentException] { PythonRunner.formatPath("one:two") }
     intercept[IllegalArgumentException] { PythonRunner.formatPath("hdfs:s3:xtremeFS") }
     intercept[IllegalArgumentException] { PythonRunner.formatPath("hdfs:/path/to/some.py") }
@@ -45,14 +49,14 @@ class PythonRunnerSuite extends FunSuite {
       Array("/app.py", "/spark.py"))
     assert(PythonRunner.formatPaths("me.py,file:/you.py,local:/we.py") ===
       Array("me.py", "/you.py", "/we.py"))
-    assert(PythonRunner.formatPaths("C:/a/b/spark.py", testWindows = true) ===
-      Array("C:/a/b/spark.py"))
-    assert(PythonRunner.formatPaths("/C:/a/b/spark.py", testWindows = true) ===
-      Array("C:/a/b/spark.py"))
-    assert(PythonRunner.formatPaths("C:/free.py,pie.py", testWindows = true) ===
-      Array("C:/free.py", "pie.py"))
-    assert(PythonRunner.formatPaths("lovely.py,C:/free.py,file:/d:/fry.py", testWindows = true) ===
-      Array("lovely.py", "C:/free.py", "d:/fry.py"))
+    if (Utils.isWindows) {
+      assert(PythonRunner.formatPaths("C:\\a\\b\\spark.py", testWindows = true) ===
+        Array("C:/a/b/spark.py"))
+      assert(PythonRunner.formatPaths("C:\\free.py,pie.py", testWindows = true) ===
+        Array("C:/free.py", "pie.py"))
+      assert(PythonRunner.formatPaths("lovely.py,C:\\free.py,file:/d:/fry.py", testWindows = true) ===
+        Array("lovely.py", "C:/free.py", "d:/fry.py"))
+    }
     intercept[IllegalArgumentException] { PythonRunner.formatPaths("one:two,three") }
     intercept[IllegalArgumentException] { PythonRunner.formatPaths("two,three,four:five:six") }
     intercept[IllegalArgumentException] { PythonRunner.formatPaths("hdfs:/some.py,foo.py") }

core/src/test/scala/org/apache/spark/util/UtilsSuite.scala

@@ -28,7 +28,6 @@ import java.util.Locale
 import com.google.common.base.Charsets.UTF_8
 import com.google.common.io.Files
 import org.scalatest.FunSuite
-import org.apache.commons.lang3.SystemUtils
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
@@ -222,58 +221,57 @@ class UtilsSuite extends FunSuite with ResetSystemProperties {
   }
 
   test("resolveURI") {
-    def assertResolves(before: String, after: String, testWindows: Boolean = false): Unit = {
+    def assertResolves(before: String, after: String): Unit = {
       // This should test only single paths
       assume(before.split(",").length === 1)
       // Repeated invocations of resolveURI should yield the same result
-      def resolve(uri: String): String = Utils.resolveURI(uri, testWindows).toString
+      def resolve(uri: String): String = Utils.resolveURI(uri).toString
       assert(resolve(after) === after)
       assert(resolve(resolve(after)) === after)
       assert(resolve(resolve(resolve(after))) === after)
       // Also test resolveURIs with single paths
-      assert(new URI(Utils.resolveURIs(before, testWindows)) === new URI(after))
-      assert(new URI(Utils.resolveURIs(after, testWindows)) === new URI(after))
+      assert(new URI(Utils.resolveURIs(before)) === new URI(after))
+      assert(new URI(Utils.resolveURIs(after)) === new URI(after))
     }
     val rawCwd = System.getProperty("user.dir")
-    val cwd = if (SystemUtils.IS_OS_WINDOWS) s"/$rawCwd".replace("\\", "/") else rawCwd
+    val cwd = if (Utils.isWindows) s"/$rawCwd".replace("\\", "/") else rawCwd
     assertResolves("hdfs:/root/spark.jar", "hdfs:/root/spark.jar")
     assertResolves("hdfs:///root/spark.jar#app.jar", "hdfs:/root/spark.jar#app.jar")
     assertResolves("spark.jar", s"file:$cwd/spark.jar")
     assertResolves("spark.jar#app.jar", s"file:$cwd/spark.jar%23app.jar")
     assertResolves("path to/file.txt", s"file:$cwd/path%20to/file.txt")
-    if (SystemUtils.IS_OS_WINDOWS) {
-      assertResolves("C:\\path\\to\\file.txt", "file:/C:/path/to/file.txt", testWindows = true)
-      assertResolves("C:\\path to\\file.txt", "file:/C:/path%20to/file.txt", testWindows = true)
+    if (Utils.isWindows) {
+      assertResolves("C:\\path\\to\\file.txt", "file:/C:/path/to/file.txt")
+      assertResolves("C:\\path to\\file.txt", "file:/C:/path%20to/file.txt")
     }
-    assertResolves("file:/C:/path/to/file.txt", "file:/C:/path/to/file.txt", testWindows = true)
-    assertResolves("file:///C:/path/to/file.txt", "file:/C:/path/to/file.txt", testWindows = true)
-    assertResolves("file:/C:/file.txt#alias.txt", "file:/C:/file.txt#alias.txt", testWindows = true)
+    assertResolves("file:/C:/path/to/file.txt", "file:/C:/path/to/file.txt")
+    assertResolves("file:///C:/path/to/file.txt", "file:/C:/path/to/file.txt")
+    assertResolves("file:/C:/file.txt#alias.txt", "file:/C:/file.txt#alias.txt")
     assertResolves("file:foo", s"file:foo")
     assertResolves("file:foo:baby", s"file:foo:baby")
   }
 
   test("resolveURIs with multiple paths") {
-    def assertResolves(before: String, after: String, testWindows: Boolean = false): Unit = {
+    def assertResolves(before: String, after: String): Unit = {
       assume(before.split(",").length > 1)
-      assert(Utils.resolveURIs(before, testWindows) === after)
-      assert(Utils.resolveURIs(after, testWindows) === after)
+      assert(Utils.resolveURIs(before) === after)
+      assert(Utils.resolveURIs(after) === after)
       // Repeated invocations of resolveURIs should yield the same result
-      def resolve(uri: String): String = Utils.resolveURIs(uri, testWindows)
+      def resolve(uri: String): String = Utils.resolveURIs(uri)
       assert(resolve(after) === after)
       assert(resolve(resolve(after)) === after)
       assert(resolve(resolve(resolve(after))) === after)
     }
     val rawCwd = System.getProperty("user.dir")
-    val cwd = if (SystemUtils.IS_OS_WINDOWS) s"/$rawCwd".replace("\\", "/") else rawCwd
+    val cwd = if (Utils.isWindows) s"/$rawCwd".replace("\\", "/") else rawCwd
     assertResolves("jar1,jar2", s"file:$cwd/jar1,file:$cwd/jar2")
     assertResolves("file:/jar1,file:/jar2", "file:/jar1,file:/jar2")
     assertResolves("hdfs:/jar1,file:/jar2,jar3", s"hdfs:/jar1,file:/jar2,file:$cwd/jar3")
     assertResolves("hdfs:/jar1,file:/jar2,jar3,jar4#jar5,path to/jar6",
       s"hdfs:/jar1,file:/jar2,file:$cwd/jar3,file:$cwd/jar4%23jar5,file:$cwd/path%20to/jar6")
-    if (SystemUtils.IS_OS_WINDOWS) {
-      assertResolves( """hdfs:/jar1,file:/jar2,jar3,C:\pi.py#py.pi,C:\path to\jar4""",
-        s"hdfs:/jar1,file:/jar2,file:$cwd/jar3,file:/C:/pi.py%23py.pi,file:/C:/path%20to/jar4",
-        testWindows = true)
+    if (Utils.isWindows) {
+      assertResolves("""hdfs:/jar1,file:/jar2,jar3,C:\pi.py#py.pi,C:\path to\jar4""",
+        s"hdfs:/jar1,file:/jar2,file:$cwd/jar3,file:/C:/pi.py%23py.pi,file:/C:/path%20to/jar4")
     }
   }
 
@@ -404,7 +402,7 @@ class UtilsSuite extends FunSuite with ResetSystemProperties {
     Files.write("some text", sourceFile, UTF_8)
 
     val path =
-      if (SystemUtils.IS_OS_WINDOWS) {
+      if (Utils.isWindows) {
         new Path("file:/" + sourceDir.getAbsolutePath.replace("\\", "/"))
       } else {
         new Path("file://" + sourceDir.getAbsolutePath)
@@ -429,7 +427,7 @@ class UtilsSuite extends FunSuite with ResetSystemProperties {
     assert(destInnerFile.isFile())
 
     val filePath =
-      if (SystemUtils.IS_OS_WINDOWS) {
+      if (Utils.isWindows) {
         new Path("file:/" + sourceFile.getAbsolutePath.replace("\\", "/"))
       } else {
         new Path("file://" + sourceFile.getAbsolutePath)