fix

Author: gatorsmile

sql/core/src/main/scala/org/apache/spark/sql/execution/OptimizeMetadataOnlyQuery.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.execution
 
+import java.util.Locale
+
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.catalog.{HiveTableRelation, SessionCatalog}
@@ -81,9 +83,9 @@ case class OptimizeMetadataOnlyQuery(catalog: SessionCatalog) extends Rule[Logic
   private def getPartitionAttrs(
       partitionColumnNames: Seq[String],
       relation: LogicalPlan): Seq[Attribute] = {
-    val attrMap = relation.output.map(_.name).zip(relation.output).toMap
+    val attrMap = relation.output.map(_.name.toLowerCase(Locale.ROOT)).zip(relation.output).toMap
     partitionColumnNames.map { colName =>
-      attrMap.getOrElse(colName,
+      attrMap.getOrElse(colName.toLowerCase(Locale.ROOT),
         throw new AnalysisException(s"Unable to find the column `$colName` " +
           s"given [${relation.output.map(_.name).mkString(", ")}]")
       )

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

@@ -67,6 +67,9 @@ case class HadoopFsRelation(
     }
   }
 
+  // When data schema and partition schema have the overlapped columns, the output
+  // schema respects the order of data schema for the overlapped columns, but respect
+  // the data types of partition schema
   val schema: StructType = {
     StructType(dataSchema.map(f => overlappedPartCols.getOrElse(getColName(f), f)) ++
       partitionSchema.filterNot(f => overlappedPartCols.contains(getColName(f))))