[SPARK-25767][SQL] Fix lazily evaluated stream of expressions in code generation

## What changes were proposed in this pull request?

Code generation is incorrect if `outputVars` parameter of `consume` method in `CodegenSupport` contains a lazily evaluated stream of expressions.
This PR fixes the issue by forcing the evaluation of `inputVars` before generating the code for UnsafeRow.

## How was this patch tested?

Tested with the sample program provided in https://issues.apache.org/jira/browse/SPARK-25767

Closes apache#22789 from peter-toth/SPARK-25767.

Authored-by: Peter Toth <[email protected]>
Signed-off-by: Herman van Hovell <[email protected]>
(cherry picked from commit 7fe5cff)
Signed-off-by: Herman van Hovell <[email protected]>

Author: peter-toth

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

@@ -146,7 +146,10 @@ trait CodegenSupport extends SparkPlan {
       if (outputVars != null) {
         assert(outputVars.length == output.length)
         // outputVars will be used to generate the code for UnsafeRow, so we should copy them
-        outputVars.map(_.copy())
+        outputVars.map(_.copy()) match {
+          case stream: Stream[ExprCode] => stream.force
+          case other => other
+        }
       } else {
         assert(row != null, "outputVars and row cannot both be null.")
         ctx.currentVars = null

sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala

@@ -319,4 +319,15 @@ class WholeStageCodegenSuite extends QueryTest with SharedSQLContext {
       assert(df.limit(1).collect() === Array(Row("bat", 8.0)))
     }
   }
+
+  test("SPARK-25767: Lazy evaluated stream of expressions handled correctly") {
+    val a = Seq(1).toDF("key")
+    val b = Seq((1, "a")).toDF("key", "value")
+    val c = Seq(1).toDF("key")
+
+    val ab = a.join(b, Stream("key"), "left")
+    val abc = ab.join(c, Seq("key"), "left")
+
+    checkAnswer(abc, Row(1, "a"))
+  }
 }