Skip to content

Commit

Permalink
removed testWindows from Utils.resolveURI and Utils.resolveURIs.
Browse files Browse the repository at this point in the history 
replaced SystemUtils.IS_OS_WINDOWS to Utils.isWindows.
removed Utils.formatPath from PythonRunner.scala.
  • Loading branch information
@tsudukim
tsudukim committed May 8, 2015
1 parent 84c33d0 commit e03f289
Show file tree
Hide file tree
Showing 4 changed files with 51 additions and 48 deletions.
23 changes: 12 additions & 11 deletions core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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._
Expand Down Expand Up @@ -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) {
Expand All @@ -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
}

Expand Down
8 changes: 4 additions & 4 deletions core/src/main/scala/org/apache/spark/util/Utils.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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(",")
}
}

Expand All @@ -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
Expand Down
24 changes: 14 additions & 10 deletions core/src/test/scala/org/apache/spark/deploy/PythonRunnerSuite.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.deploy

import org.scalatest.FunSuite
import org.apache.spark.util.Utils

class PythonRunnerSuite extends FunSuite {

Expand All @@ -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") }
Expand All @@ -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") }
Expand Down
44 changes: 21 additions & 23 deletions core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
}
}

Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down

0 comments on commit e03f289

Please sign in to comment.