Generalize nested column selection

* MISC: Change excpetion in `ParquetRowConverter.fieldConverters` to
RuntimeException

Author: kimtkyeom

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

@@ -186,7 +186,7 @@ private[parquet] class ParquetRowConverter(
     }
     parquetType.getFields.asScala.map { parquetField =>
       val fieldIndex = catalystFieldNameToIndex.getOrElse(parquetField.getName,
-        throw new IllegalArgumentException(
+        throw new RuntimeException(
           s"${parquetField.getName} does not exist. " +
             s"Available: ${catalystType.fieldNames.mkString(", ")}")
       )

sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala

@@ -843,36 +843,41 @@ class FileBasedDataSourceSuite extends QueryTest
     }
   }
 
-  test("SPARK-31116: Select nested parquet with case insensitive mode") {
-    Seq("true", "false").foreach { nestedSchemaPruningEnabled =>
-      withSQLConf(
-        SQLConf.CASE_SENSITIVE.key -> "false",
-        SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> nestedSchemaPruningEnabled) {
-        withTempPath { dir =>
-          val path = dir.getCanonicalPath
-
-          // Prepare values for testing nested parquet data
-          spark
-            .range(1L)
-            .selectExpr("NAMED_STRUCT('lowercase', id, 'camelCase', id + 1) AS StructColumn")
-            .write.parquet(path)
-
-          val exactSchema = "StructColumn struct<lowercase: LONG, camelCase: LONG>"
-
-          checkAnswer(spark.read.schema(exactSchema).parquet(path), Row(Row(0, 1)))
-
-          // In case insensitive manner, parquet's column cases are ignored
-          val innerColumnCaseInsensitiveSchema =
-            "StructColumn struct<Lowercase: LONG, camelcase: LONG>"
-          checkAnswer(
-            spark.read.schema(innerColumnCaseInsensitiveSchema).parquet(path),
-            Row(Row(0, 1)))
-
-          val rootColumnCaseInsensitiveSchema =
-            "structColumn struct<lowercase: LONG, camelCase: LONG>"
-          checkAnswer(
-            spark.read.schema(rootColumnCaseInsensitiveSchema).parquet(path),
-            Row(Row(0, 1)))
+  test("SPARK-31116: Select nested schema with case insensitive mode") {
+    // This test case failed at only Parquet. ORC is added for test coverage parity.
+    Seq("orc", "parquet").foreach { format =>
+      Seq("true", "false").foreach { nestedSchemaPruningEnabled =>
+        withSQLConf(
+          SQLConf.CASE_SENSITIVE.key -> "false",
+          SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> nestedSchemaPruningEnabled) {
+          withTempPath { dir =>
+            val path = dir.getCanonicalPath
+
+            // Prepare values for testing nested parquet data
+            spark
+              .range(1L)
+              .selectExpr("NAMED_STRUCT('lowercase', id, 'camelCase', id + 1) AS StructColumn")
+              .write
+              .format(format)
+              .save(path)
+
+            val exactSchema = "StructColumn struct<lowercase: LONG, camelCase: LONG>"
+
+            checkAnswer(spark.read.schema(exactSchema).format(format).load(path), Row(Row(0, 1)))
+
+            // In case insensitive manner, parquet's column cases are ignored
+            val innerColumnCaseInsensitiveSchema =
+              "StructColumn struct<Lowercase: LONG, camelcase: LONG>"
+            checkAnswer(
+              spark.read.schema(innerColumnCaseInsensitiveSchema).format(format).load(path),
+              Row(Row(0, 1)))
+
+            val rootColumnCaseInsensitiveSchema =
+              "structColumn struct<lowercase: LONG, camelCase: LONG>"
+            checkAnswer(
+              spark.read.schema(rootColumnCaseInsensitiveSchema).format(format).load(path),
+              Row(Row(0, 1)))
+          }
         }
       }
     }