Merge pull request #143 from ibm-et/FixDuplicateOutput

Fixed duplicate output caused by Scala interpreter not setting a new …

kernel-api/src/main/scala/com/ibm/spark/interpreter/Interpreter.scala

@@ -105,10 +105,10 @@ trait Interpreter {
   def completion(code: String, pos: Int): (Int, List[String] )
 
   /**
-   * Returns the most recent variable name.
-   * @return the String name
+   * Returns the name of the variable created from the last execution.
+   * @return Some String name if a variable was created, otherwise None
    */
-  def mostRecentVar: String
+  def lastExecutionVariableName: Option[String]
 
   /**
    * Returns the class loader used by this interpreter.

kernel-api/src/main/scala/com/ibm/spark/interpreter/ScalaInterpreter.scala

@@ -470,9 +470,19 @@ class ScalaInterpreter(
     else variable
   }
 
-  override def mostRecentVar: String = {
+  override def lastExecutionVariableName: Option[String] = {
     require(sparkIMain != null)
-    sparkIMain.mostRecentVar
+
+    // TODO: Get this API method changed back to public in Apache Spark
+    val lastRequestMethod = classOf[SparkIMain].getDeclaredMethod("lastRequest")
+    lastRequestMethod.setAccessible(true)
+
+    val request =
+      lastRequestMethod.invoke(sparkIMain).asInstanceOf[SparkIMain#Request]
+
+    val mostRecentVariableName = sparkIMain.mostRecentVar
+
+    request.definedNames.map(_.toString).find(_ == mostRecentVariableName)
   }
 
   override def completion(code: String, pos: Int): (Int, List[String]) = {

kernel/src/main/scala/com/ibm/spark/kernel/protocol/v5/magic/PostProcessor.scala

@@ -9,10 +9,7 @@ class PostProcessor(interpreter: Interpreter) extends LogLike {
   val defaultErr = "Something went wrong in postprocessor!"
 
   def process(codeOutput: ExecuteOutput): Data = {
-    val varName = interpreter.mostRecentVar
-    val v = interpreter.read(varName)
-
-    v match {
+    interpreter.lastExecutionVariableName.flatMap(interpreter.read) match {
       case Some(l: Left[_, _]) => matchCellMagic(codeOutput, l)
       case Some(r: Right[_, _]) => matchLineMagic(codeOutput, r)
       case _ => Data(MIMEType.PlainText -> codeOutput)

kernel/src/main/scala/com/ibm/spark/magic/builtin/RDD.scala

@@ -33,7 +33,7 @@ class RDD extends CellMagic with IncludeKernelInterpreter with LogLike {
     val (result, message) = kernelInterpreter.interpret(code)
     result match {
       case Results.Success =>
-        val rddVarName = kernelInterpreter.mostRecentVar
+        val rddVarName = kernelInterpreter.lastExecutionVariableName.getOrElse("")
         kernelInterpreter.read(rddVarName).map(rddVal => {
           try{
             CellMagicOutput(MIMEType.ApplicationJson -> RddToJson.convert(rddVal.asInstanceOf[SchemaRDD]))

kernel/src/test/scala/com/ibm/spark/kernel/protocol/v5/magic/PostProcessorSpec.scala

@@ -65,6 +65,9 @@ class PostProcessorSpec extends FunSpec with Matchers with MockitoSugar{
     it("should call matchCellMagic when the last variable is a Left") {
       val intp = mock[Interpreter]
       val left = Left("")
+      // Need to mock lastExecutionVariableName as it is being chained with
+      // the read method
+      doReturn(Some("")).when(intp).lastExecutionVariableName
       doReturn(Some(left)).when(intp).read(anyString())
 
       val processor = spy(new PostProcessor(intp))
@@ -76,6 +79,9 @@ class PostProcessorSpec extends FunSpec with Matchers with MockitoSugar{
     it("should call matchLineMagic when the last variable is a Right") {
       val intp = mock[Interpreter]
       val right = Right("")
+      // Need to mock lastExecutionVariableName as it is being chained with
+      // the read method
+      doReturn(Some("")).when(intp).lastExecutionVariableName
       doReturn(Some(right)).when(intp).read(anyString())
 
       val processor = spy(new PostProcessor(intp))
@@ -88,6 +94,9 @@ class PostProcessorSpec extends FunSpec with Matchers with MockitoSugar{
       "Left[CellMagicOutput, Nothing]") {
       val intp = mock[Interpreter]
       val left = Left("")
+      // Need to mock lastExecutionVariableName as it is being chained with
+      // the read method
+      doReturn(Some("")).when(intp).lastExecutionVariableName
       doReturn(Some(left)).when(intp).read(anyString())
 
       val processor = spy(new PostProcessor(intp))
@@ -100,6 +109,9 @@ class PostProcessorSpec extends FunSpec with Matchers with MockitoSugar{
        "Right[LineMagicOutput, Nothing]") {
       val intp = mock[Interpreter]
       val right = Right("")
+      // Need to mock lastExecutionVariableName as it is being chained with
+      // the read method
+      doReturn(Some("")).when(intp).lastExecutionVariableName
       doReturn(Some(right)).when(intp).read(anyString())
 
       val processor = spy(new PostProcessor(intp))

kernel/src/test/scala/com/ibm/spark/magic/builtin/RDDSpec.scala

@@ -49,7 +49,7 @@ class RDDSpec extends FunSpec with Matchers with MockitoSugar with BeforeAndAfte
   }
 
   before {
-    doReturn("someRDD").when(mockInterpreter).mostRecentVar
+    doReturn(Some("someRDD")).when(mockInterpreter).lastExecutionVariableName
     doReturn(Some(mockDataFrame)).when(mockInterpreter).read(anyString())
     doReturn((Results.Success, Left(resOutput)))
       .when(mockInterpreter).interpret(anyString(), anyBoolean())

kernel/src/test/scala/integration/PostProcessorSpecForIntegration.scala

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2015 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package integration
+
+import java.io.OutputStream
+
+import com.ibm.spark.interpreter._
+import com.ibm.spark.kernel.protocol.v5.magic.PostProcessor
+import org.scalatest.mock.MockitoSugar
+import org.scalatest.{BeforeAndAfter, Matchers, FunSpec}
+
+class PostProcessorSpecForIntegration extends FunSpec with Matchers
+  with BeforeAndAfter with MockitoSugar
+{
+  private var scalaInterpreter: ScalaInterpreter = _
+  private var postProcessor: PostProcessor = _
+
+  before {
+    // TODO: Move instantiation and start of interpreter to a beforeAll
+    //       for performance improvements
+    scalaInterpreter = new ScalaInterpreter(Nil, mock[OutputStream])
+      with StandardSparkIMainProducer
+      with StandardTaskManagerProducer
+      with StandardSettingsProducer
+
+    scalaInterpreter.start()
+
+    postProcessor = new PostProcessor(scalaInterpreter)
+  }
+
+  describe("PostProcessor") {
+    describe("#process") {
+      describe("https://github.com/ibm-et/spark-kernel/issues/137") {
+        it(Seq(
+          "should not return a previous execution's result for a",
+          "new execution with no result").mkString(" ")) {
+          val result = scalaInterpreter.interpret("1+1")
+          val postResult = postProcessor.process(result._2.left.get)
+
+          // Imports in Scala do not create a new variable based on execution
+          val result2 = scalaInterpreter.interpret("import java.lang._")
+          val postResult2 = postProcessor.process(result2._2.left.get)
+
+          postResult should not be (postResult2)
+        }
+      }
+    }
+  }
+}