[SPARK-11243] [SQL] output UnsafeRow from columnar cache

[davies]

This PR change InMemoryTableScan to output UnsafeRow, and optimize the unrolling and scanning by coping the bytes for var-length types between UnsafeRow and ByteBuffer directly without creating the wrapper objects. When scanning the decimals in TPC-DS store_sales table, it's 80% faster (copy it as long without create Decimal objects).

[SparkQA]

Test build #44083 has finished for PR 9203 at commit 15ebb3b.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * abstract class ColumnarIterator extends Iterator[InternalRow]\n * class MutableUnsafeRow(val writer: UnsafeRowWriter) extends GenericMutableRow(null)\n * class SpecificColumnarIterator extends $\n

[rxin]

as discussed offline, maybe change this to assert?

[davies]

will do in separate PR

[JoshRosen]

I'm really excited to see how much of a performance difference this makes when scanning string columns, since this could potentially provide a big perf. boost to the sc.textFile replacement that I showed the other day.

[rxin]

Do we ever start from the middle of an existing buffer? If not, can we move startingOffset into holder class?

[davies]

Yes, in UnsafeProjection, there could be multiple UnsafeRowWriter share a single BufferHolder.

[rxin]

LGTM otherwise.

[SparkQA]

Test build #44089 has finished for PR 9203 at commit 81e3fad.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class MutableUnsafeRow(val writer: UnsafeRowWriter) extends GenericMutableRow(null)\n

[davies]

@JoshRosen If we only scan the cached string and access it (without adding a ConvertToUnsafe), this could be a little bit slower, because we copy the bytes into UnsafeRow now. For the UTF8String object, we still will create it anyway, before this patch, we create it in ColumnAccessor, after this patch, we create in UnsafeRow.getUTF8String() (we could re-use that actually).

[SparkQA]

Test build #44092 has finished for PR 9203 at commit e33170d.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class MutableUnsafeRow(val writer: UnsafeRowWriter) extends GenericMutableRow(null)\n

[cloud-fan]

Why implement these primitive write method in writer class not in codegen?

[davies]

Those in UnsafeRowWriter are needed for ColumnAccessor. The purpose of these *Writer is to simplify generated code (which is already hard to understand), these method also go in this direction.

In terms of performance, I think there will not difference, they all could be JITed and inlined.

[rxin]

I think it is simpler to put them there rather than in generated code.

[SparkQA]

Test build #44098 has finished for PR 9203 at commit 1421bce.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class MutableUnsafeRow(val writer: UnsafeRowWriter) extends GenericMutableRow(null)\n

[SparkQA]

Test build #44100 has finished for PR 9203 at commit cab9286.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class MutableUnsafeRow(val writer: UnsafeRowWriter) extends GenericMutableRow(null)\n

[rxin]

I'm going to merge this first. @davies please take a look at @cloud-fan 's comments, and address them in follow-up prs if necessary.