SPARK-5684: Change the way partition values are passed to ParquetScan

Author: Yash Datta

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

@@ -49,7 +49,8 @@ private[sql] case class ParquetRelation(
     path: String,
     @transient conf: Option[Configuration],
     @transient sqlContext: SQLContext,
-    partitioningAttributes: Seq[Attribute] = Nil)
+    partitioningAttributes: Seq[Attribute] = Nil,
+    partitionValues: String = "")
   extends LeafNode with MultiInstanceRelation {
 
   self: Product =>
@@ -65,7 +66,7 @@ private[sql] case class ParquetRelation(
   override val output =
     partitioningAttributes ++
     ParquetTypesConverter.readSchemaFromFile(
-      new Path(path.split(",").head.split("->").head),
+      new Path(path.split(",").head),
       conf,
       sqlContext.conf.isParquetBinaryAsString,
       sqlContext.conf.isParquetINT96AsTimestamp)

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

@@ -88,10 +88,11 @@ private[sql] case class ParquetTableScan(
     val conf: Configuration = ContextUtil.getConfiguration(job)
 
     if (requestedPartitionOrdinals.nonEmpty) {
-      relation.path.split(",").foreach { tcurPath =>
-        val p = tcurPath.split("->")
-        val curPath = p.apply(0)
-        val partition = p.apply(1)
+      val partVals = relation.partitionValues.split(",")
+      var i = 0
+      relation.path.split(",").foreach { curPath =>
+        val partition = partVals.apply(i)
+        i += 1
         val qualifiedPath = {
           val path = new Path(curPath)
           path.getFileSystem(conf).makeQualified(path)

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

@@ -60,15 +60,16 @@ private[hive] trait HiveStrategies {
     implicit class LogicalPlanHacks(s: DataFrame) {
       def lowerCase: DataFrame = DataFrame(s.sqlContext, s.logicalPlan)
 
-      def addPartitioningAttributes(attrs: Seq[Attribute]): DataFrame = {
+      def addPartitioningAttributes(attrs: Seq[Attribute], partVals: String): DataFrame = {
         // Don't add the partitioning key if its already present in the data.
         if (attrs.map(_.name).toSet.subsetOf(s.logicalPlan.output.map(_.name).toSet)) {
           s
         } else {
           DataFrame(
             s.sqlContext,
             s.logicalPlan transform {
-              case p: ParquetRelation => p.copy(partitioningAttributes = attrs)
+              case p: ParquetRelation => p.copy(partitioningAttributes = attrs, 
+                partitionValues = partVals)
             })
         }
       }
@@ -137,15 +138,15 @@ private[hive] trait HiveStrategies {
               pruningCondition(inputData)
             }
 
-            val partitionLocations = partitions.map(part =>
-              part.getLocation() + "->" + part.getName())
+            val partitionLocations = partitions.map(part => part.getLocation)
+            val partitionNames = partitions.map(part => part.getName)
 
             if (partitionLocations.isEmpty) {
               PhysicalRDD(plan.output, sparkContext.emptyRDD[Row]) :: Nil
             } else {
               hiveContext
                 .parquetFile(partitionLocations: _*)
-                .addPartitioningAttributes(relation.partitionKeys)
+                .addPartitioningAttributes(relation.partitionKeys, partitionNames.mkString(","))
                 .lowerCase
                 .where(unresolvedOtherPredicates)
                 .select(unresolvedProjection: _*)