[SPARK-27162][SQL] Add new method asCaseSensitiveMap in CaseInsensitiveStringMap

sql/catalyst/src/main/java/org/apache/spark/sql/util/CaseInsensitiveStringMap.java

@@ -40,9 +40,12 @@ public static CaseInsensitiveStringMap empty() {
     return new CaseInsensitiveStringMap(new HashMap<>(0));
   }
 
+  private final Map<String, String> original;
+
   private final Map<String, String> delegate;
 
   public CaseInsensitiveStringMap(Map<String, String> originalMap) {
+    this.original = new HashMap<>(originalMap.size());
     this.delegate = new HashMap<>(originalMap.size());
     putAll(originalMap);
   }
@@ -78,11 +81,13 @@ public String get(Object key) {
 
   @Override
   public String put(String key, String value) {
+    original.put(key, value);
     return delegate.put(toLowerCase(key), value);
   }
 
   @Override
   public String remove(Object key) {
+    original.remove(key);
     return delegate.remove(toLowerCase(key));
   }
 
@@ -95,6 +100,7 @@ public void putAll(Map<? extends String, ? extends String> m) {
 
   @Override
   public void clear() {
+    original.clear();
     delegate.clear();
   }
 
@@ -157,4 +163,11 @@ public double getDouble(String key, double defaultValue) {
     String value = get(key);
     return value == null ? defaultValue : Double.parseDouble(value);
   }
+
+  /**
+   * Returns the original case-sensitive map.
+   */
+  public Map<String, String> getOriginalMap() {
+    return original;
+  }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/util/CaseInsensitiveStringMapSuite.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.util
 
+import java.util
+
 import scala.collection.JavaConverters._
 
 import org.apache.spark.SparkFunSuite
@@ -80,4 +82,30 @@ class CaseInsensitiveStringMapSuite extends SparkFunSuite {
       options.getDouble("foo", 0.1d)
     }
   }
+
+  test("getOriginalMap") {
+    val originalMap = new util.HashMap[String, String] {
+      put("Foo", "Bar")
+      put("OFO", "ABR")
+      put("OoF", "bar")
+    }
+
+    val options = new CaseInsensitiveStringMap(originalMap)
+    assert(options.getOriginalMap.equals(originalMap))
+
+    val key = "Key"
+    val value = "value"
+    originalMap.put(key, value)
+    options.put(key, value)
+    originalMap.put(key, value)
+    assert(options.getOriginalMap.equals(originalMap))
+
+    val removedKey = "OFO"
+    originalMap.remove(removedKey)
+    options.remove(removedKey)
+    assert(options.getOriginalMap.equals(originalMap))
+
+    options.clear()
+    assert(options.getOriginalMap.isEmpty)
+  }
 }

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

