[SPARK-14993] [SQL] Fix Partition Discovery Inconsistency when Input is a Path to Parquet File

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala

@@ -184,8 +184,10 @@ private[sql] object PartitioningUtils {
         return (None, None)
       }
 
-      if (basePaths.contains(currentPath)) {
+      if (basePaths.contains(currentPath) ||
+        basePaths.exists(_.toString.startsWith(currentPath.toString))) {
         // If the currentPath is one of base paths. We should stop.
+        // If the currentPath is the basis of one of base paths. We should stop.
         finished = true
       } else {
         // Let's say currentPath is a path of "/table/a=1/", currentPath.getName will give us a=1.

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala

@@ -191,20 +191,39 @@ class ParquetPartitionDiscoverySuite extends QueryTest with ParquetTest with Sha
     checkThrows[AssertionError]("file://path/a=", "Empty partition column value")
   }
 
+  test("parse partition with base paths") {
+    val partitionSpec: Option[PartitionValues] = parsePartition(
+      path = new Path("file://path/a=10"),
+      defaultPartitionName = defaultPartitionName,
+      typeInference = true,
+      basePaths = Set(new Path("file://path/a=10/p.parquet")))._1
+
+    assert(partitionSpec.isEmpty)
+  }
+
   test("parse partitions") {
     def check(
         paths: Seq[String],
         spec: PartitionSpec,
-        rootPaths: Set[Path] = Set.empty[Path]): Unit = {
+        rootPaths: Set[String] = Set.empty[String]): Unit = {
       val actualSpec =
         parsePartitions(
           paths.map(new Path(_)),
           defaultPartitionName,
           true,
-          rootPaths)
+          rootPaths.map(new Path(_)))
       assert(actualSpec === spec)
     }
 
+    check(Seq(
+      "hdfs://host:9000/path/a=10/b=20",
+      "hdfs://host:9000/path/a=10.5/_temporary",
+      "hdfs://host:9000/path/a=10.5/b=hello"),
+      PartitionSpec.emptySpec,
+      Set(
+        "hdfs://host:9000/path/a=10/b=20/a.parquet",
+        "hdfs://host:9000/path/a=10.5/b=hello/p.parquet"))
+
     check(Seq(
       "hdfs://host:9000/path/a=10/b=hello"),
       PartitionSpec(
@@ -413,6 +432,32 @@ class ParquetPartitionDiscoverySuite extends QueryTest with ParquetTest with Sha
     }
   }
 
+  test("read partitioned table using different path options") {
+    withTempDir { base =>
+      val pi = 1
+      val ps = "foo"
+      val path = makePartitionDir(base, defaultPartitionName, "pi" -> pi, "ps" -> ps)
+      makeParquetFile(
+        (1 to 10).map(i => ParquetData(i, i.toString)), path)
+
+      // when the input is the base path containing partitioned directories
+      val baseDf = sqlContext.read.parquet(base.getCanonicalPath)
+      assert(baseDf.schema.map(_.name) === Seq("intField", "stringField", "pi", "ps"))
+
+      // when the input is a path to a partitioned directory containing a parquet file
+      val partDf = sqlContext.read.parquet(path.getCanonicalPath)
+      assert(partDf.schema.map(_.name) === Seq("intField", "stringField"))
+
+      path.listFiles().foreach { f =>
+        if (f.getName.toLowerCase().endsWith(".parquet")) {
+          // when the input is a path to a parquet file
+          val df = sqlContext.read.parquet(f.getCanonicalPath)
+          assert(df.schema.map(_.name) === Seq("intField", "stringField"))
+        }
+      }
+    }
+  }
+
   test("read partitioned table - partition key included in Parquet file") {
     withTempDir { base =>
       for {