Change from hardcoded path separator

Author: Mridul Muralidharan

core/src/main/scala/org/apache/spark/rdd/PipedRDD.scala

@@ -87,10 +87,10 @@ private[spark] class PipedRDD[T: ClassTag](
     // When spark.worker.separated.working.directory option is turned on, each
     // task will be run in separate directory. This should be resolve file
     // access conflict issue
-    val taskDirectory = "./tasks/" + java.util.UUID.randomUUID.toString
+    val taskDirectory = "tasks" + File.separator + java.util.UUID.randomUUID.toString
     var workInTaskDirectory = false
     logDebug("taskDirectory = " + taskDirectory)
-    if (separateWorkingDir == true) {
+    if (separateWorkingDir) {
       val currentDir = new File(".")
       logDebug("currentDir = " + currentDir.getAbsolutePath())
       val taskDirFile = new File(taskDirectory)
@@ -106,13 +106,13 @@ private[spark] class PipedRDD[T: ClassTag](
         for (file <- currentDir.list(tasksDirFilter)) {
           val fileWithDir = new File(currentDir, file)
           Utils.symlink(new File(fileWithDir.getAbsolutePath()),
-            new File(taskDirectory + "/" + fileWithDir.getName()))
+            new File(taskDirectory + File.separator + fileWithDir.getName()))
         }
         pb.directory(taskDirFile)
         workInTaskDirectory = true
       } catch {
         case e: Exception => logError("Unable to setup task working directory: " + e.getMessage +
-          " (" + taskDirectory + ")")
+          " (" + taskDirectory + ")", e)
       }
     }
 

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

@@ -30,6 +30,7 @@ import scala.io.Source
 import scala.reflect.ClassTag
 
 import com.google.common.io.Files
+import org.apache.commons.lang.SystemUtils
 import com.google.common.util.concurrent.ThreadFactoryBuilder
 import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
 import org.json4s._
@@ -45,7 +46,10 @@ import org.apache.spark.serializer.{DeserializationStream, SerializationStream,
  */
 private[spark] object Utils extends Logging {
 
-  val osName = System.getProperty("os.name")
+  def sparkBin(sparkHome: String, which: String): File = {
+    val suffix = if (SystemUtils.IS_OS_WINDOWS) ".cmd" else ""
+    new File(sparkHome + File.separator + "bin", which + suffix)
+  }
 
   /** Serialize an object using Java serialization */
   def serialize[T](o: T): Array[Byte] = {
@@ -603,7 +607,7 @@ private[spark] object Utils extends Logging {
    */
   def isSymlink(file: File): Boolean = {
     if (file == null) throw new NullPointerException("File must not be null")
-    if (osName.startsWith("Windows")) return false
+    if (SystemUtils.IS_OS_WINDOWS) return false
     val fileInCanonicalDir = if (file.getParent() == null) {
       file
     } else {
@@ -1007,9 +1011,17 @@ private[spark] object Utils extends Logging {
     if (dst.isAbsolute()) {
       throw new IOException("Destination must be relative")
     }
-    val linkCmd = if (osName.startsWith("Windows")) "copy" else "ln -sf"
+    var cmdSuffix = ""
+    val linkCmd = if (SystemUtils.IS_OS_WINDOWS) {
+      // refer to http://technet.microsoft.com/en-us/library/cc771254.aspx
+      cmdSuffix = " /s /e /k /h /y /i"
+      "cmd /c xcopy "
+    } else {
+      cmdSuffix = ""
+      "ln -sf "
+    }
     import scala.sys.process._
-    (linkCmd + " " + src.getAbsolutePath() + " " + dst.getPath()) lines_! ProcessLogger(line =>
+    (linkCmd + src.getAbsolutePath() + " " + dst.getPath() + cmdSuffix) lines_! ProcessLogger(line =>
        (logInfo(line)))
   }
 

repl/src/test/scala/org/apache/spark/repl/ReplSuite.scala

@@ -25,6 +25,7 @@ import scala.collection.mutable.ArrayBuffer
 import com.google.common.io.Files
 import org.scalatest.FunSuite
 import org.apache.spark.SparkContext
+import org.apache.commons.lang.StringEscapeUtils
 
 
 class ReplSuite extends FunSuite {
@@ -185,11 +186,12 @@ class ReplSuite extends FunSuite {
     out.close()
     val output = runInterpreter("local",
       """
-        |var file = sc.textFile("%s/input").cache()
+        |var file = sc.textFile("%s").cache()
         |file.count()
         |file.count()
         |file.count()
-      """.stripMargin.format(tempDir.getAbsolutePath))
+      """.stripMargin.format(StringEscapeUtils.escapeJava(
+        tempDir.getAbsolutePath + File.separator + "input")))
     assertDoesNotContain("error:", output)
     assertDoesNotContain("Exception", output)
     assertContains("res0: Long = 3", output)

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

@@ -100,14 +100,15 @@ class TestHiveContext(sc: SparkContext) extends LocalHiveContext(sc) {
   hiveFilesTemp.delete()
   hiveFilesTemp.mkdir()
 
-  val inRepoTests = if (System.getProperty("user.dir").endsWith("sql/hive")) {
-    new File("src/test/resources/")
+  val inRepoTests = if (System.getProperty("user.dir").endsWith("sql" + File.separator + "hive")) {
+    new File("src" + File.separator + "test" + File.separator + "resources" + File.separator)
   } else {
-    new File("sql/hive/src/test/resources")
+    new File("sql" + File.separator + "hive" + File.separator + "src" + File.separator + "test" + 
+      File.separator + "resources")
   }
 
   def getHiveFile(path: String): File = {
-    val stripped = path.replaceAll("""\.\.\/""", "")
+    val stripped = path.replaceAll("""\.\.\/""", "").replace('/', File.separatorChar)
     hiveDevHome
       .map(new File(_, stripped))
       .filter(_.exists)

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/BigDataBenchmarkSuite.scala

@@ -26,7 +26,7 @@ import org.apache.spark.sql.hive.test.TestHive._
  * https://amplab.cs.berkeley.edu/benchmark/
  */
 class BigDataBenchmarkSuite extends HiveComparisonTest {
-  val testDataDirectory = new File("target/big-data-benchmark-testdata")
+  val testDataDirectory = new File("target" + File.separator + "big-data-benchmark-testdata")
 
   val testTables = Seq(
     TestTable(

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala

@@ -19,14 +19,16 @@ package org.apache.spark.sql.hive.execution
 
 import org.scalatest.BeforeAndAfter
 
-import org.apache.spark.sql.hive.test.TestHive
+import org.apache.spark.sql.hive.TestHive
+import java.io.File
 
 /**
  * Runs the test cases that are included in the hive distribution.
  */
 class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
   // TODO: bundle in jar files... get from classpath
-  lazy val hiveQueryDir = TestHive.getHiveFile("ql/src/test/queries/clientpositive")
+  lazy val hiveQueryDir = TestHive.getHiveFile("ql" + File.separator + "src" +
+    File.separator + "test" + File.separator + "queries" + File.separator + "clientpositive")
   def testCases = hiveQueryDir.listFiles.map(f => f.getName.stripSuffix(".q") -> f)
 
   override def beforeAll() {