[SPARK-8990] [SQL] SPARK-8990 DataFrameReader.parquet() should respec…

…t user specified options (for branch-1.4) Backports PR apache#7347 (SPARK-8990) to branch-1.4. Author: Cheng Lian <[email protected]> Closes apache#7351 from liancheng/spark-8990-for-1.4 and squashes the following commits: ffb5a73 [Cheng Lian] Backports PR apache#7347 (SPARK-8990) to branch-1.4
leonelfolmer · Jul 11, 2015 · 898e5f7 · 898e5f7
1 parent bef0591
commit 898e5f7
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 1 deletion.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
@@ -263,7 +263,7 @@ class DataFrameReader private[sql](sqlContext: SQLContext) {
       val globbedPaths = paths.map(new Path(_)).flatMap(SparkHadoopUtil.get.globPath).toArray
       sqlContext.baseRelationToDataFrame(
         new ParquetRelation2(
-          globbedPaths.map(_.toString), None, None, Map.empty[String, String])(sqlContext))
+          globbedPaths.map(_.toString), None, None, extraOptions.toMap)(sqlContext))
     }
   }
 

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetQuerySuite.scala
@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.parquet
 
+import org.apache.hadoop.fs.Path
 import org.scalatest.BeforeAndAfterAll
 
 import org.apache.spark.sql.types._
@@ -124,6 +125,24 @@ class ParquetQuerySuiteBase extends QueryTest with ParquetTest {
       checkAnswer(df2, df.collect().toSeq)
     }
   }
+
+  test("SPARK-8990 DataFrameReader.parquet() should respect user specified options") {
+    withTempPath { dir =>
+      val basePath = dir.getCanonicalPath
+      sqlContext.range(0, 10).toDF("a").write.parquet(new Path(basePath, "foo=1").toString)
+      sqlContext.range(0, 10).toDF("b").write.parquet(new Path(basePath, "foo=a").toString)
+
+      assertResult(2) {
+        // Disables schema merging via data source option
+        sqlContext.read.option("mergeSchema", "false").parquet(basePath).columns.length
+      }
+
+      assertResult(3) {
+        // Enables schema merging via data source option
+        sqlContext.read.option("mergeSchema", "true").parquet(basePath).columns.length
+      }
+    }
+  }
 }
 
 class ParquetDataSourceOnQuerySuite extends ParquetQuerySuiteBase with BeforeAndAfterAll {