@@ -35,11 +35,11 @@ abstract class FileTable(
   extends Table with SupportsBatchRead with SupportsBatchWrite {
 
   lazy val fileIndex: PartitioningAwareFileIndex = {
-    val scalaMap = options.asScala.toMap
-    val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(scalaMap)
+    val hadoopConf = sparkSession.sessionState.newHadoopConfWithCaseInsensitiveOptions(options)
     val rootPathsSpecified = DataSource.checkAndGlobPathIfNecessary(paths, hadoopConf,
       checkEmptyGlobPath = true, checkFilesExist = true)
     val fileStatusCache = FileStatusCache.getOrCreate(sparkSession)
+    val scalaMap = options.getOriginalMap.asScala.toMap
     new InMemoryFileIndex(
       sparkSession, rootPathsSpecified, scalaMap, userSpecifiedSchema, fileStatusCache)
   }

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

@@ -64,16 +64,14 @@ abstract class FileWriteBuilder(options: CaseInsensitiveStringMap, paths: Seq[St
     val sparkSession = SparkSession.active
     validateInputs(sparkSession.sessionState.conf.caseSensitiveAnalysis)
     val path = new Path(paths.head)
-    val optionsAsScala = options.asScala.toMap
-
-    val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(optionsAsScala)
+    val hadoopConf = sparkSession.sessionState.newHadoopConfWithCaseInsensitiveOptions(options)
     val job = getJobInstance(hadoopConf, path)
     val committer = FileCommitProtocol.instantiate(
       sparkSession.sessionState.conf.fileCommitProtocolClass,
       jobId = java.util.UUID.randomUUID().toString,
       outputPath = paths.head)
     lazy val description =
-      createWriteJobDescription(sparkSession, hadoopConf, job, paths.head, optionsAsScala)
+      createWriteJobDescription(sparkSession, hadoopConf, job, paths.head, options.asScala.toMap)
 
     val fs = path.getFileSystem(hadoopConf)
     mode match {

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcScanBuilder.scala

@@ -37,7 +37,7 @@ case class OrcScanBuilder(
     dataSchema: StructType,
     options: CaseInsensitiveStringMap)
   extends FileScanBuilder(schema) with SupportsPushDownFilters {
-  lazy val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(options.asScala.toMap)
+  lazy val hadoopConf = sparkSession.sessionState.newHadoopConfWithCaseInsensitiveOptions(options)
 
   override def build(): Scan = {
     OrcScan(sparkSession, hadoopConf, fileIndex, dataSchema, readSchema, options)

sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala

@@ -19,6 +19,8 @@ package org.apache.spark.sql.internal
 
 import java.io.File
 
+import scala.collection.JavaConverters._
+
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 
@@ -32,7 +34,7 @@ import org.apache.spark.sql.catalyst.parser.ParserInterface
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution._
 import org.apache.spark.sql.streaming.StreamingQueryManager
-import org.apache.spark.sql.util.{ExecutionListenerManager, QueryExecutionListener}
+import org.apache.spark.sql.util.{CaseInsensitiveStringMap, ExecutionListenerManager, QueryExecutionListener}
 
 /**
  * A class that holds all session-specific state in a given [[SparkSession]].
@@ -96,6 +98,11 @@ private[sql] class SessionState(
     hadoopConf
   }
 
+  def newHadoopConfWithCaseInsensitiveOptions(options: CaseInsensitiveStringMap): Configuration = {
+    // Hadoop configurations are case sensitive.
+    newHadoopConfWithOptions(options.getOriginalMap.asScala.toMap)
+  }
+
   /**
    * Get an identical copy of the `SessionState` and associate it with the given `SparkSession`
    */

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

@@ -19,6 +19,8 @@ package org.apache.spark.sql.execution.datasources.orc
 
 import java.io.File
 
+import org.apache.hadoop.fs.{Path, PathFilter}
+
 import org.apache.spark.SparkConf
 import org.apache.spark.sql._
 import org.apache.spark.sql.internal.SQLConf
@@ -30,6 +32,10 @@ case class OrcParData(intField: Int, stringField: String)
 // The data that also includes the partitioning key
 case class OrcParDataWithKey(intField: Int, pi: Int, stringField: String, ps: String)
 
+class TestFileFilter extends PathFilter {
+  override def accept(path: Path): Boolean = path.getParent.getName != "p=2"
+}
+
 abstract class OrcPartitionDiscoveryTest extends OrcTest {
   val defaultPartitionName = "__HIVE_DEFAULT_PARTITION__"
 
@@ -226,6 +232,23 @@ abstract class OrcPartitionDiscoveryTest extends OrcTest {
       }
     }
   }
+
+  test("SPARK-27162: handle pathfilter configuration correctly") {
+    withTempPath { dir =>
+      val path = dir.getCanonicalPath
+
+      val df = spark.range(2)
+      df.write.orc(path + "/p=1")
+      df.write.orc(path + "/p=2")
+      assert(spark.read.orc(path).count() === 4)
+
+      val extraOptions = Map(
+        "mapred.input.pathFilter.class" -> classOf[TestFileFilter].getName,
+        "mapreduce.input.pathFilter.class" -> classOf[TestFileFilter].getName
+      )
+      assert(spark.read.options(extraOptions).orc(path).count() === 2)
+    }
+  }
 }
 
 class OrcPartitionDiscoverySuite extends OrcPartitionDiscoveryTest with SharedSQLContext