Address comment.

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

@@ -105,7 +105,7 @@ abstract class Expression extends TreeNode[Expression] {
       val isNull = ctx.freshName("isNull")
       val value = ctx.freshName("value")
       val eval = doGenCode(ctx, ExprCode("",
-        VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true)),
+        VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN)),
         VariableValue(value, ExprType(ctx, dataType))))
       reduceCodeSize(ctx, eval)
       if (eval.code.nonEmpty) {
@@ -123,7 +123,7 @@ abstract class Expression extends TreeNode[Expression] {
       val setIsNull = if (!eval.isNull.isInstanceOf[LiteralValue]) {
         val globalIsNull = ctx.addMutableState(ctx.JAVA_BOOLEAN, "globalIsNull")
         val localIsNull = eval.isNull
-        eval.isNull = GlobalValue(globalIsNull, ExprType(ctx.JAVA_BOOLEAN, true))
+        eval.isNull = GlobalValue(globalIsNull, ExprType(ctx.JAVA_BOOLEAN))
         s"$globalIsNull = $localIsNull;"
       } else {
         ""

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

@@ -603,7 +603,7 @@ case class Least(children: Seq[Expression]) extends Expression {
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     val evalChildren = children.map(_.genCode(ctx))
     ev.isNull = GlobalValue(ctx.addMutableState(ctx.JAVA_BOOLEAN, ev.isNull),
-      ExprType(ctx.JAVA_BOOLEAN, true))
+      ExprType(ctx.JAVA_BOOLEAN))
     val evals = evalChildren.map(eval =>
       s"""
          |${eval.code}
@@ -683,7 +683,7 @@ case class Greatest(children: Seq[Expression]) extends Expression {
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     val evalChildren = children.map(_.genCode(ctx))
     ev.isNull = GlobalValue(ctx.addMutableState(ctx.JAVA_BOOLEAN, ev.isNull),
-      ExprType(ctx.JAVA_BOOLEAN, true))
+      ExprType(ctx.JAVA_BOOLEAN))
     val evals = evalChildren.map(eval =>
       s"""
          |${eval.code}

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

@@ -1208,7 +1208,7 @@ class CodegenContext {
       // at least two nodes) as the cost of doing it is expected to be low.
 
       subexprFunctions += s"${addNewFunction(fnName, fn)}($INPUT_ROW);"
-      val state = SubExprEliminationState(GlobalValue(isNull, ExprType(JAVA_BOOLEAN, true)),
+      val state = SubExprEliminationState(GlobalValue(isNull, ExprType(JAVA_BOOLEAN)),
         GlobalValue(value, ExprType(this, expr.dataType)))
       e.foreach(subExprEliminationExprs.put(_, state))
     }

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

@@ -30,6 +30,8 @@ abstract class ExprValue {
   // For example, a variable created outside a method can not be accessed inside the method.
   // For such cases, we may need to pass the evaluation as parameter.
   val canDirectAccess: Boolean
+
+  def isPrimitive(ctx: CodegenContext): Boolean = javaType.isPrimitive(ctx)
 }
 
 object ExprValue {
@@ -51,9 +53,9 @@ object LiteralValue {
 // A variable evaluation of [[ExprCode]].
 case class VariableValue(
     val variableName: String,
-    val javaType: ExprType,
-    val canDirectAccess: Boolean = false) extends ExprValue {
+    val javaType: ExprType) extends ExprValue {
   override def toString: String = variableName
+  override val canDirectAccess: Boolean = false
 }
 
 // A statement evaluation of [[ExprCode]].
@@ -70,13 +72,14 @@ case class GlobalValue(val value: String, val javaType: ExprType) extends ExprVa
   override val canDirectAccess: Boolean = true
 }
 
-case object TrueLiteral extends LiteralValue("true", ExprType("boolean", true))
-case object FalseLiteral extends LiteralValue("false", ExprType("boolean", true))
+case object TrueLiteral extends LiteralValue("true", ExprType("boolean"))
+case object FalseLiteral extends LiteralValue("false", ExprType("boolean"))
 
 // Represents the java type of an evaluation.
-case class ExprType(val typeName: String, val isPrimitive: Boolean)
+case class ExprType(val typeName: String) {
+  def isPrimitive(ctx: CodegenContext): Boolean = ctx.isPrimitiveType(typeName)
+}
 
 object ExprType {
-  def apply(ctx: CodegenContext, dataType: DataType): ExprType = ExprType(ctx.javaType(dataType),
-    ctx.isPrimitiveType(dataType))
+  def apply(ctx: CodegenContext, dataType: DataType): ExprType = ExprType(ctx.javaType(dataType))
 }

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

@@ -69,7 +69,7 @@ object GenerateMutableProjection extends CodeGenerator[Seq[Expression], MutableP
               |${ev.code}
               |$isNull = ${ev.isNull};
               |$value = ${ev.value};
-            """.stripMargin, GlobalValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true)), value, i)
+            """.stripMargin, GlobalValue(isNull, ExprType(ctx.JAVA_BOOLEAN)), value, i)
         } else {
           (s"""
               |${ev.code}

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

@@ -75,7 +75,7 @@ object GenerateSafeProjection extends CodeGenerator[Seq[Expression], Projection]
          |final InternalRow $output = new $rowClass($values);
        """.stripMargin
 
-    ExprCode(code, FalseLiteral, VariableValue(output, ExprType("InternalRow", false)))
+    ExprCode(code, FalseLiteral, VariableValue(output, ExprType("InternalRow")))
   }
 
   private def createCodeForArray(
@@ -106,7 +106,7 @@ object GenerateSafeProjection extends CodeGenerator[Seq[Expression], Projection]
       final ArrayData $output = new $arrayClass($values);
     """
 
-    ExprCode(code, FalseLiteral, VariableValue(output, ExprType("ArrayData", false)))
+    ExprCode(code, FalseLiteral, VariableValue(output, ExprType("ArrayData")))
   }
 
   private def createCodeForMap(
@@ -127,7 +127,7 @@ object GenerateSafeProjection extends CodeGenerator[Seq[Expression], Projection]
       final MapData $output = new $mapClass(${keyConverter.value}, ${valueConverter.value});
     """
 
-    ExprCode(code, FalseLiteral, VariableValue(output, ExprType("MapData", false)))
+    ExprCode(code, FalseLiteral, VariableValue(output, ExprType("MapData")))
   }
 
   @tailrec

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

@@ -52,7 +52,7 @@ object GenerateUnsafeProjection extends CodeGenerator[Seq[Expression], UnsafePro
     // Puts `input` in a local variable to avoid to re-evaluate it if it's a statement.
     val tmpInput = ctx.freshName("tmpInput")
     val fieldEvals = fieldTypes.zipWithIndex.map { case (dt, i) =>
-      ExprCode("", StatementValue(s"$tmpInput.isNullAt($i)", ExprType(ctx.JAVA_BOOLEAN, true)),
+      ExprCode("", StatementValue(s"$tmpInput.isNullAt($i)", ExprType(ctx.JAVA_BOOLEAN)),
         StatementValue(ctx.getValue(tmpInput, dt, i.toString), ExprType(ctx, dt)))
     }
 
@@ -348,7 +348,7 @@ object GenerateUnsafeProjection extends CodeGenerator[Seq[Expression], UnsafePro
         $writeExpressions
         $updateRowSize
       """
-    ExprCode(code, FalseLiteral, GlobalValue(result, ExprType("UnsafeRow", false)))
+    ExprCode(code, FalseLiteral, GlobalValue(result, ExprType("UnsafeRow")))
   }
 
   protected def canonicalize(in: Seq[Expression]): Seq[Expression] =

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

@@ -73,7 +73,7 @@ case class Coalesce(children: Seq[Expression]) extends Expression {
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     ev.isNull = GlobalValue(ctx.addMutableState(ctx.JAVA_BOOLEAN, ev.isNull),
-      ExprType(ctx.JAVA_BOOLEAN, true))
+      ExprType(ctx.JAVA_BOOLEAN))
 
     // all the evals are meant to be in a do { ... } while (false); loop
     val evals = children.map { e =>
@@ -323,9 +323,9 @@ case class IsNull(child: Expression) extends UnaryExpression with Predicate {
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     val eval = child.genCode(ctx)
     val value = if (eval.isNull.isInstanceOf[LiteralValue]) {
-      LiteralValue(eval.isNull, ExprType(ctx.JAVA_BOOLEAN, true))
+      LiteralValue(eval.isNull, ExprType(ctx.JAVA_BOOLEAN))
     } else {
-      VariableValue(eval.isNull, ExprType(ctx.JAVA_BOOLEAN, true))
+      VariableValue(eval.isNull, ExprType(ctx.JAVA_BOOLEAN))
     }
     ExprCode(code = eval.code, isNull = FalseLiteral, value = value)
   }

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

@@ -63,7 +63,7 @@ trait InvokeLike extends Expression with NonSQLExpression {
 
     val resultIsNull = if (needNullCheck) {
       val resultIsNull = ctx.addMutableState(ctx.JAVA_BOOLEAN, "resultIsNull")
-      GlobalValue(resultIsNull, ExprType(ctx.JAVA_BOOLEAN, true))
+      GlobalValue(resultIsNull, ExprType(ctx.JAVA_BOOLEAN))
     } else {
       FalseLiteral
     }
@@ -445,7 +445,7 @@ case class LambdaVariable(
 
   override def genCode(ctx: CodegenContext): ExprCode = {
     val isNullValue = if (nullable) {
-      VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true))
+      VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN))
     } else {
       FalseLiteral
     }

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

@@ -28,15 +28,17 @@ class ExprValueSuite extends SparkFunSuite {
     assert(trueLit.value == "true")
     assert(falseLit.value == "false")
 
+    val ctx = new CodegenContext()
+
     trueLit match {
       case LiteralValue(value, javaType) =>
-        assert(value == "true" && javaType.typeName == "boolean" && javaType.isPrimitive)
+        assert(value == "true" && javaType.typeName == "boolean" && javaType.isPrimitive(ctx))
       case _ => fail()
     }
 
     falseLit match {
       case LiteralValue(value, javaType) =>
-        assert(value == "false" && javaType.typeName == "boolean" && javaType.isPrimitive)
+        assert(value == "false" && javaType.typeName == "boolean" && javaType.isPrimitive(ctx))
       case _ => fail()
     }
   }

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

@@ -52,7 +52,7 @@ private[sql] trait ColumnarBatchScan extends CodegenSupport {
     val javaType = ctx.javaType(dataType)
     val value = ctx.getValueFromVector(columnVar, dataType, ordinal)
     val isNullVar = if (nullable) {
-      VariableValue(ctx.freshName("isNull"), ExprType(ctx.JAVA_BOOLEAN, true))
+      VariableValue(ctx.freshName("isNull"), ExprType(ctx.JAVA_BOOLEAN))
     } else {
       FalseLiteral
     }

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

@@ -156,7 +156,7 @@ case class ExpandExec(
           |boolean $isNull = true;
           |${ctx.javaType(firstExpr.dataType)} $value = ${ctx.defaultValue(firstExpr.dataType)};
          """.stripMargin
-        ExprCode(code, VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true)),
+        ExprCode(code, VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN)),
           VariableValue(value, ExprType(ctx, firstExpr.dataType)))
       }
     }

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

@@ -170,10 +170,10 @@ case class GenerateExec(
     // Add position
     val position = if (e.position) {
       if (outer) {
-        Seq(ExprCode("", StatementValue(s"$index == -1", ExprType(ctx.JAVA_BOOLEAN, true)),
-          VariableValue(index, ExprType(ctx.JAVA_INT, true))))
+        Seq(ExprCode("", StatementValue(s"$index == -1", ExprType(ctx.JAVA_BOOLEAN)),
+          VariableValue(index, ExprType(ctx.JAVA_INT))))
       } else {
-        Seq(ExprCode("", FalseLiteral, VariableValue(index, ExprType(ctx.JAVA_INT, true))))
+        Seq(ExprCode("", FalseLiteral, VariableValue(index, ExprType(ctx.JAVA_INT))))
       }
     } else {
       Seq.empty
@@ -316,7 +316,7 @@ case class GenerateExec(
            |boolean $isNull = ${checks.mkString(" || ")};
            |$javaType $value = $isNull ? ${ctx.defaultValue(dt)} : $getter;
          """.stripMargin
-      ExprCode(code, VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true)),
+      ExprCode(code, VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN)),
         VariableValue(value, ExprType(ctx, dt)))
     } else {
       ExprCode(s"$javaType $value = $getter;", FalseLiteral,

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

@@ -111,7 +111,7 @@ trait CodegenSupport extends SparkPlan {
 
   private def prepareRowVar(ctx: CodegenContext, row: String, colVars: Seq[ExprCode]): ExprCode = {
     if (row != null) {
-      ExprCode("", FalseLiteral, VariableValue(row, ExprType("UnsafeRow", false)))
+      ExprCode("", FalseLiteral, VariableValue(row, ExprType("UnsafeRow")))
     } else {
       if (colVars.nonEmpty) {
         val colExprs = output.zipWithIndex.map { case (attr, i) =>
@@ -129,7 +129,7 @@ trait CodegenSupport extends SparkPlan {
         ExprCode(code, FalseLiteral, ev.value)
       } else {
         // There is no columns
-        ExprCode("", FalseLiteral, VariableValue("unsafeRow", ExprType("UnsafeRow", false)))
+        ExprCode("", FalseLiteral, VariableValue("unsafeRow", ExprType("UnsafeRow")))
       }
     }
   }
@@ -245,7 +245,7 @@ trait CodegenSupport extends SparkPlan {
         val isNull = ctx.freshName(s"exprIsNull_$i")
         arguments += ev.isNull
         parameters += s"boolean $isNull"
-        VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true))
+        VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN))
       }
 
       paramVars += ExprCode("", paramIsNull,

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala

@@ -194,7 +194,7 @@ case class HashAggregateExec(
          | $isNull = ${ev.isNull};
          | $value = ${ev.value};
        """.stripMargin
-      ExprCode(ev.code + initVars, GlobalValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true)),
+      ExprCode(ev.code + initVars, GlobalValue(isNull, ExprType(ctx.JAVA_BOOLEAN)),
         GlobalValue(value, ExprType(ctx, e.dataType)))
     }
     val initBufVar = evaluateVariables(bufVars)

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashMapGenerator.scala

@@ -54,7 +54,7 @@ abstract class HashMapGenerator(
            | $isNull = ${ev.isNull};
            | $value = ${ev.value};
        """.stripMargin
-      ExprCode(ev.code + initVars, GlobalValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true)),
+      ExprCode(ev.code + initVars, GlobalValue(isNull, ExprType(ctx.JAVA_BOOLEAN)),
         GlobalValue(value, ExprType(ctx, e.dataType)))
     }
   }

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

@@ -368,7 +368,7 @@ case class RangeExec(range: org.apache.spark.sql.catalyst.plans.logical.Range)
     val number = ctx.addMutableState(ctx.JAVA_LONG, "number")
 
     val value = ctx.freshName("value")
-    val ev = ExprCode("", FalseLiteral, VariableValue(value, ExprType(ctx.JAVA_LONG, true)))
+    val ev = ExprCode("", FalseLiteral, VariableValue(value, ExprType(ctx.JAVA_LONG)))
     val BigInt = classOf[java.math.BigInteger].getName
 
     // Inline mutable state since not many Range operations in a task

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala

@@ -191,7 +191,7 @@ case class BroadcastHashJoinExec(
           |  $value = ${ev.value};
           |}
          """.stripMargin
-        ExprCode(code, VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true)),
+        ExprCode(code, VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN)),
           VariableValue(value, ExprType(ctx, a.dataType)))
       }
     }
@@ -488,7 +488,7 @@ case class BroadcastHashJoinExec(
     }
 
     val resultVar = input ++ Seq(ExprCode("", FalseLiteral,
-      VariableValue(existsVar, ExprType(ctx.JAVA_BOOLEAN, true))))
+      VariableValue(existsVar, ExprType(ctx.JAVA_BOOLEAN))))
     if (broadcastRelation.value.keyIsUnique) {
       s"""
          |// generate join key for stream side

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

@@ -531,7 +531,7 @@ case class SortMergeJoinExec(
              |boolean $isNull = false;
              |$javaType $value = $defaultValue;
            """.stripMargin
-        (ExprCode(code, VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN, true)),
+        (ExprCode(code, VariableValue(isNull, ExprType(ctx.JAVA_BOOLEAN)),
           VariableValue(value, ExprType(ctx, a.dataType))), leftVarsDecl)
       } else {
         val code = s"$value = $valueCode;"