[SPARK-3197] [SQL] Reduce the Expression tree object creations for aggregation function (min/max)

[chenghao-intel]

Aggregation function min/max in catalyst will create expression tree for each single row, however, the expression tree creation is quite expensive in a multithreading env currently. Hence we got a very bad performance for the min/max.
Here is the benchmark that I've done in my local.

Master	Previous Result (ms)	Current Result (ms)
local	3645	3416
local[6]	3602	1002

The Benchmark source code.

case class Record(key: Int, value: Int)

object TestHive2 extends HiveContext(new SparkContext("local[6]", "TestSQLContext", new SparkConf()))

object DataPrepare extends App {
  import TestHive2._

  val rdd = sparkContext.parallelize((1 to 10000000).map(i => Record(i % 3000, i)), 12)

  runSqlHive("SHOW TABLES")
  runSqlHive("DROP TABLE if exists a")
  runSqlHive("DROP TABLE if exists result")
  rdd.registerAsTable("records")

  runSqlHive("""CREATE TABLE a (key INT, value INT)
                 | ROW FORMAT SERDE 
                 | 'org.apache.hadoop.hive.serde2.columnar.LazyBinaryColumnarSerDe' 
                 | STORED AS RCFILE
               """.stripMargin)
  runSqlHive("""CREATE TABLE result (key INT, value INT)
                 | ROW FORMAT SERDE 
                 | 'org.apache.hadoop.hive.serde2.columnar.LazyBinaryColumnarSerDe' 
                 | STORED AS RCFILE
               """.stripMargin)

  hql(s"""from records 
             | insert into table a
             | select key, value
           """.stripMargin)
}

object PerformanceTest extends App {
  import TestHive2._

  hql("SHOW TABLES")
  hql("set spark.sql.shuffle.partitions=12")

  val cmd = "select min(value), max(value) from a group by key"

  val results = ("Result1", benchmark(cmd)) :: 
                ("Result2", benchmark(cmd)) :: 
                ("Result3", benchmark(cmd)) :: Nil
  results.foreach { case (prompt, result) => {
      println(s"$prompt: took ${result._1} ms (${result._2} records)")
    }
  }

  def benchmark(cmd: String) = {
    val begin = System.currentTimeMillis()
    val count = hql(cmd).count
    val end = System.currentTimeMillis()
    ((end - begin), count)
  }
}

[SparkQA]

QA tests have started for PR 2113 at commit 03c6d4f.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2113 at commit 03c6d4f.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class MutableLiteral(var value: Any, dataType: DataType, nullable: Boolean = true)

[marmbrus]

Nit: I don't think this needs to be @transient (this function will never be serialized), and it could also be a val.

[chenghao-intel]

Oh, yes, you're right, I will update this.

[chenghao-intel]

test this please

[chenghao-intel]

Jenkins, retest this please.

[SparkQA]

QA tests have started for PR 2113 at commit db40395.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2113 at commit db40395.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class MutableLiteral(var value: Any, dataType: DataType, nullable: Boolean = true)

[marmbrus]

Thanks! I've merged this to master and 1.1.

[yjshen]

@chenghao-intel , will this cause twice evaluation of expr on the same input line? why not eval once and store in a tmp val? Did I miss something?

[chenghao-intel]

Yes, you're right! I think now it can be replaced by MinOf.

[yjshen]

thanks, just start to read aggregate code :)

[cloud-fan]

@yjshen , the new aggregation has been merged, you have a lot more to read now :)

[yjshen]

yes, it would be a long journey, but I could ask the author whenever I have a question ;)