disable codegen for ScalaUDF

Author: Davies Liu

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala

@@ -957,4 +957,6 @@ case class ScalaUdf(function: AnyRef, dataType: DataType, children: Seq[Expressi
   private[this] val converter = CatalystTypeConverters.createToCatalystConverter(dataType)
   override def eval(input: Row): Any = converter(f(input))
 
+  // TODO(davies): make ScalaUdf work with codegen
+  override def isThreadSafe: Boolean = false
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/arithmetic.scala

@@ -349,7 +349,7 @@ case class MaxOf(left: Expression, right: Expression) extends BinaryArithmetic {
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = {
     val eval1 = left.gen(ctx)
     val eval2 = right.gen(ctx)
-    val compCode = ctx.compFunc(dataType)(eval1.primitive, eval2.primitive)
+    val compCode = ctx.genCmop(dataType, eval1.primitive, eval2.primitive)
 
     eval1.code + eval2.code + s"""
       boolean ${ev.isNull} = false;
@@ -401,7 +401,7 @@ case class MinOf(left: Expression, right: Expression) extends BinaryArithmetic {
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = {
     val eval1 = left.gen(ctx)
     val eval2 = right.gen(ctx)
-    val compCode = ctx.compFunc(dataType)(eval1.primitive, eval2.primitive)
+    val compCode = ctx.genCmop(dataType, eval1.primitive, eval2.primitive)
 
     eval1.code + eval2.code + s"""
       boolean ${ev.isNull} = false;

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

@@ -161,28 +161,21 @@ class CodeGenContext {
   /**
    * Returns a function to generate equal expression in Java
    */
-  def equalFunc(dataType: DataType): ((String, String) => String) = dataType match {
-    case BinaryType => { case (eval1, eval2) =>
-      s"java.util.Arrays.equals($eval1, $eval2)" }
+  def genEqual(dataType: DataType, c1: String, c2: String): String = dataType match {
+    case BinaryType =>  s"java.util.Arrays.equals($c1, $c2)"
     case IntegerType | BooleanType | LongType | DoubleType | FloatType | ShortType | ByteType
-         | DateType =>
-      { case (eval1, eval2) => s"$eval1 == $eval2" }
-    case other =>
-      { case (eval1, eval2) => s"$eval1.equals($eval2)" }
+         | DateType => s"$c1 == $c2"
+    case other => s"$c1.equals($c2)"
   }
 
   /**
    * Return a function to generate compare expression in Java
    */
-  def compFunc(dataType: DataType): (String, String) => String = dataType match {
-    case BinaryType => {
-      case (c1, c2) =>
-        s"org.apache.spark.sql.catalyst.util.TypeUtils.compareBinary($c1, $c2)"
-    }
-    case IntegerType | LongType | DoubleType | FloatType | ShortType | ByteType | DateType => {
-      case (c1, c2) => s"$c1 - $c2"
-    }
-    case other => { case (c1, c2) => s"$c1.compare($c2)" }
+  def genCmop(dataType: DataType, c1: String, c2: String): String = dataType match {
+    case BinaryType => s"org.apache.spark.sql.catalyst.util.TypeUtils.compareBinary($c1, $c2)"
+    case IntegerType | LongType | DoubleType | FloatType | ShortType | ByteType | DateType =>
+      s"$c1 - $c2"
+    case other => s"$c1.compare($c2)"
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypes.scala

@@ -61,6 +61,7 @@ case class CreateStruct(children: Seq[NamedExpression]) extends Expression {
   override lazy val dataType: StructType = {
     assert(resolved,
       s"CreateStruct contains unresolvable children: ${children.filterNot(_.resolved)}.")
+    println(s"$children")
     val fields = children.map { child =>
       StructField(child.name, child.dataType, child.nullable, child.metadata)
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/conditionals.scala

@@ -261,7 +261,7 @@ case class CaseKeyWhen(key: Expression, branches: Seq[Expression]) extends CaseW
           ${cond.code}
           if (${keyEval.isNull} && ${cond.isNull} ||
             !${keyEval.isNull} && !${cond.isNull}
-             && ${ctx.equalFunc(key.dataType)(keyEval.primitive, cond.primitive)}) {
+             && ${ctx.genEqual(key.dataType, keyEval.primitive, cond.primitive)}) {
             $got = true;
             ${res.code}
             ${ev.isNull} = ${res.isNull};

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala

@@ -250,37 +250,9 @@ abstract class BinaryComparison extends BinaryExpression with Predicate {
   }
 
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = {
-    left.dataType match {
-      case dt: NumericType if ctx.isNativeType(dt) => defineCodeGen (ctx, ev, {
-        (c1, c3) => s"$c1 $symbol $c3"
-      })
-      case DateType => defineCodeGen (ctx, ev, {
-        (c1, c3) => s"$c1 $symbol $c3"
-      })
-      case BinaryType =>
-        val eval1 = left.gen(ctx)
-        val eval2 = right.gen(ctx)
-        s"""
-          ${eval1.code}
-          boolean ${ev.isNull} = ${eval1.isNull};
-          boolean ${ev.primitive} = ${ctx.defaultValue(dataType)};
-          if (!${ev.isNull}) {
-            ${eval2.code}
-            if (!${eval2.isNull}) {
-              ${ev.primitive} = org.apache.spark.sql.catalyst.util.TypeUtils.compareBinary(
-                ${eval1.primitive}, ${eval2.primitive}) $symbol 0;
-            } else {
-              ${ev.isNull} = true;
-            }
-          }
-        """
-      case TimestampType =>
-        // java.sql.Timestamp does not have compare()
-        super.genCode(ctx, ev)
-      case other => defineCodeGen (ctx, ev, {
-        (c1, c2) => s"$c1.compare($c2) $symbol 0"
-      })
-    }
+    defineCodeGen(ctx, ev, {
+      (c1, c2) => s"${ctx.genCmop(left.dataType, c1, c2)} $symbol 0"
+    })
   }
 
   protected def evalInternal(evalE1: Any, evalE2: Any): Any =
@@ -301,7 +273,7 @@ case class EqualTo(left: Expression, right: Expression) extends BinaryComparison
     else java.util.Arrays.equals(l.asInstanceOf[Array[Byte]], r.asInstanceOf[Array[Byte]])
   }
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = {
-    defineCodeGen(ctx, ev, ctx.equalFunc(left.dataType))
+    defineCodeGen(ctx, ev, (c1, c2) => ctx.genEqual(left.dataType, c1, c2))
   }
 }
 
@@ -327,7 +299,7 @@ case class EqualNullSafe(left: Expression, right: Expression) extends BinaryComp
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = {
     val eval1 = left.gen(ctx)
     val eval2 = right.gen(ctx)
-    val equalCode = ctx.equalFunc(left.dataType)(eval1.primitive, eval2.primitive)
+    val equalCode = ctx.genEqual(left.dataType, eval1.primitive, eval2.primitive)
     ev.isNull = "false"
     eval1.code + eval2.code + s"""
         boolean ${ev.primitive} = (${eval1.isNull} && ${eval2.isNull}) ||