[SPARK-13271][SQL] Better error message if 'path' is not specified

Improved the error message as per discussion in https://github.com/apache/spark/pull/11034#discussion_r52111238. Also made `path` and `metadataPath` in FileStreamSource case insensitive.

Author: Shixiong Zhu <shixiong@databricks.com>

Closes #11154 from zsxwing/path.

3 files changed, 13 insertions, 6 deletions

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ResolvedDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ResolvedDataSource.scala
index cefa8be0c6..eec9070bee 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ResolvedDataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ResolvedDataSource.scala
@@ -156,7 +156,9 @@ object ResolvedDataSource extends Logging {
             }
             caseInsensitiveOptions.get("paths")
               .map(_.split("(?<!\\\\),").map(StringUtils.unEscapeString(_, '\\', ',')))
-              .getOrElse(Array(caseInsensitiveOptions("path")))
+              .getOrElse(Array(caseInsensitiveOptions.getOrElse("path", {
+                throw new IllegalArgumentException("'path' is not specified")
+              })))
               .flatMap{ pathString =>
                 val hdfsPath = new Path(pathString)
                 val fs = hdfsPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
@@ -197,7 +199,9 @@ object ResolvedDataSource extends Logging {
             }
             caseInsensitiveOptions.get("paths")
               .map(_.split("(?<!\\\\),").map(StringUtils.unEscapeString(_, '\\', ',')))
-              .getOrElse(Array(caseInsensitiveOptions("path")))
+              .getOrElse(Array(caseInsensitiveOptions.getOrElse("path", {
+                throw new IllegalArgumentException("'path' is not specified")
+              })))
               .flatMap{ pathString =>
                 val hdfsPath = new Path(pathString)
                 val fs = hdfsPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
@@ -260,7 +264,9 @@ object ResolvedDataSource extends Logging {
         //  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 path = new Path(caseInsensitiveOptions.getOrElse("path", {
+            throw new IllegalArgumentException("'path' is not specified")
+          }))
           val fs = path.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
           path.makeQualified(fs.getUri, fs.getWorkingDirectory)
         }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala b/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
index 428a313ca9..87ea7f510e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala
@@ -203,10 +203,11 @@ trait HadoopFsRelationProvider extends StreamSourceProvider {
       schema: Option[StructType],
       providerName: String,
       parameters: Map[String, String]): Source = {
-    val path = parameters.getOrElse("path", {
+    val caseInsensitiveOptions = new CaseInsensitiveMap(parameters)
+    val path = caseInsensitiveOptions.getOrElse("path", {
       throw new IllegalArgumentException("'path' is not specified")
     })
-    val metadataPath = parameters.getOrElse("metadataPath", s"$path/_metadata")
+    val metadataPath = caseInsensitiveOptions.getOrElse("metadataPath", s"$path/_metadata")
 
     def dataFrameBuilder(files: Array[String]): DataFrame = {
       val relation = createRelation(
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
index 0c288bdf8a..cd85e23d8b 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
@@ -548,7 +548,7 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv
                 "org.apache.spark.sql.json",
                 schema,
                 Map.empty[String, String])
-            }.getMessage.contains("key not found: path"),
+            }.getMessage.contains("'path' is not specified"),
             "We should complain that path is not specified.")
         }
       }