address review comments

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

@@ -799,13 +799,10 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
 
   private[this] def castToByteCode(from: DataType, ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val wrapper = "intWrapper"
-      if (!ctx.mutableStates.exists(s => s._1 == wrapper)) {
-        ctx.addMutableState("UTF8String.IntWrapper", wrapper,
-          s"$wrapper = new UTF8String.IntWrapper();")
-      }
+      val wrapper = ctx.freshName("intWrapper")
       (c, evPrim, evNull) =>
         s"""
+          UTF8String.IntWrapper $wrapper = new UTF8String.IntWrapper();
           if ($c.toByte($wrapper)) {
             $evPrim = (byte) $wrapper.value;
           } else {
@@ -828,13 +825,10 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
       from: DataType,
       ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val wrapper = "intWrapper"
-      if (!ctx.mutableStates.exists(s => s._1 == wrapper)) {
-        ctx.addMutableState("UTF8String.IntWrapper", wrapper,
-          s"$wrapper = new UTF8String.IntWrapper();")
-      }
+      val wrapper = ctx.freshName("intWrapper")
       (c, evPrim, evNull) =>
         s"""
+          UTF8String.IntWrapper $wrapper = new UTF8String.IntWrapper();
           if ($c.toShort($wrapper)) {
             $evPrim = (short) $wrapper.value;
           } else {
@@ -855,13 +849,10 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
 
   private[this] def castToIntCode(from: DataType, ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val wrapper = "intWrapper"
-      if (!ctx.mutableStates.exists(s => s._1 == wrapper)) {
-        ctx.addMutableState("UTF8String.IntWrapper", wrapper,
-          s"$wrapper = new UTF8String.IntWrapper();")
-      }
+      val wrapper = ctx.freshName("intWrapper")
       (c, evPrim, evNull) =>
         s"""
+          UTF8String.IntWrapper $wrapper = new UTF8String.IntWrapper();
           if ($c.toInt($wrapper)) {
             $evPrim = $wrapper.value;
           } else {
@@ -882,14 +873,11 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
 
   private[this] def castToLongCode(from: DataType, ctx: CodegenContext): CastFunction = from match {
     case StringType =>
-      val wrapper = "longWrapper"
-      if (!ctx.mutableStates.exists(s => s._1 == wrapper)) {
-        ctx.addMutableState("UTF8String.LongWrapper", wrapper,
-          s"$wrapper = new UTF8String.LongWrapper();")
-      }
+      val wrapper = ctx.freshName("longWrapper")
 
       (c, evPrim, evNull) =>
         s"""
+          UTF8String.LongWrapper $wrapper = new UTF8String.LongWrapper();
           if ($c.toLong($wrapper)) {
             $evPrim = $wrapper.value;
           } else {

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

@@ -184,6 +184,23 @@ class CodegenContext {
     mutableStates += ((javaType, variableName, initCode))
   }
 
+  /**
+   * Add a mutable state as a field to the generated class if the name has not been added
+   *
+   * @param javaType Java type of the field.
+   * @param variableName Name of the field.
+   * @param initCode The statement(s) to put into the init() method to initialize this field.
+   *                 If left blank, the field will be default-initialized.
+   */
+  def reuseOrAddMutableState(
+      javaType: String,
+      variableName: String,
+      initCode: String = ""): Unit = {
+    if (!mutableStates.exists(s => s._1 == variableName)) {
+      addMutableState(javaType, variableName, initCode)
+    }
+  }
+
   /**
    * Add buffer variable which stores data coming from an [[InternalRow]]. This methods guarantees
    * that the variable is safely stored, which is important for (potentially) byte array backed

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

@@ -93,25 +93,19 @@ private [sql] object GenArrayData {
     if (!ctx.isPrimitiveType(elementType)) {
       val arrayName = "arrayObject"
       val genericArrayClass = classOf[GenericArrayData].getName
-      if (!ctx.mutableStates.exists(s => s._1 == arrayName)) {
-        ctx.addMutableState("Object[]", arrayName)
-      }
+      ctx.reuseOrAddMutableState("Object[]", arrayName)
 
       val assignments = elementsCode.zipWithIndex.map { case (eval, i) =>
-        val isNullAssignment = if (eval.isNull == "false") {
-          ""
+        val isNullAssignment = if (!isMapKey) {
+          s"$arrayName[$i] = null;"
         } else {
-          if (!isMapKey) {
-            s"$arrayName[$i] = null;"
-          } else {
-            "throw new RuntimeException(\"Cannot use null as map key!\");"
-          }
+          "throw new RuntimeException(\"Cannot use null as map key!\");"
         }
         eval.code + s"""
-         if (!${eval.isNull}) {
-           $arrayName[$i] = ${eval.value};
-         } else {
+         if (${eval.isNull}) {
            $isNullAssignment
+         } else {
+           $arrayName[$i] = ${eval.value};
          }
        """
       }

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

@@ -334,10 +334,8 @@ case class RegExpReplace(subject: Expression, regexp: Expression, rep: Expressio
     ctx.addMutableState("String", termLastReplacement, s"${termLastReplacement} = null;")
     ctx.addMutableState("UTF8String",
       termLastReplacementInUTF8, s"${termLastReplacementInUTF8} = null;")
-    if (!ctx.mutableStates.exists(s => s._1 == termResult)) {
-      ctx.addMutableState(classNameStringBuffer,
-        termResult, s"${termResult} = new $classNameStringBuffer();")
-    }
+    ctx.reuseOrAddMutableState(classNameStringBuffer,
+      termResult, s"${termResult} = new $classNameStringBuffer();")
 
     val setEvNotNull = if (nullable) {
       s"${ev.isNull} = false;"