Enables HDFS style globbing

Author: liancheng

core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala

@@ -231,6 +231,13 @@ class SparkHadoopUtil extends Logging {
     recurse(baseStatus)
   }
 
+  def globPath(pattern: Path): Seq[Path] = {
+    val fs = pattern.getFileSystem(conf)
+    Option(fs.globStatus(pattern)).map { statuses =>
+      statuses.map(_.getPath.makeQualified(fs.getUri, fs.getWorkingDirectory)).toSeq
+    }.getOrElse(Seq.empty[Path])
+  }
+
   /**
    * Lists all the files in a directory with the specified prefix, and does not end with the
    * given suffix. The returned {{FileStatus}} instances are sorted by the modification times of

sql/core/src/main/scala/org/apache/spark/sql/sources/ddl.scala

@@ -20,7 +20,10 @@ package org.apache.spark.sql.sources
 import scala.language.{existentials, implicitConversions}
 import scala.util.matching.Regex
 
+import org.apache.hadoop.fs.Path
+
 import org.apache.spark.Logging
+import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.sql.catalyst.AbstractSparkSQLParser
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Row}
@@ -157,7 +160,7 @@ private[sql] class DDLParser(
   protected lazy val className: Parser[String] = repsep(ident, ".") ^^ { case s => s.mkString(".")}
 
   override implicit def regexToParser(regex: Regex): Parser[String] = acceptMatch(
-    s"identifier matching regex ${regex}", {
+    s"identifier matching regex $regex", {
       case lexical.Identifier(str) if regex.unapplySeq(str).isDefined => str
       case lexical.Keyword(str) if regex.unapplySeq(str).isDefined => str
     }
@@ -230,11 +233,18 @@ private[sql] object ResolvedDataSource {
             Some(partitionColumnsSchema(schema, partitionColumns))
           }
 
+          val caseInsensitiveOptions= new CaseInsensitiveMap(options)
+          val paths = {
+            val patternPath = new Path(caseInsensitiveOptions("path"))
+            SparkHadoopUtil.get.globPath(patternPath).map(_.toString).toArray
+          }
+
           dataSource.createRelation(
             sqlContext,
+            paths,
             Some(schema),
             maybePartitionsSchema,
-            new CaseInsensitiveMap(options))
+            caseInsensitiveOptions)
         case dataSource: org.apache.spark.sql.sources.RelationProvider =>
           throw new AnalysisException(s"$className does not allow user-specified schemas.")
         case _ =>
@@ -245,7 +255,12 @@ private[sql] object ResolvedDataSource {
         case dataSource: RelationProvider =>
           dataSource.createRelation(sqlContext, new CaseInsensitiveMap(options))
         case dataSource: FSBasedRelationProvider =>
-          dataSource.createRelation(sqlContext, None, None, new CaseInsensitiveMap(options))
+          val caseInsensitiveOptions = new CaseInsensitiveMap(options)
+          val paths = {
+            val patternPath = new Path(caseInsensitiveOptions("path"))
+            SparkHadoopUtil.get.globPath(patternPath).map(_.toString).toArray
+          }
+          dataSource.createRelation(sqlContext, paths, None, None, caseInsensitiveOptions)
         case dataSource: org.apache.spark.sql.sources.SchemaRelationProvider =>
           throw new AnalysisException(
             s"A schema needs to be specified when using $className.")
@@ -280,11 +295,22 @@ private[sql] object ResolvedDataSource {
       case dataSource: CreatableRelationProvider =>
         dataSource.createRelation(sqlContext, mode, options, data)
       case dataSource: FSBasedRelationProvider =>
+        // Don't glob path for the write path.  The contracts here are:
+        //  1. Only one output path can be specified on the write path;
+        //  2. Output path must be a legal HDFS style file system path;
+        //  3. It's OK that the output path doesn't exist yet;
+        val caseInsensitiveOptions = new CaseInsensitiveMap(options)
+        val outputPath = {
+          val path = new Path(caseInsensitiveOptions("path"))
+          val fs = path.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+          path.makeQualified(fs.getUri, fs.getWorkingDirectory)
+        }
         val r = dataSource.createRelation(
           sqlContext,
+          Array(outputPath.toString),
           Some(data.schema),
           Some(partitionColumnsSchema(data.schema, partitionColumns)),
-          options)
+          caseInsensitiveOptions)
         sqlContext.executePlan(
           InsertIntoFSBasedRelation(
             r,

sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala

@@ -114,6 +114,7 @@ trait FSBasedRelationProvider {
    */
   def createRelation(
       sqlContext: SQLContext,
+      paths: Array[String],
       schema: Option[StructType],
       partitionColumns: Option[StructType],
       parameters: Map[String, String]): FSBasedRelation

sql/hive/src/test/scala/org/apache/spark/sql/sources/FSBasedRelationSuite.scala

@@ -483,4 +483,43 @@ class FSBasedRelationSuite extends QueryTest with ParquetTest {
       assert(table("t").collect().isEmpty)
     }
   }
+
+  test("Hadoop style globbing") {
+    withTempPath { file =>
+      partitionedTestDF.save(
+        source = classOf[SimpleTextSource].getCanonicalName,
+        mode = SaveMode.Overwrite,
+        options = Map("path" -> file.getCanonicalPath),
+        partitionColumns = Seq("p1", "p2"))
+
+      val df = load(
+        source = classOf[SimpleTextSource].getCanonicalName,
+        options = Map(
+          "path" -> s"${file.getCanonicalPath}/p1=*/p2=???",
+          "dataSchema" -> dataSchema.json))
+
+      val expectedPaths = Set(
+        s"${file.getCanonicalFile}/p1=1/p2=foo",
+        s"${file.getCanonicalFile}/p1=2/p2=foo",
+        s"${file.getCanonicalFile}/p1=1/p2=bar",
+        s"${file.getCanonicalFile}/p1=2/p2=bar"
+      ).map { p =>
+        val path = new Path(p)
+        val fs = path.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+        path.makeQualified(fs.getUri, fs.getWorkingDirectory).toString
+      }
+
+      println(df.queryExecution)
+
+      val actualPaths = df.queryExecution.analyzed.collectFirst {
+        case LogicalRelation(relation: FSBasedRelation) =>
+          relation.paths.toSet
+      }.getOrElse {
+        fail("Expect an FSBasedRelation, but none could be found")
+      }
+
+      assert(actualPaths === expectedPaths)
+      checkAnswer(df, partitionedTestDF.collect())
+    }
+  }
 }

sql/hive/src/test/scala/org/apache/spark/sql/sources/SimpleTextRelation.scala

@@ -37,12 +37,12 @@ import org.apache.spark.sql.{Row, SQLContext}
 class SimpleTextSource extends FSBasedRelationProvider {
   override def createRelation(
       sqlContext: SQLContext,
+      paths: Array[String],
       schema: Option[StructType],
       partitionColumns: Option[StructType],
       parameters: Map[String, String]): FSBasedRelation = {
-    new SimpleTextRelation(
-      Array(parameters("path")), schema,
-      partitionColumns.getOrElse(StructType(Array.empty[StructField])), parameters)(sqlContext)
+    val partitionsSchema = partitionColumns.getOrElse(StructType(Array.empty[StructField]))
+    new SimpleTextRelation(paths, schema, partitionsSchema, parameters)(sqlContext)
   }
 }