[SPARK-14116][SQL] Implements buildReader() for ORC data source

[liancheng]

What changes were proposed in this pull request?

This PR implements FileFormat.buildReader() for our ORC data source. It also fixed several minor styling issues related to HadoopFsRelation planning code path.

Note that OrcNewInputFormat doesn't rely on OrcNewSplit for creating OrcRecordReaders, plain FileSplit is just fine. That's why we can simply create the record reader with the help of OrcNewInputFormat and FileSplit.

How was this patch tested?

Existing test cases should do the work

[SparkQA]

Test build #54048 has finished for PR 11936 at commit 057b6f2.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class FilePartition(index: Int, files: Seq[PartitionedFile]) extends Partition

[SparkQA]

Test build #54169 has finished for PR 11936 at commit b1d630b.

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

[liancheng]

cc @cloud-fan @yhuai @marmbrus

[cloud-fan]

seems this variable is never used?

[liancheng]

Yea... Didn't remove it since I haven't gone through that part thoroughly.

[cloud-fan]

LGTM

[SparkQA]

Test build #54172 has finished for PR 11936 at commit 7d628ed.

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

[liancheng]

@marmbrus Unfortunately our current strategy for generating Spark partitions doesn't play well with ORC version of buildReader(), or to be more specific, OrcRecordReader. The following Spark shell snippet shows the problem:

import org.apache.spark.sql.types._

// Just creates a random file that is large enough
val df = sqlContext.range(1000000).select(
  'id cast StringType as 'a,
  'id cast StringType as 'b,
  'id cast StringType as 'c
)

val path = "/tmp/large.orc"
df.write.mode("overwrite").orc(path)

sqlContext.sql(s"SET spark.sql.files.maxPartitionBytes=${1024 * 1024}")
sqlContext.sql(s"SET spark.sql.parquet.enableVectorizedReader=false")

sqlContext.read.orc(path).count() // <-- Gives 500,000 instead of 1,000,000

Please refer to discussion here for details.

[SparkQA]

Test build #54182 has finished for PR 11936 at commit 2cbb56d.

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

[liancheng]

Just realized the problem while washing dishes...

The above comment is a false alarm. The real problem is that I ignored start position and length of the PartitionedFile and created ORC splits using OrcNewInputFormat. After digging for a while, I found that although instantiating OrcNewSplit is pretty complicated, OrcNewInputFormat doesn't really require an OrcNewSplit, a plain FileSplit is just enough. So now the OrcRecordReader is simply created using OrcNewInputFormat and a FileSplit, which works well.

[cloud-fan]

should we move this out of the function? Then we don't need to create them everytime this function is used.

[liancheng]

Yea, we should. Michael also mentioned this in this thread. We can do it in a follow-up PR after finishing other data sources to avoid conflicts.

[SparkQA]

Test build #54183 has finished for PR 11936 at commit a26973e.

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

[yhuai]

I feel that using a fold at here make the code harder to understand.

[yhuai]

How about we use a more straightforward way.

[liancheng]

Done.

[SparkQA]

Test build #54254 has finished for PR 11936 at commit 25d894f.

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

[yhuai]

Let's rename this conf.

[yhuai]

Thanks. I am merging this to master.

@liancheng @cloud-fan Let's address https://github.com/apache/spark/pull/11936/files#r57520723 in either of your PR for other formats.