Use cache recursively and fix some compile errors

Author: sryza

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -171,7 +171,7 @@ private[spark] class Executor(
       startGCTime = gcTime
 
       try {
-        val (taskFiles, taskJars, taskDirs, taskBytes) =
+        val (taskFiles, taskJars, taskBytes) =
           Task.deserializeWithDependencies(serializedTask)
         updateDependencies(taskFiles, taskJars)
         task = ser.deserialize[Task[Any]](taskBytes, Thread.currentThread.getContextClassLoader)

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

@@ -481,7 +481,7 @@ private[spark] object Utils extends Logging {
       removeSourceFile: Boolean = false): Unit = {
 
     if (destFile.exists) {
-      if (!Files.equal(sourceFile, destFile)) {
+      if (!filesEqualRecursive(sourceFile, destFile)) {
         if (fileOverwrite) {
           logInfo(
             s"File $destFile exists and does not match contents of $url, replacing it with $url"
@@ -516,7 +516,34 @@ private[spark] object Utils extends Logging {
       Files.move(sourceFile, destFile)
     } else {
       logInfo(s"Copying ${sourceFile.getAbsolutePath} to ${destFile.getAbsolutePath}")
-      Files.copy(sourceFile, destFile)
+      copyRecursive(sourceFile, destFile)
+    }
+  }
+
+  private def filesEqualRecursive(file1: File, file2: File): Boolean = {
+    if (file1.isDirectory && file2.isDirectory) {
+      val subfiles1 = file1.listFiles()
+      val subfiles2 = file2.listFiles()
+      if (subfiles1.size != subfiles2.size) {
+        return false
+      }
+      subfiles1.sortBy(_.getName).zip(subfiles2.sortBy(_.getName)).dropWhile {
+        case (f1, f2) => filesEqualRecursive(f1, f2)
+      }.isEmpty
+    } else if (file1.isFile && file2.isFile) {
+      Files.equal(file1, file2)
+    } else {
+      false
+    }
+  }
+
+  private def copyRecursive(source: File, dest: File): Unit = {
+    if (source.isDirectory) {
+      dest.mkdir()
+      val subfiles = source.listFiles()
+      subfiles.foreach(f => copyRecursive(f, new File(dest, f.getName)))
+    } else {
+      Files.copy(source, dest)
     }
   }
 

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

@@ -17,10 +17,11 @@
 
 package org.apache.spark
 
+import java.io.File
+
 import org.scalatest.FunSuite
 
 import org.apache.hadoop.io.BytesWritable
-import java.io.File
 
 class SparkContextSuite extends FunSuite with LocalSparkContext {