[SPARK-9462][SQL] Initialize nondeterministic expressions in code gen fallback mode.

Author: rxin

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateMutableProjection.scala

@@ -122,6 +122,11 @@ object GenerateMutableProjection extends CodeGenerator[Seq[Expression], () => Mu
 
     logDebug(s"code for ${expressions.mkString(",")}:\n${CodeFormatter.format(code)}")
 
+    ctx.references.foreach {
+      case n: Nondeterministic => n.setInitialValues()
+      case _ =>
+    }
+
     val c = compile(code)
     () => {
       c.generate(ctx.references.toArray).asInstanceOf[MutableProjection]

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GeneratePredicate.scala

@@ -62,6 +62,11 @@ object GeneratePredicate extends CodeGenerator[Expression, (InternalRow) => Bool
 
     logDebug(s"Generated predicate '$predicate':\n${CodeFormatter.format(code)}")
 
+    ctx.references.foreach {
+      case n: Nondeterministic => n.setInitialValues()
+      case _ =>
+    }
+
     val p = compile(code).generate(ctx.references.toArray).asInstanceOf[Predicate]
     (r: InternalRow) => p.eval(r)
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateProjection.scala

@@ -233,6 +233,11 @@ object GenerateProjection extends CodeGenerator[Seq[Expression], Projection] {
     logDebug(s"MutableRow, initExprs: ${expressions.mkString(",")} code:\n" +
       CodeFormatter.format(code))
 
+    ctx.references.foreach {
+      case n: Nondeterministic => n.setInitialValues()
+      case _ =>
+    }
+
     compile(code).generate(ctx.references.toArray).asInstanceOf[Projection]
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateUnsafeProjection.scala

@@ -285,6 +285,11 @@ object GenerateUnsafeProjection extends CodeGenerator[Seq[Expression], UnsafePro
 
     logDebug(s"code for ${expressions.mkString(",")}:\n${CodeFormatter.format(code)}")
 
+    ctx.references.foreach {
+      case n: Nondeterministic => n.setInitialValues()
+      case _ =>
+    }
+
     val c = compile(code)
     c.generate(ctx.references.toArray).asInstanceOf[UnsafeProjection]
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodegenExpressionCachingSuite.scala

@@ -19,15 +19,39 @@ package org.apache.spark.sql.catalyst.expressions.codegen
 
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{UnsafeProjection, LeafExpression}
+import org.apache.spark.sql.catalyst.expressions.{Nondeterministic, UnsafeProjection, LeafExpression}
 import org.apache.spark.sql.types.{BooleanType, DataType}
 
 /**
  * A test suite that makes sure code generation handles expression internally states correctly.
  */
 class CodegenExpressionCachingSuite extends SparkFunSuite {
 
-  test("GenerateUnsafeProjection") {
+  test("GenerateUnsafeProjection should initialize expressions") {
+    val expr = NondeterministicExpression()
+    val instance = UnsafeProjection.create(Seq(expr))
+    assert(instance.apply(null).getBoolean(0) === false)
+  }
+
+  test("GenerateProjection should initialize expressions") {
+    val expr = NondeterministicExpression()
+    val instance = GenerateProjection.generate(Seq(expr))
+    assert(instance.apply(null).getBoolean(0) === false)
+  }
+
+  test("GenerateMutableProjection should initialize expressions") {
+    val expr = NondeterministicExpression()
+    val instance = GenerateMutableProjection.generate(Seq(expr))()
+    assert(instance.apply(null).getBoolean(0) === false)
+  }
+
+  test("GeneratePredicate should initialize expressions") {
+    val expr = NondeterministicExpression()
+    val instance = GeneratePredicate.generate(expr)
+    assert(instance.apply(null) === false)
+  }
+
+  test("GenerateUnsafeProjection should not share expression instances") {
     val expr1 = MutableExpression()
     val instance1 = UnsafeProjection.create(Seq(expr1))
     assert(instance1.apply(null).getBoolean(0) === false)
@@ -39,7 +63,7 @@ class CodegenExpressionCachingSuite extends SparkFunSuite {
     assert(instance2.apply(null).getBoolean(0) === true)
   }
 
-  test("GenerateProjection") {
+  test("GenerateProjection should not share expression instances") {
     val expr1 = MutableExpression()
     val instance1 = GenerateProjection.generate(Seq(expr1))
     assert(instance1.apply(null).getBoolean(0) === false)
@@ -51,7 +75,7 @@ class CodegenExpressionCachingSuite extends SparkFunSuite {
     assert(instance2.apply(null).getBoolean(0) === true)
   }
 
-  test("GenerateMutableProjection") {
+  test("GenerateMutableProjection should not share expression instances") {
     val expr1 = MutableExpression()
     val instance1 = GenerateMutableProjection.generate(Seq(expr1))()
     assert(instance1.apply(null).getBoolean(0) === false)
@@ -63,7 +87,7 @@ class CodegenExpressionCachingSuite extends SparkFunSuite {
     assert(instance2.apply(null).getBoolean(0) === true)
   }
 
-  test("GeneratePredicate") {
+  test("GeneratePredicate should not share expression instances") {
     val expr1 = MutableExpression()
     val instance1 = GeneratePredicate.generate(expr1)
     assert(instance1.apply(null) === false)
@@ -77,6 +101,17 @@ class CodegenExpressionCachingSuite extends SparkFunSuite {
 
 }
 
+/**
+ * An expression that's non-deterministic and doesn't support codegen.
+ */
+case class NondeterministicExpression()
+  extends LeafExpression with Nondeterministic with CodegenFallback {
+  override protected def initInternal(): Unit = {}
+  override protected def evalInternal(input: InternalRow): Any = false
+  override def nullable: Boolean = false
+  override def dataType: DataType = BooleanType
+}
+
 
 /**
  * An expression with mutable state so we can change it freely in our test suite.