[SPARK-11149] [SQL] Improve cache performance for primitive types

[davies]

This PR improve the performance by:

1. Generate an Iterator that take Iterator[CachedBatch] as input, and call accessors (unroll the loop for columns), avoid the expensive Iterator.flatMap.

2. Use Unsafe.getInt/getLong/getFloat/getDouble instead of ByteBuffer.getInt/getLong/getFloat/getDouble, the later one actually read byte by byte.

3. Remove the unnecessary copy() in Coalesce(), which is not related to memory cache, found during benchmark.

The following benchmark showed that we can speedup the columnar cache of int by 2x.

path = '/opt/tpcds/store_sales/'
int_cols = ['ss_sold_date_sk', 'ss_sold_time_sk', 'ss_item_sk','ss_customer_sk']
df = sqlContext.read.parquet(path).select(int_cols).cache()
df.count()

t = time.time()
print df.select("*")._jdf.queryExecution().toRdd().count()
print time.time() - t

[SparkQA]

Test build #43824 has finished for PR 9145 at commit 7ee54a9.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class ColumnarIterator extends Iterator[InternalRow]
  • class SpecificColumnarIterator extends $

[SparkQA]

Test build #43834 has finished for PR 9145 at commit 8a49887.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class ColumnarIterator extends Iterator[InternalRow]
  • class SpecificColumnarIterator extends $

[SparkQA]

Test build #43836 has finished for PR 9145 at commit 1ef3e18.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class ColumnarIterator extends Iterator[InternalRow]
  • class SpecificColumnarIterator extends $

[rxin]

as discussed offline, we can add /* ... */ to still enable pasting this into an IDE.

[rxin]

creators

[rxin]

actually probably more clear to say

initializeAccessors and extractors

[SparkQA]

Test build #43844 has finished for PR 9145 at commit 4511781.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class ColumnarIterator extends Iterator[InternalRow]
  • class SpecificColumnarIterator extends $

[rxin]

LGTM - although I didn't look super closely so might be good for an extra pair of eyes too.

[SparkQA]

Test build #43936 has finished for PR 9145 at commit f9151cc.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * abstract class ColumnarIterator extends Iterator[InternalRow]\n * class SpecificColumnarIterator extends $\n

[davies]

I'm going to merge this first, unblock me to work on output UnsafeRow for columnar cache, because having Unsafe format inside a MutableRow could result unexpected behavior, any new comments will be address in follow up PR.

[tedyu]

Around line 332, there is call to buffer.getShort()

Is it worth adding corresponding method to ByteBufferHelper ?

If so, I can send a PR.

Thanks

[davies]

I think it does not worth it.

[cloud-fan]

LGTM