insert assertion

add test case

Author: kiszk

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

@@ -217,6 +217,18 @@ class CodegenContext {
     splitExpressions(expressions = initCodes, funcName = "init", arguments = Nil)
   }
 
+  /**
+   * Return true if a given variable has been described as a global variable
+   */
+  def isDeclaredMutableState(varName: String): Boolean = {
+    val j = varName.indexOf("[")
+    val qualifiedName = if (j < 0) varName else varName.substring(0, j)
+    mutableStates.find { s =>
+      val i = s._2.indexOf("[")
+      qualifiedName == (if (i < 0) s._2 else s._2.substring(0, i))
+    }.isDefined
+  }
+
   /**
    * Code statements to initialize states that depend on the partition index.
    * An integer `partitionIndex` will be made available within the scope.
@@ -842,7 +854,10 @@ class CodegenContext {
       blocks.head
     } else {
       val func = freshName(funcName)
-      val argString = arguments.map { case (t, name) => s"$t $name" }.mkString(", ")
+      val argString = arguments.map { case (t, name) =>
+        assert(!isDeclaredMutableState(name),
+          s"$name in arguments should not be declared as a global variable")
+        s"$t $name" }.mkString(", ")
       val functions = blocks.zipWithIndex.map { case (body, i) =>
         val name = s"${func}_$i"
         val code = s"""

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

@@ -394,4 +394,23 @@ class CodeGenerationSuite extends SparkFunSuite with ExpressionEvalHelper {
       Map("add" -> Literal(1))).genCode(ctx)
     assert(ctx.mutableStates.isEmpty)
   }
+
+  test("SPARK-22668: ensure no global variables in split method arguments") {
+    val ctx = new CodegenContext
+    ctx.addMutableState(ctx.JAVA_INT, "ij")
+    ctx.addMutableState("int[]", "array")
+    ctx.addMutableState("int[][]", "b")
+
+    assert(ctx.isDeclaredMutableState("ij"))
+    assert(ctx.isDeclaredMutableState("array"))
+    assert(ctx.isDeclaredMutableState("array[1]"))
+    assert(ctx.isDeclaredMutableState("b[]"))
+    assert(ctx.isDeclaredMutableState("b[1][]"))
+
+    assert(!ctx.isDeclaredMutableState("i"))
+    assert(!ctx.isDeclaredMutableState("j"))
+    assert(!ctx.isDeclaredMutableState("ij1"))
+    assert(!ctx.isDeclaredMutableState("arr"))
+    assert(!ctx.isDeclaredMutableState("bb[]"))
+  }
 }