[SPARK-1368][SQL] Optimized HiveTableScan

[liancheng]

JIRA issue: SPARK-1368

This PR introduces two major updates:

• Replaced FP style code with while loop and reusable GenericMutableRow object in critical path of HiveTableScan.
• Using ColumnProjectionUtils to help optimizing RCFile and ORC column pruning.

My quick micro benchmark suggests these two optimizations made the optimized version 2x and 2.5x faster when scanning CSV table and RCFile table respectively:

Original:

[info] CSV: 27676 ms, RCFile: 26415 ms
[info] CSV: 27703 ms, RCFile: 26029 ms
[info] CSV: 27511 ms, RCFile: 25962 ms

Optimized:

[info] CSV: 13820 ms, RCFile: 10402 ms
[info] CSV: 14158 ms, RCFile: 10691 ms
[info] CSV: 13606 ms, RCFile: 10346 ms

The micro benchmark loads a 609MB CVS file (structurally similar to the src test table) into a normal Hive table with LazySimpleSerDe and a RCFile table, then scans these tables respectively.

Preparation code:

package org.apache.spark.examples.sql.hive

import org.apache.spark.sql.hive.LocalHiveContext
import org.apache.spark.{SparkConf, SparkContext}

object HiveTableScanPrepare extends App {
  val sparkContext = new SparkContext(
    new SparkConf()
      .setMaster("local")
      .setAppName(getClass.getSimpleName.stripSuffix("$")))

  val hiveContext = new LocalHiveContext(sparkContext)

  import hiveContext._

  hql("drop table scan_csv")
  hql("drop table scan_rcfile")

  hql("""create table scan_csv (key int, value string)
        |  row format serde 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
        |  with serdeproperties ('field.delim'=',')
      """.stripMargin)

  hql(s"""load data local inpath "${args(0)}" into table scan_csv""")

  hql("""create table scan_rcfile (key int, value string)
        |  row format serde 'org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe'
        |stored as
        |  inputformat 'org.apache.hadoop.hive.ql.io.RCFileInputFormat'
        |  outputformat 'org.apache.hadoop.hive.ql.io.RCFileOutputFormat'
      """.stripMargin)

  hql(
    """
      |from scan_csv
      |insert overwrite table scan_rcfile
      |select scan_csv.key, scan_csv.value
    """.stripMargin)
}

Benchmark code:

package org.apache.spark.examples.sql.hive

import org.apache.spark.sql.hive.LocalHiveContext
import org.apache.spark.{SparkConf, SparkContext}

object HiveTableScanBenchmark extends App {
  val sparkContext = new SparkContext(
    new SparkConf()
      .setMaster("local")
      .setAppName(getClass.getSimpleName.stripSuffix("$")))

  val hiveContext = new LocalHiveContext(sparkContext)

  import hiveContext._

  val scanCsv = hql("select key from scan_csv")
  val scanRcfile = hql("select key from scan_rcfile")

  val csvDuration = benchmark(scanCsv.count())
  val rcfileDuration = benchmark(scanRcfile.count())

  println(s"CSV: $csvDuration ms, RCFile: $rcfileDuration ms")

  def benchmark(f: => Unit) = {
    val begin = System.currentTimeMillis()
    f
    val end = System.currentTimeMillis()
    end - begin
  }
}

@marmbrus Please help review, thanks!

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[marmbrus]

I think this is allocating a new array every time.

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14941/

[marmbrus]

I'm curious if there is a cost to pattern matching here instead of using _1 and _2?

[liancheng]

Yes there is a Product2[A, B].unapply function call cost. Removing this gains 0.3% speed up.

[marmbrus]

Nice speed up! :)

I looked at the test failure. Looks like this TODO is finally coming back to bite us. Instead of looking for any Sort we should walk the tree until we find either a Sort or an operation that doesn't preserve ordering (join , aggregate, etc).

Once we fix that I'd propose merging this right away and then addressing the other possible suggestions in a followup PR.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished. All automated tests passed.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15017/

[liancheng]

@marmbrus I worked around the test failure by adding a SortedOperation pattern that conservatively matches some definitely sorted operations (false negative rather than false positive). This may slow down the test suite a bit. Since most test output are empty or very small, this shouldn't be an issue right now.

Two new optimizations applied:

• Using mutable pairs
• Avoiding pattern matching function calls (Array.unapplySeq)

New micro benchmark data:

Original:

[info] CSV: 27676 ms, RCFile: 26415 ms
[info] CSV: 27703 ms, RCFile: 26029 ms
[info] CSV: 27511 ms, RCFile: 25962 ms

Optimized:

[info] CSV: 12357 ms, RCFile: 9283 ms
[info] CSV: 12291 ms, RCFile: 9298 ms
[info] CSV: 12325 ms, RCFile: 9242 ms

As for Hive data unwrapping, I couldn't find a "static" method to eliminate right now. Any hints?

[marmbrus]

@marmbrus I worked around the test failure by adding a SortedOperation pattern that conservatively matches some definitely sorted operations (false negative rather than false positive). This may slow down the test suite a bit. Since most test output are empty or very small, this shouldn't be an issue right now.

I think false negatives are the wrong direction to go here. A false negative means that we think the query is not ordered when it should be and thus are disregarding the order when we should in fact be checking it.

Maybe it would be better to recursively walk the tree looking explicitly for nodes that do not preserve order (aggregation, join, base relations) and then return false. Sorts would return true. Thoughts?

New micro benchmark data:

Sweet, looks like we shaved off a little bit more, so these optimizations were worth it! It would be good to make notes on which changes lead to what kind of speed up here. That way, we can better focus our efforts when we optimize in the future.

As for Hive data unwrapping, I couldn't find a "static" method to eliminate right now. Any hints?

My thought was that you will create an Array of Any => Any functions that can be applied to each column. This way you only match on the datatype once, at the beginning, and then simply index into this array instead of matching for each data item.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished. All automated tests passed.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15069/

[liancheng]

@marmbrus Updated HiveComparisonTest and removed SortedOperation, how about this version?

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished. All automated tests passed.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15251/

[marmbrus]

First merge as a committer :)

Thanks for doing this!