add method PartitioningUtils.requestedPartitionColumnIds

Author: gengliangwang

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

@@ -539,6 +539,25 @@ object PartitioningUtils {
     }).asNullable
   }
 
+  def requestedPartitionColumnIds(
+      requiredSchema: StructType,
+      partitionSchema: StructType,
+      caseSensitive: Boolean): Array[Int] = {
+    val columnNameMap =
+      partitionSchema.fields.map(getColName(_, caseSensitive)).zipWithIndex.toMap
+    requiredSchema.fields.map { field =>
+      columnNameMap.getOrElse(getColName(field, caseSensitive), -1)
+    }
+  }
+
+  private def getColName(f: StructField, caseSensitive: Boolean): String = {
+    if (caseSensitive) {
+      f.name
+    } else {
+      f.name.toLowerCase(Locale.ROOT)
+    }
+  }
+
   private def columnNameEquality(caseSensitive: Boolean): (String, String) => Boolean = {
     if (caseSensitive) {
       org.apache.spark.sql.catalyst.analysis.caseSensitiveResolution

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

@@ -207,8 +207,8 @@ class OrcFileFormat
           val iter = new RecordReaderIterator(batchReader)
           Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => iter.close()))
           val requestedDataColIds = requestedColIds ++ Array.fill(partitionSchema.length)(-1)
-          val requestedPartitionColIds =
-            Array.fill(requiredSchema.length)(-1) ++ Range(0, partitionSchema.length)
+          val requestedPartitionColIds = PartitioningUtils.requestedPartitionColumnIds(
+            resultSchema, partitionSchema, isCaseSensitive)
           batchReader.initialize(fileSplit, taskAttemptContext)
           batchReader.initBatch(
             reader.getSchema,