initial commit for regexpreplace

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

@@ -322,7 +322,7 @@ case class RegExpReplace(subject: Expression, regexp: Expression, rep: Expressio
     val termLastReplacement = ctx.freshName("lastReplacement")
     val termLastReplacementInUTF8 = ctx.freshName("lastReplacementInUTF8")
 
-    val termResult = ctx.freshName("result")
+    val termResult = "termResult"
 
     val classNamePattern = classOf[Pattern].getCanonicalName
     val classNameStringBuffer = classOf[java.lang.StringBuffer].getCanonicalName
@@ -334,8 +334,10 @@ case class RegExpReplace(subject: Expression, regexp: Expression, rep: Expressio
     ctx.addMutableState("String", termLastReplacement, s"${termLastReplacement} = null;")
     ctx.addMutableState("UTF8String",
       termLastReplacementInUTF8, s"${termLastReplacementInUTF8} = null;")
-    ctx.addMutableState(classNameStringBuffer,
-      termResult, s"${termResult} = new $classNameStringBuffer();")
+    if (!ctx.mutableStates.exists(s => s._1 == termResult)) {
+      ctx.addMutableState(classNameStringBuffer,
+        termResult, s"${termResult} = new $classNameStringBuffer();")
+    }
 
     val setEvNotNull = if (nullable) {
       s"${ev.isNull} = false;"

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala

@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.expressions
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodeAndComment, CodeGenerator, CodegenContext}
 import org.apache.spark.sql.types.{IntegerType, StringType}
 
 /**
@@ -178,6 +179,29 @@ class RegexpExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(nonNullExpr, "num-num", row1)
   }
 
+  test("SPARK-22570: should not create a lot of instance variables") {
+    val expr = RegExpReplace(Literal("100"), Literal("(\\d+)"), Literal("num"))
+    val ctx = new CodegenContext
+    val codes = (1 to 16000).map(_ => expr.genCode(ctx).code)
+    val eval = ctx.splitExpressions(ctx.INPUT_ROW, codes)
+    val codeBody = s"""
+      public RegexpExpressionsTest generate(Object[] references) {
+        return new RegexpExpressionsTest(references);
+      }
+      class RegexpExpressionsTest {
+        Object[] references;
+        ${ctx.declareMutableStates()}
+        public RegexpExpressionsTest(Object[] references) {
+          ${ctx.initMutableStates()}
+        }
+        public void apply(InternalRow ${ctx.INPUT_ROW}) {
+          ${eval}
+        }
+       ${ctx.declareAddedFunctions()}
+     }"""
+    CodeGenerator.compile(new CodeAndComment(codeBody, ctx.getPlaceHolderToComments()))
+  }
+
   test("RegexExtract") {
     val row1 = create_row("100-200", "(\\d+)-(\\d+)", 1)
     val row2 = create_row("100-200", "(\\d+)-(\\d+)", 2)