[SPARK-16610][SQL] Add orc.compress as an alias for compression option.

sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala

@@ -67,10 +67,10 @@ private[sql] class OrcFileFormat
       job: Job,
       options: Map[String, String],
       dataSchema: StructType): OutputWriterFactory = {
-    val orcOptions = new OrcOptions(options)
-
     val configuration = job.getConfiguration
 
+    val orcOptions = new OrcOptions(options, configuration)
+
     configuration.set(OrcRelation.ORC_COMPRESSION, orcOptions.compressionCodec)
     configuration match {
       case conf: JobConf =>

sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcOptions.scala

@@ -17,27 +17,40 @@
 
 package org.apache.spark.sql.hive.orc
 
+import org.apache.hadoop.conf.Configuration
+
 /**
  * Options for the ORC data source.
  */
 private[orc] class OrcOptions(
-    @transient private val parameters: Map[String, String])
+    @transient private val parameters: Map[String, String],
+    @transient private val conf: Configuration)
   extends Serializable {
 
   import OrcOptions._
 
   /**
-   * Compression codec to use. By default snappy compression.
+   * Compression codec to use. By default use the value specified in Hadoop configuration.
+   * If `orc.compress` is unset, then we use snappy.
    * Acceptable values are defined in [[shortOrcCompressionCodecNames]].
    */
   val compressionCodec: String = {
-    val codecName = parameters.getOrElse("compression", "snappy").toLowerCase
-    if (!shortOrcCompressionCodecNames.contains(codecName)) {
-      val availableCodecs = shortOrcCompressionCodecNames.keys.map(_.toLowerCase)
-      throw new IllegalArgumentException(s"Codec [$codecName] " +
-        s"is not available. Available codecs are ${availableCodecs.mkString(", ")}.")
+    val default = conf.get(OrcRelation.ORC_COMPRESSION, "SNAPPY")
+
+    // Because the ORC configuration value in `default` is not guaranteed to be the same
+    // with keys in `shortOrcCompressionCodecNames` in Spark, this value should not be
+    // used as the key for `shortOrcCompressionCodecNames` but just a return value.
+    parameters.get("compression") match {
+      case None => default
+      case Some(name) =>
+        val lowerCaseName = name.toLowerCase
+        if (!shortOrcCompressionCodecNames.contains(lowerCaseName)) {
+          val availableCodecs = shortOrcCompressionCodecNames.keys.map(_.toLowerCase)
+          throw new IllegalArgumentException(s"Codec [$lowerCaseName] " +
+            s"is not available. Available codecs are ${availableCodecs.mkString(", ")}.")
+        }
+        shortOrcCompressionCodecNames(lowerCaseName)
     }
-    shortOrcCompressionCodecNames(codecName)
   }
 }
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala

@@ -161,6 +161,29 @@ class OrcQuerySuite extends QueryTest with BeforeAndAfterAll with OrcTest {
     }
   }
 
+  test("SPARK-16610: Respect orc.compress configuration when compression is unset") {
+    // Respect `orc.compress`.
+    withTempPath { file =>
+      spark.range(0, 10).write
+        .option("orc.compress", "ZLIB")
+        .orc(file.getCanonicalPath)
+      val expectedCompressionKind =
+        OrcFileOperator.getFileReader(file.getCanonicalPath).get.getCompression
+      assert("ZLIB" === expectedCompressionKind.name())
+    }
+
+    // `compression` overrides `orc.compress`.
+    withTempPath { file =>
+      spark.range(0, 10).write
+        .option("compression", "ZLIB")
+        .option("orc.compress", "SNAPPY")
+        .orc(file.getCanonicalPath)
+      val expectedCompressionKind =
+        OrcFileOperator.getFileReader(file.getCanonicalPath).get.getCompression
+      assert("ZLIB" === expectedCompressionKind.name())
+    }
+  }
+
   // Hive supports zlib, snappy and none for Hive 1.2.1.
   test("Compression options for writing to an ORC file (SNAPPY, ZLIB and NONE)") {
     withTempPath { file =>