Do not pass caseSensitive arument into ParquetRowConverter use

SQLConf.get instead.

* Also non-existing column in parquet requested column is filled null,
so do not calling getOrElse directly applying instead.

Author: kimtkyeom

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

@@ -122,15 +122,12 @@ class ParquetReadSupport(val convertTz: Option[ZoneId],
       keyValueMetaData: JMap[String, String],
       fileSchema: MessageType,
       readContext: ReadContext): RecordMaterializer[InternalRow] = {
-    val caseSensitive = conf.getBoolean(SQLConf.CASE_SENSITIVE.key,
-      SQLConf.CASE_SENSITIVE.defaultValue.get)
     val parquetRequestedSchema = readContext.getRequestedSchema
     new ParquetRecordMaterializer(
       parquetRequestedSchema,
       ParquetReadSupport.expandUDT(catalystRequestedSchema),
       new ParquetToSparkSchemaConverter(conf),
-      convertTz,
-      caseSensitive)
+      convertTz)
   }
 }
 

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

@@ -36,8 +36,7 @@ private[parquet] class ParquetRecordMaterializer(
     parquetSchema: MessageType,
     catalystSchema: StructType,
     schemaConverter: ParquetToSparkSchemaConverter,
-    convertTz: Option[ZoneId],
-    caseSensitive: Boolean)
+    convertTz: Option[ZoneId])
   extends RecordMaterializer[InternalRow] {
 
   private val rootConverter =
@@ -46,7 +45,6 @@ private[parquet] class ParquetRecordMaterializer(
       parquetSchema,
       catalystSchema,
       convertTz,
-      caseSensitive,
       NoopUpdater)
 
   override def getCurrentRecord: InternalRow = rootConverter.currentRecord

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

@@ -35,6 +35,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, CaseInsensitiveMap, DateTimeUtils, GenericArrayData}
 import org.apache.spark.sql.catalyst.util.DateTimeUtils.SQLTimestamp
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -126,7 +127,6 @@ private[parquet] class ParquetRowConverter(
     parquetType: GroupType,
     catalystType: StructType,
     convertTz: Option[ZoneId],
-    caseSensitive: Boolean,
     updater: ParentContainerUpdater)
   extends ParquetGroupConverter(updater) with Logging {
 
@@ -179,17 +179,15 @@ private[parquet] class ParquetRowConverter(
 
   // Converters for each field.
   private[this] val fieldConverters: Array[Converter with HasParentContainerUpdater] = {
-    val catalystFieldNameToIndex = if (caseSensitive) {
+    // (SPARK-31116) Use case insensitive map if spark.sql.caseSensitive is false
+    // to prevent throwing IllegalArgumentException when searching catalyst type's field index
+    val catalystFieldNameToIndex = if (SQLConf.get.caseSensitiveAnalysis) {
       catalystType.fieldNames.zipWithIndex.toMap
     } else {
       CaseInsensitiveMap(catalystType.fieldNames.zipWithIndex.toMap)
     }
     parquetType.getFields.asScala.map { parquetField =>
-      val fieldIndex = catalystFieldNameToIndex.getOrElse(parquetField.getName,
-        throw new RuntimeException(
-          s"${parquetField.getName} does not exist. " +
-            s"Available: ${catalystType.fieldNames.mkString(", ")}")
-      )
+      val fieldIndex = catalystFieldNameToIndex(parquetField.getName)
       val catalystField = catalystType(fieldIndex)
       // Converted field value should be set to the `fieldIndex`-th cell of `currentRow`
       newConverter(parquetField, catalystField.dataType, new RowUpdater(currentRow, fieldIndex))
@@ -362,7 +360,6 @@ private[parquet] class ParquetRowConverter(
           parquetType.asGroupType(),
           t,
           convertTz,
-          caseSensitive,
           wrappedUpdater)
 
       case t =>