Add new conf for maxParamNumInJavaMethod

Author: maropu

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -607,6 +607,17 @@ object SQLConf {
     .intConf
     .createWithDefault(100)
 
+  val MAX_PARAM_NUM_IN_JAVA_METHOD =
+     buildConf("spark.sql.codegen.maxParamNumInJavaMethod")
+      .internal()
+      .doc("The maximum number of parameters in codegened Java functions. When a function " +
+         "exceeds this threshold, the code generator gives up splitting the function code. " +
+         "This default value is 127 because the maximum length of parameters in non-static Java " +
+         "methods is 254 and a parameter of type long or double contributes " +
+         "two units to the length.")
+      .intConf
+      .createWithDefault(127)
+
   val CODEGEN_FALLBACK = buildConf("spark.sql.codegen.fallback")
     .internal()
     .doc("When true, (whole stage) codegen could be temporary disabled for the part of query that" +
@@ -1156,6 +1167,8 @@ class SQLConf extends Serializable with Logging {
 
   def wholeStageMaxNumFields: Int = getConf(WHOLESTAGE_MAX_NUM_FIELDS)
 
+  def maxParamNumInJavaMethod: Int = getConf(MAX_PARAM_NUM_IN_JAVA_METHOD)
+
   def codegenFallback: Boolean = getConf(CODEGEN_FALLBACK)
 
   def loggingMaxLinesForCodegen: Int = getConf(CODEGEN_LOGGING_MAX_LINES)

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

@@ -309,18 +309,11 @@ case class HashAggregateExec(
       subExprs: Map[Expression, SubExprEliminationState],
       otherArgs: Seq[(String, String)] = Seq.empty): Seq[String] = {
     aggExprs.zipWithIndex.map { case (aggExpr, i) =>
-      // The maximum length of parameters in non-static Java methods is 254, but a parameter of
-      // type long or double contributes two units to the length. So, this method gives up
-      // splitting the code if the parameter length goes over 127.
       val args = (getInputVariableReferences(ctx, aggExpr, subExprs) ++ otherArgs).toSeq
 
-      // This is for testing/benchmarking only
-      val maxParamNumInJavaMethod =
-          sqlContext.getConf("spark.sql.codegen.aggregate.maxParamNumInJavaMethod", null) match {
-        case null | "" => 127
-        case param => param.toInt
-      }
-      if (args.size <= maxParamNumInJavaMethod) {
+      // This method gives up splitting the code if the parameter length goes over
+      // `maxParamNumInJavaMethod`.
+      if (args.size <= sqlContext.conf.maxParamNumInJavaMethod) {
         val doAggVal = ctx.freshName(s"doAggregateVal_${aggExpr.prettyName}")
         val argList = args.map(a => s"${a._1} ${a._2}").mkString(", ")
         val doAggValFuncName = ctx.addNewFunction(doAggVal,

sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala

@@ -238,7 +238,7 @@ class WholeStageCodegenSuite extends QueryTest with SharedSQLContext {
   }
 
   test("SPARK-21870 check the case where the number of parameters goes over the limit") {
-    withSQLConf("spark.sql.codegen.aggregate.maxParamNumInJavaMethod" -> "2") {
+    withSQLConf("spark.sql.codegen.maxParamNumInJavaMethod" -> "2") {
       sql("CREATE OR REPLACE TEMPORARY VIEW t AS SELECT * FROM VALUES (1, 1, 1) AS t(a, b, c)")
       val df = sql("SELECT SUM(a + b + c) AS sum FROM t")
       assert(df.collect === Seq(Row(3)))