[SPARK-13255][SQL] Update vectorized reader to directly return ColumnarBatch instead of InternalRows.

[nongli]

What changes were proposed in this pull request?

(Please fill in changes proposed in this fix)

Currently, the parquet reader returns rows one by one which is bad for performance. This patch
updates the reader to directly return ColumnarBatches. This is only enabled with whole stage
codegen, which is the only operator currently that is able to consume ColumnarBatches (instead
of rows). The current implementation is a bit of a hack to get this to work and we should do
more refactoring of these low level interfaces to make this work better.

How was this patch tested?

Results:
TPCDS:                             Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)
---------------------------------------------------------------------------------
q55 (before)                             8897 / 9265         12.9          77.2
q55                                      5486 / 5753         21.0          47.6

[SparkQA]

Test build #52198 has finished for PR 11435 at commit c1b6ab1.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public class UnsafeRowParquetRecordReader extends SpecificParquetRecordReaderBase<Object>

[SparkQA]

Test build #52199 has finished for PR 11435 at commit 490189c.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #2592 has finished for PR 11435 at commit 490189c.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #52344 has finished for PR 11435 at commit c5a041e.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[nongli]

This patch is abusing the existing abstractions which will be/is being cleaned up in other patches. For a benchmark of the partition reading speed, I tested this on a table with 2 int cols, one in the data, one as the partition column. The results are:

Partitioned Table:                  Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative
-------------------------------------------------------------------------------------------
Read data column                          217 /  262         72.3          13.8       1.0X
Read partition column                     104 /  150        150.6           6.6       2.1X
Read both columns                         240 /  266         65.5          15.3       0.9X

[SparkQA]

Test build #52345 has finished for PR 11435 at commit ce4099c.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #52350 has finished for PR 11435 at commit 2908d01.

• This patch fails Scala style tests.
• This patch does not merge cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #52351 has finished for PR 11435 at commit 670dee7.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[davies]

If row is UnsafeRow, v.getBytes() will copy the bytes, we should pull that out of the loop.

[SparkQA]

Test build #2607 has finished for PR 11435 at commit 670dee7.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #52410 has finished for PR 11435 at commit 2a16ec0.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #2611 has finished for PR 11435 at commit 2a16ec0.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #2612 has finished for PR 11435 at commit 2a16ec0.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #52446 has finished for PR 11435 at commit f5f1e2b.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #52450 has finished for PR 11435 at commit ed79eee.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[davies]

How about this:

if ($batch != null) {
  $scanBatches();
} else if ($input.hasNext()) {
  Object $value = $input.next();
  if ($value instanceof $columnarBatchClz) {
       $batch = ($columnarBatchClz)$value;
       $scanBatches();
  } else {
       $scanRows((InternalRow) $value);
  }
}

[SparkQA]

Test build #52479 has finished for PR 11435 at commit 48102e3.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[davies]

LGTM, merging this into master, thanks!