diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala
index df3ab05e02c76..77582e10f9ff2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.expressions
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.errors.attachTree
-import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, ExprCode, FalseLiteral}
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, ExprCode, FalseLiteral, JavaCode}
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
 import org.apache.spark.sql.types._
 
@@ -53,7 +53,7 @@ case class BoundReference(ordinal: Int, dataType: DataType, nullable: Boolean)
       ev.copy(code = oev.code)
     } else {
       assert(ctx.INPUT_ROW != null, "INPUT_ROW and currentVars cannot both be null.")
-      val javaType = CodeGenerator.javaType(dataType)
+      val javaType = JavaCode.javaType(dataType)
       val value = CodeGenerator.getValue(ctx.INPUT_ROW, dataType, ordinal.toString)
       if (nullable) {
         ev.copy(code =
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala
index 7971ae602bd37..c5c2ee62c5cd8 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala
@@ -645,25 +645,21 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
     val eval = child.genCode(ctx)
     val nullSafeCast = nullSafeCastFunction(child.dataType, dataType, ctx)
 
-    ev.copy(code =
-      code"""
-        ${eval.code}
-        // This comment is added for manually tracking reference of ${eval.value}, ${eval.isNull}
-        ${castCode(ctx, eval.value, eval.isNull, ev.value, ev.isNull, dataType, nullSafeCast)}
-      """)
+    ev.copy(code = eval.code +
+      castCode(ctx, eval.value, eval.isNull, ev.value, ev.isNull, dataType, nullSafeCast))
   }
 
   // The function arguments are: `input`, `result` and `resultIsNull`. We don't need `inputIsNull`
   // in parameter list, because the returned code will be put in null safe evaluation region.
-  private[this] type CastFunction = (String, String, String) => String
+  private[this] type CastFunction = (ExprValue, ExprValue, ExprValue) => Block
 
   private[this] def nullSafeCastFunction(
       from: DataType,
       to: DataType,
       ctx: CodegenContext): CastFunction = to match {
 
-    case _ if from == NullType => (c, evPrim, evNull) => s"$evNull = true;"
-    case _ if to == from => (c, evPrim, evNull) => s"$evPrim = $c;"
+    case _ if from == NullType => (c, evPrim, evNull) => code"$evNull = true;"
+    case _ if to == from => (c, evPrim, evNull) => code"$evPrim = $c;"
     case StringType => castToStringCode(from, ctx)
     case BinaryType => castToBinaryCode(from)
     case DateType => castToDateCode(from, ctx)
@@ -684,18 +680,19 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
     case struct: StructType => castStructCode(from.asInstanceOf[StructType], struct, ctx)
     case udt: UserDefinedType[_]
       if udt.userClass == from.asInstanceOf[UserDefinedType[_]].userClass =>
-      (c, evPrim, evNull) => s"$evPrim = $c;"
+      (c, evPrim, evNull) => code"$evPrim = $c;"
     case _: UserDefinedType[_] =>
       throw new SparkException(s"Cannot cast $from to $to.")
   }
 
   // Since we need to cast input expressions recursively inside ComplexTypes, such as Map's
   // Key and Value, Struct's field, we need to name out all the variable names involved in a cast.
-  private[this] def castCode(ctx: CodegenContext, input: String, inputIsNull: String,
-    result: String, resultIsNull: String, resultType: DataType, cast: CastFunction): String = {
-    s"""
+  private[this] def castCode(ctx: CodegenContext, input: ExprValue, inputIsNull: ExprValue,
+    result: ExprValue, resultIsNull: ExprValue, resultType: DataType, cast: CastFunction): Block = {
+    val javaType = JavaCode.javaType(resultType)
+    code"""
       boolean $resultIsNull = $inputIsNull;
-      ${CodeGenerator.javaType(resultType)} $result = ${CodeGenerator.defaultValue(resultType)};
+      $javaType $result = ${CodeGenerator.defaultValue(resultType)};
       if (!$inputIsNull) {
         ${cast(input, result, resultIsNull)}
       }
@@ -704,22 +701,24 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
 
   private def writeArrayToStringBuilder(
       et: DataType,
-      array: String,
-      buffer: String,
-      ctx: CodegenContext): String = {
+      array: ExprValue,
+      buffer: ExprValue,
+      ctx: CodegenContext): Block = {
     val elementToStringCode = castToStringCode(et, ctx)
     val funcName = ctx.freshName("elementToString")
-    val elementToStringFunc = ctx.addNewFunction(funcName,
+    val element = JavaCode.variable("element", et)
+    val elementStr = JavaCode.variable("elementStr", StringType)
+    val elementToStringFunc = inline"${ctx.addNewFunction(funcName,
       s"""
-         |private UTF8String $funcName(${CodeGenerator.javaType(et)} element) {
-         |  UTF8String elementStr = null;
-         |  ${elementToStringCode("element", "elementStr", null /* resultIsNull won't be used */)}
+         |private UTF8String $funcName(${CodeGenerator.javaType(et)} $element) {
+         |  UTF8String $elementStr = null;
+         |  ${elementToStringCode(element, elementStr, null /* resultIsNull won't be used */)}
          |  return elementStr;
          |}
-       """.stripMargin)
+       """.stripMargin)}"
 
-    val loopIndex = ctx.freshName("loopIndex")
-    s"""
+    val loopIndex = ctx.freshVariable("loopIndex", IntegerType)
+    code"""
        |$buffer.append("[");
        |if ($array.numElements() > 0) {
        |  if (!$array.isNullAt(0)) {
@@ -740,31 +739,37 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
   private def writeMapToStringBuilder(
       kt: DataType,
       vt: DataType,
-      map: String,
-      buffer: String,
-      ctx: CodegenContext): String = {
+      map: ExprValue,
+      buffer: ExprValue,
+      ctx: CodegenContext): Block = {
 
     def dataToStringFunc(func: String, dataType: DataType) = {
       val funcName = ctx.freshName(func)
       val dataToStringCode = castToStringCode(dataType, ctx)
-      ctx.addNewFunction(funcName,
+      val data = JavaCode.variable("data", dataType)
+      val dataStr = JavaCode.variable("dataStr", StringType)
+      val functionCall = ctx.addNewFunction(funcName,
         s"""
-           |private UTF8String $funcName(${CodeGenerator.javaType(dataType)} data) {
-           |  UTF8String dataStr = null;
-           |  ${dataToStringCode("data", "dataStr", null /* resultIsNull won't be used */)}
+           |private UTF8String $funcName(${CodeGenerator.javaType(dataType)} $data) {
+           |  UTF8String $dataStr = null;
+           |  ${dataToStringCode(data, dataStr, null /* resultIsNull won't be used */)}
            |  return dataStr;
            |}
          """.stripMargin)
+      inline"$functionCall"
     }
 
     val keyToStringFunc = dataToStringFunc("keyToString", kt)
     val valueToStringFunc = dataToStringFunc("valueToString", vt)
-    val loopIndex = ctx.freshName("loopIndex")
-    val getMapFirstKey = CodeGenerator.getValue(s"$map.keyArray()", kt, "0")
-    val getMapFirstValue = CodeGenerator.getValue(s"$map.valueArray()", vt, "0")
-    val getMapKeyArray = CodeGenerator.getValue(s"$map.keyArray()", kt, loopIndex)
-    val getMapValueArray = CodeGenerator.getValue(s"$map.valueArray()", vt, loopIndex)
-    s"""
+    val loopIndex = ctx.freshVariable("loopIndex", IntegerType)
+    val mapKeyArray = JavaCode.expression(s"$map.keyArray()", classOf[ArrayData])
+    val mapValueArray = JavaCode.expression(s"$map.valueArray()", classOf[ArrayData])
+    val getMapFirstKey = CodeGenerator.getValue(mapKeyArray, kt, JavaCode.literal("0", IntegerType))
+    val getMapFirstValue = CodeGenerator.getValue(mapValueArray, vt,
+      JavaCode.literal("0", IntegerType))
+    val getMapKeyArray = CodeGenerator.getValue(mapKeyArray, kt, loopIndex)
+    val getMapValueArray = CodeGenerator.getValue(mapValueArray, vt, loopIndex)
+    code"""
        |$buffer.append("[");
        |if ($map.numElements() > 0) {
        |  $buffer.append($keyToStringFunc($getMapFirstKey));
@@ -789,20 +794,21 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
 
   private def writeStructToStringBuilder(
       st: Seq[DataType],
-      row: String,
-      buffer: String,
-      ctx: CodegenContext): String = {
+      row: ExprValue,
+      buffer: ExprValue,
+      ctx: CodegenContext): Block = {
     val structToStringCode = st.zipWithIndex.map { case (ft, i) =>
       val fieldToStringCode = castToStringCode(ft, ctx)
-      val field = ctx.freshName("field")
-      val fieldStr = ctx.freshName("fieldStr")
-      s"""
-         |${if (i != 0) s"""$buffer.append(",");""" else ""}
+      val field = ctx.freshVariable("field", ft)
+      val fieldStr = ctx.freshVariable("fieldStr", StringType)
+      val javaType = JavaCode.javaType(ft)
+      code"""
+         |${if (i != 0) code"""$buffer.append(",");""" else EmptyBlock}
          |if (!$row.isNullAt($i)) {
-         |  ${if (i != 0) s"""$buffer.append(" ");""" else ""}
+         |  ${if (i != 0) code"""$buffer.append(" ");""" else EmptyBlock}
          |
          |  // Append $i field into the string buffer
-         |  ${CodeGenerator.javaType(ft)} $field = ${CodeGenerator.getValue(row, ft, s"$i")};
+         |  $javaType $field = ${CodeGenerator.getValue(row, ft, s"$i")};
          |  UTF8String $fieldStr = null;
          |  ${fieldToStringCode(field, fieldStr, null /* resultIsNull won't be used */)}
          |  $buffer.append($fieldStr);
@@ -811,11 +817,12 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
     }
 
     val writeStructCode = ctx.splitExpressions(
-      expressions = structToStringCode,
+      expressions = structToStringCode.map(_.code),
       funcName = "fieldToString",
-      arguments = ("InternalRow", row) :: (classOf[UTF8StringBuilder].getName, buffer) :: Nil)
+      arguments = ("InternalRow", row.code) ::
+        (classOf[UTF8StringBuilder].getName, buffer.code) :: Nil)
 
-    s"""
+    code"""
        |$buffer.append("[");
        |$writeStructCode
        |$buffer.append("]");
@@ -825,20 +832,20 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
   private[this] def castToStringCode(from: DataType, ctx: CodegenContext): CastFunction = {
     from match {
       case BinaryType =>
-        (c, evPrim, evNull) => s"$evPrim = UTF8String.fromBytes($c);"
+        (c, evPrim, evNull) => code"$evPrim = UTF8String.fromBytes($c);"
       case DateType =>
-        (c, evPrim, evNull) => s"""$evPrim = UTF8String.fromString(
+        (c, evPrim, evNull) => code"""$evPrim = UTF8String.fromString(
           org.apache.spark.sql.catalyst.util.DateTimeUtils.dateToString($c));"""
       case TimestampType =>
-        val tz = ctx.addReferenceObj("timeZone", timeZone)
-        (c, evPrim, evNull) => s"""$evPrim = UTF8String.fromString(
+        val tz = JavaCode.global(ctx.addReferenceObj("timeZone", timeZone), timeZone.getClass)
+        (c, evPrim, evNull) => code"""$evPrim = UTF8String.fromString(
           org.apache.spark.sql.catalyst.util.DateTimeUtils.timestampToString($c, $tz));"""
       case ArrayType(et, _) =>
         (c, evPrim, evNull) => {
-          val buffer = ctx.freshName("buffer")
-          val bufferClass = classOf[UTF8StringBuilder].getName
+          val buffer = ctx.freshVariable("buffer", classOf[UTF8StringBuilder])
+          val bufferClass = JavaCode.javaType(classOf[UTF8StringBuilder])
           val writeArrayElemCode = writeArrayToStringBuilder(et, c, buffer, ctx)
-          s"""
+          code"""
              |$bufferClass $buffer = new $bufferClass();
              |$writeArrayElemCode;
              |$evPrim = $buffer.build();
@@ -846,10 +853,10 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
         }
       case MapType(kt, vt, _) =>
         (c, evPrim, evNull) => {
-          val buffer = ctx.freshName("buffer")
-          val bufferClass = classOf[UTF8StringBuilder].getName
+          val buffer = ctx.freshVariable("buffer", classOf[UTF8StringBuilder])
+          val bufferClass = JavaCode.javaType(classOf[UTF8StringBuilder])
           val writeMapElemCode = writeMapToStringBuilder(kt, vt, c, buffer, ctx)
-          s"""
+          code"""
              |$bufferClass $buffer = new $bufferClass();
              |$writeMapElemCode;
              |$evPrim = $buffer.build();
@@ -857,11 +864,11 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
         }
       case StructType(fields) =>
         (c, evPrim, evNull) => {
-          val row = ctx.freshName("row")
-          val buffer = ctx.freshName("buffer")
-          val bufferClass = classOf[UTF8StringBuilder].getName
+          val row = ctx.freshVariable("row", classOf[InternalRow])
+          val buffer = ctx.freshVariable("buffer", classOf[UTF8StringBuilder])
+          val bufferClass = JavaCode.javaType(classOf[UTF8StringBuilder])
           val writeStructCode = writeStructToStringBuilder(fields.map(_.dataType), row, buffer, ctx)
-          s"""
+          code"""
              |InternalRow $row = $c;
              |$bufferClass $buffer = new $bufferClass();
              |$writeStructCode
@@ -870,26 +877,26 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
         }
       case pudt: PythonUserDefinedType => castToStringCode(pudt.sqlType, ctx)
       case udt: UserDefinedType[_] =>
-        val udtRef = ctx.addReferenceObj("udt", udt)
+        val udtRef = JavaCode.global(ctx.addReferenceObj("udt", udt), udt.sqlType)
         (c, evPrim, evNull) => {
-          s"$evPrim = UTF8String.fromString($udtRef.deserialize($c).toString());"
+          code"$evPrim = UTF8String.fromString($udtRef.deserialize($c).toString());"
         }
       case _ =>
-        (c, evPrim, evNull) => s"$evPrim = UTF8String.fromString(String.valueOf($c));"
+        (c, evPrim, evNull) => code"$evPrim = UTF8String.fromString(String.valueOf($c));"
     }
   }
 
   private[this] def castToBinaryCode(from: DataType): CastFunction = from match {
     case StringType =>
-      (c, evPrim, evNull) => s"$evPrim = $c.getBytes();"
+      (c, evPrim, evNull) => code"$evPrim = $c.getBytes();"
   }
 
   private[this] def castToDateCode(
       from: DataType,
       ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val intOpt = ctx.freshName("intOpt")
-      (c, evPrim, evNull) => s"""
+      val intOpt = ctx.freshVariable("intOpt", classOf[Option[Integer]])
+      (c, evPrim, evNull) => code"""
         scala.Option<Integer> $intOpt =
           org.apache.spark.sql.catalyst.util.DateTimeUtils.stringToDate($c);
         if ($intOpt.isDefined()) {
@@ -899,16 +906,17 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
         }
        """
     case TimestampType =>
-      val tz = ctx.addReferenceObj("timeZone", timeZone)
+      val tz = JavaCode.global(ctx.addReferenceObj("timeZone", timeZone), timeZone.getClass)
       (c, evPrim, evNull) =>
-        s"$evPrim = org.apache.spark.sql.catalyst.util.DateTimeUtils.millisToDays($c / 1000L, $tz);"
+        code"""$evPrim =
+          org.apache.spark.sql.catalyst.util.DateTimeUtils.millisToDays($c / 1000L, $tz);"""
     case _ =>
-      (c, evPrim, evNull) => s"$evNull = true;"
+      (c, evPrim, evNull) => code"$evNull = true;"
   }
 
-  private[this] def changePrecision(d: String, decimalType: DecimalType,
-      evPrim: String, evNull: String): String =
-    s"""
+  private[this] def changePrecision(d: ExprValue, decimalType: DecimalType,
+      evPrim: ExprValue, evNull: ExprValue): Block =
+    code"""
       if ($d.changePrecision(${decimalType.precision}, ${decimalType.scale})) {
         $evPrim = $d;
       } else {
@@ -920,11 +928,11 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
       from: DataType,
       target: DecimalType,
       ctx: CodegenContext): CastFunction = {
-    val tmp = ctx.freshName("tmpDecimal")
+    val tmp = ctx.freshVariable("tmpDecimal", classOf[Decimal])
     from match {
       case StringType =>
         (c, evPrim, evNull) =>
-          s"""
+          code"""
             try {
               Decimal $tmp = Decimal.apply(new java.math.BigDecimal($c.toString()));
               ${changePrecision(tmp, target, evPrim, evNull)}
@@ -934,37 +942,37 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           """
       case BooleanType =>
         (c, evPrim, evNull) =>
-          s"""
+          code"""
             Decimal $tmp = $c ? Decimal.apply(1) : Decimal.apply(0);
             ${changePrecision(tmp, target, evPrim, evNull)}
           """
       case DateType =>
         // date can't cast to decimal in Hive
-        (c, evPrim, evNull) => s"$evNull = true;"
+        (c, evPrim, evNull) => code"$evNull = true;"
       case TimestampType =>
         // Note that we lose precision here.
         (c, evPrim, evNull) =>
-          s"""
+          code"""
             Decimal $tmp = Decimal.apply(
               scala.math.BigDecimal.valueOf(${timestampToDoubleCode(c)}));
             ${changePrecision(tmp, target, evPrim, evNull)}
           """
       case DecimalType() =>
         (c, evPrim, evNull) =>
-          s"""
+          code"""
             Decimal $tmp = $c.clone();
             ${changePrecision(tmp, target, evPrim, evNull)}
           """
       case x: IntegralType =>
         (c, evPrim, evNull) =>
-          s"""
+          code"""
             Decimal $tmp = Decimal.apply((long) $c);
             ${changePrecision(tmp, target, evPrim, evNull)}
           """
       case x: FractionalType =>
         // All other numeric types can be represented precisely as Doubles
         (c, evPrim, evNull) =>
-          s"""
+          code"""
             try {
               Decimal $tmp = Decimal.apply(scala.math.BigDecimal.valueOf((double) $c));
               ${changePrecision(tmp, target, evPrim, evNull)}
@@ -979,10 +987,10 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
       from: DataType,
       ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val tz = ctx.addReferenceObj("timeZone", timeZone)
-      val longOpt = ctx.freshName("longOpt")
+      val tz = JavaCode.global(ctx.addReferenceObj("timeZone", timeZone), timeZone.getClass)
+      val longOpt = ctx.freshVariable("longOpt", classOf[Option[Long]])
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           scala.Option<Long> $longOpt =
             org.apache.spark.sql.catalyst.util.DateTimeUtils.stringToTimestamp($c, $tz);
           if ($longOpt.isDefined()) {
@@ -992,18 +1000,19 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           }
          """
     case BooleanType =>
-      (c, evPrim, evNull) => s"$evPrim = $c ? 1L : 0L;"
+      (c, evPrim, evNull) => code"$evPrim = $c ? 1L : 0L;"
     case _: IntegralType =>
-      (c, evPrim, evNull) => s"$evPrim = ${longToTimeStampCode(c)};"
+      (c, evPrim, evNull) => code"$evPrim = ${longToTimeStampCode(c)};"
     case DateType =>
-      val tz = ctx.addReferenceObj("timeZone", timeZone)
+      val tz = JavaCode.global(ctx.addReferenceObj("timeZone", timeZone), timeZone.getClass)
       (c, evPrim, evNull) =>
-        s"$evPrim = org.apache.spark.sql.catalyst.util.DateTimeUtils.daysToMillis($c, $tz) * 1000;"
+        code"""$evPrim =
+          org.apache.spark.sql.catalyst.util.DateTimeUtils.daysToMillis($c, $tz) * 1000;"""
     case DecimalType() =>
-      (c, evPrim, evNull) => s"$evPrim = ${decimalToTimestampCode(c)};"
+      (c, evPrim, evNull) => code"$evPrim = ${decimalToTimestampCode(c)};"
     case DoubleType =>
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           if (Double.isNaN($c) || Double.isInfinite($c)) {
             $evNull = true;
           } else {
@@ -1012,7 +1021,7 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
         """
     case FloatType =>
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           if (Float.isNaN($c) || Float.isInfinite($c)) {
             $evNull = true;
           } else {
@@ -1024,7 +1033,7 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
   private[this] def castToIntervalCode(from: DataType): CastFunction = from match {
     case StringType =>
       (c, evPrim, evNull) =>
-        s"""$evPrim = CalendarInterval.fromString($c.toString());
+        code"""$evPrim = CalendarInterval.fromString($c.toString());
            if(${evPrim} == null) {
              ${evNull} = true;
            }
@@ -1032,18 +1041,21 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
 
   }
 
-  private[this] def decimalToTimestampCode(d: String): String =
-    s"($d.toBigDecimal().bigDecimal().multiply(new java.math.BigDecimal(1000000L))).longValue()"
-  private[this] def longToTimeStampCode(l: String): String = s"$l * 1000000L"
-  private[this] def timestampToIntegerCode(ts: String): String =
-    s"java.lang.Math.floor((double) $ts / 1000000L)"
-  private[this] def timestampToDoubleCode(ts: String): String = s"$ts / 1000000.0"
+  private[this] def decimalToTimestampCode(d: ExprValue): Block = {
+    val block = inline"new java.math.BigDecimal(1000000L)"
+    code"($d.toBigDecimal().bigDecimal().multiply($block)).longValue()"
+  }
+  private[this] def longToTimeStampCode(l: ExprValue): Block = code"$l * 1000000L"
+  private[this] def timestampToIntegerCode(ts: ExprValue): Block =
+    code"java.lang.Math.floor((double) $ts / 1000000L)"
+  private[this] def timestampToDoubleCode(ts: ExprValue): Block =
+    code"$ts / 1000000.0"
 
   private[this] def castToBooleanCode(from: DataType): CastFunction = from match {
     case StringType =>
-      val stringUtils = StringUtils.getClass.getName.stripSuffix("$")
+      val stringUtils = inline"${StringUtils.getClass.getName.stripSuffix("$")}"
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           if ($stringUtils.isTrueString($c)) {
             $evPrim = true;
           } else if ($stringUtils.isFalseString($c)) {
@@ -1053,21 +1065,21 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           }
         """
     case TimestampType =>
-      (c, evPrim, evNull) => s"$evPrim = $c != 0;"
+      (c, evPrim, evNull) => code"$evPrim = $c != 0;"
     case DateType =>
       // Hive would return null when cast from date to boolean
-      (c, evPrim, evNull) => s"$evNull = true;"
+      (c, evPrim, evNull) => code"$evNull = true;"
     case DecimalType() =>
-      (c, evPrim, evNull) => s"$evPrim = !$c.isZero();"
+      (c, evPrim, evNull) => code"$evPrim = !$c.isZero();"
     case n: NumericType =>
-      (c, evPrim, evNull) => s"$evPrim = $c != 0;"
+      (c, evPrim, evNull) => code"$evPrim = $c != 0;"
   }
 
   private[this] def castToByteCode(from: DataType, ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val wrapper = ctx.freshName("intWrapper")
+      val wrapper = ctx.freshVariable("intWrapper", classOf[UTF8String.IntWrapper])
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           UTF8String.IntWrapper $wrapper = new UTF8String.IntWrapper();
           if ($c.toByte($wrapper)) {
             $evPrim = (byte) $wrapper.value;
@@ -1077,24 +1089,24 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           $wrapper = null;
         """
     case BooleanType =>
-      (c, evPrim, evNull) => s"$evPrim = $c ? (byte) 1 : (byte) 0;"
+      (c, evPrim, evNull) => code"$evPrim = $c ? (byte) 1 : (byte) 0;"
     case DateType =>
-      (c, evPrim, evNull) => s"$evNull = true;"
+      (c, evPrim, evNull) => code"$evNull = true;"
     case TimestampType =>
-      (c, evPrim, evNull) => s"$evPrim = (byte) ${timestampToIntegerCode(c)};"
+      (c, evPrim, evNull) => code"$evPrim = (byte) ${timestampToIntegerCode(c)};"
     case DecimalType() =>
-      (c, evPrim, evNull) => s"$evPrim = $c.toByte();"
+      (c, evPrim, evNull) => code"$evPrim = $c.toByte();"
     case x: NumericType =>
-      (c, evPrim, evNull) => s"$evPrim = (byte) $c;"
+      (c, evPrim, evNull) => code"$evPrim = (byte) $c;"
   }
 
   private[this] def castToShortCode(
       from: DataType,
       ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val wrapper = ctx.freshName("intWrapper")
+      val wrapper = ctx.freshVariable("intWrapper", classOf[UTF8String.IntWrapper])
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           UTF8String.IntWrapper $wrapper = new UTF8String.IntWrapper();
           if ($c.toShort($wrapper)) {
             $evPrim = (short) $wrapper.value;
@@ -1104,22 +1116,22 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           $wrapper = null;
         """
     case BooleanType =>
-      (c, evPrim, evNull) => s"$evPrim = $c ? (short) 1 : (short) 0;"
+      (c, evPrim, evNull) => code"$evPrim = $c ? (short) 1 : (short) 0;"
     case DateType =>
-      (c, evPrim, evNull) => s"$evNull = true;"
+      (c, evPrim, evNull) => code"$evNull = true;"
     case TimestampType =>
-      (c, evPrim, evNull) => s"$evPrim = (short) ${timestampToIntegerCode(c)};"
+      (c, evPrim, evNull) => code"$evPrim = (short) ${timestampToIntegerCode(c)};"
     case DecimalType() =>
-      (c, evPrim, evNull) => s"$evPrim = $c.toShort();"
+      (c, evPrim, evNull) => code"$evPrim = $c.toShort();"
     case x: NumericType =>
-      (c, evPrim, evNull) => s"$evPrim = (short) $c;"
+      (c, evPrim, evNull) => code"$evPrim = (short) $c;"
   }
 
   private[this] def castToIntCode(from: DataType, ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val wrapper = ctx.freshName("intWrapper")
+      val wrapper = ctx.freshVariable("intWrapper", classOf[UTF8String.IntWrapper])
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           UTF8String.IntWrapper $wrapper = new UTF8String.IntWrapper();
           if ($c.toInt($wrapper)) {
             $evPrim = $wrapper.value;
@@ -1129,23 +1141,23 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           $wrapper = null;
         """
     case BooleanType =>
-      (c, evPrim, evNull) => s"$evPrim = $c ? 1 : 0;"
+      (c, evPrim, evNull) => code"$evPrim = $c ? 1 : 0;"
     case DateType =>
-      (c, evPrim, evNull) => s"$evNull = true;"
+      (c, evPrim, evNull) => code"$evNull = true;"
     case TimestampType =>
-      (c, evPrim, evNull) => s"$evPrim = (int) ${timestampToIntegerCode(c)};"
+      (c, evPrim, evNull) => code"$evPrim = (int) ${timestampToIntegerCode(c)};"
     case DecimalType() =>
-      (c, evPrim, evNull) => s"$evPrim = $c.toInt();"
+      (c, evPrim, evNull) => code"$evPrim = $c.toInt();"
     case x: NumericType =>
-      (c, evPrim, evNull) => s"$evPrim = (int) $c;"
+      (c, evPrim, evNull) => code"$evPrim = (int) $c;"
   }
 
   private[this] def castToLongCode(from: DataType, ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val wrapper = ctx.freshName("longWrapper")
+      val wrapper = ctx.freshVariable("longWrapper", classOf[UTF8String.LongWrapper])
 
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           UTF8String.LongWrapper $wrapper = new UTF8String.LongWrapper();
           if ($c.toLong($wrapper)) {
             $evPrim = $wrapper.value;
@@ -1155,21 +1167,21 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           $wrapper = null;
         """
     case BooleanType =>
-      (c, evPrim, evNull) => s"$evPrim = $c ? 1L : 0L;"
+      (c, evPrim, evNull) => code"$evPrim = $c ? 1L : 0L;"
     case DateType =>
-      (c, evPrim, evNull) => s"$evNull = true;"
+      (c, evPrim, evNull) => code"$evNull = true;"
     case TimestampType =>
-      (c, evPrim, evNull) => s"$evPrim = (long) ${timestampToIntegerCode(c)};"
+      (c, evPrim, evNull) => code"$evPrim = (long) ${timestampToIntegerCode(c)};"
     case DecimalType() =>
-      (c, evPrim, evNull) => s"$evPrim = $c.toLong();"
+      (c, evPrim, evNull) => code"$evPrim = $c.toLong();"
     case x: NumericType =>
-      (c, evPrim, evNull) => s"$evPrim = (long) $c;"
+      (c, evPrim, evNull) => code"$evPrim = (long) $c;"
   }
 
   private[this] def castToFloatCode(from: DataType): CastFunction = from match {
     case StringType =>
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           try {
             $evPrim = Float.valueOf($c.toString());
           } catch (java.lang.NumberFormatException e) {
@@ -1177,21 +1189,21 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           }
         """
     case BooleanType =>
-      (c, evPrim, evNull) => s"$evPrim = $c ? 1.0f : 0.0f;"
+      (c, evPrim, evNull) => code"$evPrim = $c ? 1.0f : 0.0f;"
     case DateType =>
-      (c, evPrim, evNull) => s"$evNull = true;"
+      (c, evPrim, evNull) => code"$evNull = true;"
     case TimestampType =>
-      (c, evPrim, evNull) => s"$evPrim = (float) (${timestampToDoubleCode(c)});"
+      (c, evPrim, evNull) => code"$evPrim = (float) (${timestampToDoubleCode(c)});"
     case DecimalType() =>
-      (c, evPrim, evNull) => s"$evPrim = $c.toFloat();"
+      (c, evPrim, evNull) => code"$evPrim = $c.toFloat();"
     case x: NumericType =>
-      (c, evPrim, evNull) => s"$evPrim = (float) $c;"
+      (c, evPrim, evNull) => code"$evPrim = (float) $c;"
   }
 
   private[this] def castToDoubleCode(from: DataType): CastFunction = from match {
     case StringType =>
       (c, evPrim, evNull) =>
-        s"""
+        code"""
           try {
             $evPrim = Double.valueOf($c.toString());
           } catch (java.lang.NumberFormatException e) {
@@ -1199,31 +1211,32 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           }
         """
     case BooleanType =>
-      (c, evPrim, evNull) => s"$evPrim = $c ? 1.0d : 0.0d;"
+      (c, evPrim, evNull) => code"$evPrim = $c ? 1.0d : 0.0d;"
     case DateType =>
-      (c, evPrim, evNull) => s"$evNull = true;"
+      (c, evPrim, evNull) => code"$evNull = true;"
     case TimestampType =>
-      (c, evPrim, evNull) => s"$evPrim = ${timestampToDoubleCode(c)};"
+      (c, evPrim, evNull) => code"$evPrim = ${timestampToDoubleCode(c)};"
     case DecimalType() =>
-      (c, evPrim, evNull) => s"$evPrim = $c.toDouble();"
+      (c, evPrim, evNull) => code"$evPrim = $c.toDouble();"
     case x: NumericType =>
-      (c, evPrim, evNull) => s"$evPrim = (double) $c;"
+      (c, evPrim, evNull) => code"$evPrim = (double) $c;"
   }
 
   private[this] def castArrayCode(
       fromType: DataType, toType: DataType, ctx: CodegenContext): CastFunction = {
     val elementCast = nullSafeCastFunction(fromType, toType, ctx)
-    val arrayClass = classOf[GenericArrayData].getName
-    val fromElementNull = ctx.freshName("feNull")
-    val fromElementPrim = ctx.freshName("fePrim")
-    val toElementNull = ctx.freshName("teNull")
-    val toElementPrim = ctx.freshName("tePrim")
-    val size = ctx.freshName("n")
-    val j = ctx.freshName("j")
-    val values = ctx.freshName("values")
+    val arrayClass = JavaCode.javaType(classOf[GenericArrayData])
+    val fromElementNull = ctx.freshVariable("feNull", BooleanType)
+    val fromElementPrim = ctx.freshVariable("fePrim", fromType)
+    val toElementNull = ctx.freshVariable("teNull", BooleanType)
+    val toElementPrim = ctx.freshVariable("tePrim", toType)
+    val size = ctx.freshVariable("n", IntegerType)
+    val j = ctx.freshVariable("j", IntegerType)
+    val values = ctx.freshVariable("values", classOf[Array[Object]])
+    val javaType = JavaCode.javaType(fromType)
 
     (c, evPrim, evNull) =>
-      s"""
+      code"""
         final int $size = $c.numElements();
         final Object[] $values = new Object[$size];
         for (int $j = 0; $j < $size; $j ++) {
@@ -1231,7 +1244,7 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
             $values[$j] = null;
           } else {
             boolean $fromElementNull = false;
-            ${CodeGenerator.javaType(fromType)} $fromElementPrim =
+            $javaType $fromElementPrim =
               ${CodeGenerator.getValue(c, fromType, j)};
             ${castCode(ctx, fromElementPrim,
               fromElementNull, toElementPrim, toElementNull, toType, elementCast)}
@@ -1250,23 +1263,23 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
     val keysCast = castArrayCode(from.keyType, to.keyType, ctx)
     val valuesCast = castArrayCode(from.valueType, to.valueType, ctx)
 
-    val mapClass = classOf[ArrayBasedMapData].getName
+    val mapClass = JavaCode.javaType(classOf[ArrayBasedMapData])
 
-    val keys = ctx.freshName("keys")
-    val convertedKeys = ctx.freshName("convertedKeys")
-    val convertedKeysNull = ctx.freshName("convertedKeysNull")
+    val keys = ctx.freshVariable("keys", ArrayType(from.keyType))
+    val convertedKeys = ctx.freshVariable("convertedKeys", ArrayType(to.keyType))
+    val convertedKeysNull = ctx.freshVariable("convertedKeysNull", BooleanType)
 
-    val values = ctx.freshName("values")
-    val convertedValues = ctx.freshName("convertedValues")
-    val convertedValuesNull = ctx.freshName("convertedValuesNull")
+    val values = ctx.freshVariable("values", ArrayType(from.valueType))
+    val convertedValues = ctx.freshVariable("convertedValues", ArrayType(to.valueType))
+    val convertedValuesNull = ctx.freshVariable("convertedValuesNull", BooleanType)
 
     (c, evPrim, evNull) =>
-      s"""
+      code"""
         final ArrayData $keys = $c.keyArray();
         final ArrayData $values = $c.valueArray();
-        ${castCode(ctx, keys, "false",
+        ${castCode(ctx, keys, FalseLiteral,
           convertedKeys, convertedKeysNull, ArrayType(to.keyType), keysCast)}
-        ${castCode(ctx, values, "false",
+        ${castCode(ctx, values, FalseLiteral,
           convertedValues, convertedValuesNull, ArrayType(to.valueType), valuesCast)}
 
         $evPrim = new $mapClass($convertedKeys, $convertedValues);
@@ -1279,17 +1292,18 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
     val fieldsCasts = from.fields.zip(to.fields).map {
       case (fromField, toField) => nullSafeCastFunction(fromField.dataType, toField.dataType, ctx)
     }
-    val rowClass = classOf[GenericInternalRow].getName
-    val tmpResult = ctx.freshName("tmpResult")
-    val tmpInput = ctx.freshName("tmpInput")
+    val tmpResult = ctx.freshVariable("tmpResult", classOf[GenericInternalRow])
+    val rowClass = JavaCode.javaType(classOf[GenericInternalRow])
+    val tmpInput = ctx.freshVariable("tmpInput", classOf[InternalRow])
 
     val fieldsEvalCode = fieldsCasts.zipWithIndex.map { case (cast, i) =>
-      val fromFieldPrim = ctx.freshName("ffp")
-      val fromFieldNull = ctx.freshName("ffn")
-      val toFieldPrim = ctx.freshName("tfp")
-      val toFieldNull = ctx.freshName("tfn")
-      val fromType = CodeGenerator.javaType(from.fields(i).dataType)
-      s"""
+      val fromFieldPrim = ctx.freshVariable("ffp", from.fields(i).dataType)
+      val fromFieldNull = ctx.freshVariable("ffn", BooleanType)
+      val toFieldPrim = ctx.freshVariable("tfp", to.fields(i).dataType)
+      val toFieldNull = ctx.freshVariable("tfn", BooleanType)
+      val fromType = JavaCode.javaType(from.fields(i).dataType)
+      val setColumn = CodeGenerator.setColumn(tmpResult, to.fields(i).dataType, i, toFieldPrim)
+      code"""
         boolean $fromFieldNull = $tmpInput.isNullAt($i);
         if ($fromFieldNull) {
           $tmpResult.setNullAt($i);
@@ -1301,18 +1315,18 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           if ($toFieldNull) {
             $tmpResult.setNullAt($i);
           } else {
-            ${CodeGenerator.setColumn(tmpResult, to.fields(i).dataType, i, toFieldPrim)};
+            $setColumn;
           }
         }
        """
     }
     val fieldsEvalCodes = ctx.splitExpressions(
-      expressions = fieldsEvalCode,
+      expressions = fieldsEvalCode.map(_.code),
       funcName = "castStruct",
-      arguments = ("InternalRow", tmpInput) :: (rowClass, tmpResult) :: Nil)
+      arguments = ("InternalRow", tmpInput.code) :: (rowClass.code, tmpResult.code) :: Nil)
 
     (input, result, resultIsNull) =>
-      s"""
+      code"""
         final $rowClass $tmpResult = new $rowClass(${fieldsCasts.length});
         final InternalRow $tmpInput = $input;
         $fieldsEvalCodes
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala
index 838c045d5bcce..b0832d888b75d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala
@@ -579,6 +579,18 @@ class CodegenContext {
     s"${fullName}_$id"
   }
 
+  /**
+   * Creates an `ExprValue` representing a local java variable of required data type.
+   */
+  def freshVariable(name: String, dt: DataType): VariableValue =
+    JavaCode.variable(freshName(name), dt)
+
+  /**
+   * Creates an `ExprValue` representing a local java variable of required Java class.
+   */
+  def freshVariable(name: String, javaClass: Class[_]): VariableValue =
+    JavaCode.variable(freshName(name), javaClass)
+
   /**
    * Generates code for equal expression in Java.
    */
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/javaCode.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/javaCode.scala
index 2f8c853e836ba..3aa3bd872d3e7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/javaCode.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/javaCode.scala
@@ -114,6 +114,21 @@ object JavaCode {
   def isNullExpression(code: String): SimpleExprValue = {
     expression(code, BooleanType)
   }
+
+  /**
+   * Create an `Inline` for Java Class name.
+   */
+  def javaType(javaClass: Class[_]): Inline = Inline(javaClass.getName)
+
+  /**
+   * Create an `Inline` for Java Type name.
+   */
+  def javaType(dataType: DataType): Inline = Inline(CodeGenerator.javaType(dataType))
+
+  /**
+   * Create an `Inline` for boxed Java Type name.
+   */
+  def boxedType(dataType: DataType): Inline = Inline(CodeGenerator.boxedType(dataType))
 }
 
 /**
@@ -187,6 +202,16 @@ object Block {
 
   val CODE_BLOCK_BUFFER_LENGTH: Int = 512
 
+  /**
+   * A custom string interpolator which inlines a string into code block.
+   */
+  implicit class InlineHelper(val sc: StringContext) extends AnyVal {
+    def inline(args: Any*): Inline = {
+      val inlineString = sc.raw(args: _*)
+      Inline(inlineString)
+    }
+  }
+
   implicit def blocksToBlock(blocks: Seq[Block]): Block = blocks.reduceLeft(_ + _)
 
   implicit class BlockHelper(val sc: StringContext) extends AnyVal {
@@ -196,9 +221,8 @@ object Block {
         EmptyBlock
       } else {
         args.foreach {
-          case _: ExprValue =>
+          case _: ExprValue | _: Inline | _: Block =>
           case _: Int | _: Long | _: Float | _: Double | _: String =>
-          case _: Block =>
           case other => throw new IllegalArgumentException(
             s"Can not interpolate ${other.getClass.getName} into code block.")
         }
@@ -268,6 +292,14 @@ case object EmptyBlock extends Block with Serializable {
   override def children: Seq[Block] = Seq.empty
 }
 
+/**
+ * A piece of java code snippet inlines all types of input arguments into a string without
+ * tracking any reference of `JavaCode` instances.
+ */
+case class Inline(codeString: String) extends JavaCode {
+  override val code: String = codeString
+}
+
 /**
  * A typed java fragment that must be a valid java expression.
  */