Addresses @rxin's comment, fixes UDT schema merging

Author: liancheng

python/pyspark/sql.py

@@ -1484,10 +1484,11 @@ def parquetFile(self, *paths):
         True
         """
         gateway = self._sc._gateway
-        jpaths = gateway.new_array(gateway.jvm.java.lang.String, len(paths))
-        for i in range(0, len(paths)):
+        jpath = paths[0]
+        jpaths = gateway.new_array(gateway.jvm.java.lang.String, len(paths) - 1)
+        for i in range(1, len(paths)):
             jpaths[i] = paths[i]
-        jdf = self._ssql_ctx.parquetFile(jpaths)
+        jdf = self._ssql_ctx.parquetFile(jpath, jpaths)
         return DataFrame(jdf, self)
 
     def jsonFile(self, path, schema=None, samplingRatio=1.0):

sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala

@@ -304,9 +304,9 @@ class SQLContext(@transient val sparkContext: SparkContext)
    * @group userf
    */
   @scala.annotation.varargs
-  def parquetFile(paths: String*): DataFrame =
+  def parquetFile(path: String, paths: String*): DataFrame =
     if (conf.parquetUseDataSourceApi) {
-      baseRelationToDataFrame(parquet.ParquetRelation2(paths, Map.empty)(this))
+      baseRelationToDataFrame(parquet.ParquetRelation2(path +: paths, Map.empty)(this))
     } else {
       DataFrame(this, parquet.ParquetRelation(
         paths.mkString(","), Some(sparkContext.hadoopConfiguration), this))

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetTypes.scala

@@ -567,6 +567,9 @@ private[parquet] object ParquetTypesConverter extends Logging {
             DecimalType.Fixed(rightPrecision, rightScale)) =>
         DecimalType(leftPrecision.max(rightPrecision), leftScale.max(rightScale))
 
+      case (leftUdt: UserDefinedType[_], rightUdt: UserDefinedType[_])
+        if leftUdt.userClass == rightUdt.userClass => leftUdt
+
       case (leftType, rightType) if leftType == rightType =>
         leftType
 

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveStrategies.scala

@@ -137,8 +137,10 @@ private[hive] trait HiveStrategies {
               pruningCondition(inputData)
             }
 
+            val partitionLocations = partitions.map(_.getLocation)
+
             hiveContext
-              .parquetFile(partitions.map(_.getLocation): _*)
+              .parquetFile(partitionLocations.head, partitionLocations.tail: _*)
               .addPartitioningAttributes(relation.partitionKeys)
               .lowerCase
               .where(unresolvedOtherPredicates)