Fixed several bugs that caused weird behavior with files in spark-shell:

- SizeEstimator was following through a ClassLoader field of Hadoop
  JobConfs, which referenced the whole interpreter, Scala compiler, etc.
  Chaos ensued, giving an estimated size in the tens of gigabytes.
- Broadcast variables in local mode were only stored as MEMORY_ONLY and
  never made accessible over a server, so they fell out of the cache when
  they were deemed too large and couldn't be reloaded.

1 files changed, 22 insertions, 1 deletions

diff --git a/repl/src/test/scala/spark/repl/ReplSuite.scala b/repl/src/test/scala/spark/repl/ReplSuite.scala
index 15ebf0c9b8..0b5d439ca4 100644
--- a/repl/src/test/scala/spark/repl/ReplSuite.scala
+++ b/repl/src/test/scala/spark/repl/ReplSuite.scala
@@ -7,6 +7,7 @@ import scala.collection.mutable.ArrayBuffer
 import scala.collection.JavaConversions._
 
 import org.scalatest.FunSuite
+import com.google.common.io.Files
 
 class ReplSuite extends FunSuite {
   def runInterpreter(master: String, input: String): String = {
@@ -118,7 +119,27 @@ class ReplSuite extends FunSuite {
     assertContains("res0: Array[Int] = Array(0, 0, 0, 0, 0)", output)
     assertContains("res2: Array[Int] = Array(5, 0, 0, 0, 0)", output)
   }
-  
+
+  test ("interacting with files") {
+    val tempDir = Files.createTempDir()
+    val out = new FileWriter(tempDir + "/input")
+    out.write("Hello world!\n")
+    out.write("What's up?\n")
+    out.write("Goodbye\n")
+    out.close()
+    val output = runInterpreter("local", """
+      var file = sc.textFile("%s/input").cache()
+      file.count()
+      file.count()
+      file.count()
+      """.format(tempDir.getAbsolutePath))
+    assertDoesNotContain("error:", output)
+    assertDoesNotContain("Exception", output)
+    assertContains("res0: Long = 3", output)
+    assertContains("res1: Long = 3", output)
+    assertContains("res2: Long = 3", output)
+  }
+
   if (System.getenv("MESOS_NATIVE_LIBRARY") != null) {
     test ("running on Mesos") {
       val output = runInterpreter("localquiet", """