Data source options should be propagated in method checkAndGlobPathIfNecessary

Author: gengliangwang

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

@@ -570,7 +570,7 @@ case class DataSource(
       checkEmptyGlobPath: Boolean,
       checkFilesExist: Boolean): Seq[Path] = {
     val allPaths = caseInsensitiveOptions.get("path") ++ paths
-    val hadoopConf = sparkSession.sessionState.newHadoopConf()
+    val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(options)
 
     DataSource.checkAndGlobPathIfNecessary(allPaths.toSeq, hadoopConf,
       checkEmptyGlobPath, checkFilesExist)

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

@@ -19,11 +19,12 @@ package org.apache.spark.sql.execution.datasources
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileStatus, Path, RawLocalFileSystem}
+import org.scalatest.PrivateMethodTester
 
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.test.SharedSparkSession
 
-class DataSourceSuite extends SharedSparkSession {
+class DataSourceSuite extends SharedSparkSession with PrivateMethodTester {
   import TestPaths._
 
   test("test glob and non glob paths") {
@@ -132,6 +133,18 @@ class DataSourceSuite extends SharedSparkSession {
       )
     )
   }
+
+  test("Data source options should be propagated in method checkAndGlobPathIfNecessary") {
+    val fsDefaultName = "mockFs://mockFs"
+    val dataSource =
+      DataSource(spark, "parquet", Seq("/path1"), options = Map("fs.defaultFS" -> fsDefaultName))
+    val checkAndGlobPathIfNecessary = PrivateMethod[Seq[Path]]('checkAndGlobPathIfNecessary)
+
+    val message = intercept[java.io.IOException] {
+      dataSource invokePrivate checkAndGlobPathIfNecessary(false, false)
+    }.getMessage
+    assert(message.equals("No FileSystem for scheme: mockFs"))
+  }
 }
 
 object TestPaths {