SPARK-26990: use user specified field names if possible

Author: gengliangwang

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

@@ -111,15 +111,16 @@ object PartitioningUtils {
       caseSensitive: Boolean,
       validatePartitionColumns: Boolean,
       timeZone: TimeZone): PartitionSpec = {
-    val userSpecifiedDataTypes = if (userSpecifiedSchema.isDefined) {
+    val (userSpecifiedDataTypes, userSpecifiedNames) = if (userSpecifiedSchema.isDefined) {
       val nameToDataType = userSpecifiedSchema.get.fields.map(f => f.name -> f.dataType).toMap
+      val nameToName = userSpecifiedSchema.get.fields.map(f => f.name -> f.name).toMap
       if (!caseSensitive) {
-        CaseInsensitiveMap(nameToDataType)
+        (CaseInsensitiveMap(nameToDataType), CaseInsensitiveMap(nameToName))
       } else {
-        nameToDataType
+        (nameToDataType, nameToName)
       }
     } else {
-      Map.empty[String, DataType]
+      (Map.empty[String, DataType], Map.empty[String, String])
     }
 
     val dateFormatter = DateFormatter()
@@ -170,7 +171,9 @@ object PartitioningUtils {
         columnNames.zip(literals).map { case (name, Literal(_, dataType)) =>
           // We always assume partition columns are nullable since we've no idea whether null values
           // will be appended in the future.
-          StructField(name, userSpecifiedDataTypes.getOrElse(name, dataType), nullable = true)
+          val resultName = userSpecifiedNames.getOrElse(name, name)
+          val resultDataType = userSpecifiedDataTypes.getOrElse(name, dataType)
+          StructField(resultName, resultDataType, nullable = true)
         }
       }
 

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

@@ -65,6 +65,21 @@ class FileIndexSuite extends SharedSQLContext {
     }
   }
 
+  test("SPARK-26990: use user specified field names if possible") {
+    withTempDir { dir =>
+      val partitionDirectory = new File(dir, "a=foo")
+      partitionDirectory.mkdir()
+      val file = new File(partitionDirectory, "text.txt")
+      stringToFile(file, "text")
+      val path = new Path(dir.getCanonicalPath)
+      val schema = StructType(Seq(StructField("A", StringType, false)))
+      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+        val fileIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty, Some(schema))
+        assert(fileIndex.partitionSchema.length == 1 && fileIndex.partitionSchema.head.name == "A")
+      }
+    }
+  }
+
   test("SPARK-26230: if case sensitive, validate partitions with original column names") {
     withTempDir { dir =>
       val partitionDirectory = new File(dir, "a=1")