dedup first

Author: yucai

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

@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources.parquet
 
 import java.io.IOException
 import java.net.URI
+import java.util.Locale
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
@@ -36,7 +37,7 @@ import org.apache.parquet.hadoop._
 import org.apache.parquet.hadoop.ParquetOutputFormat.JobSummaryLevel
 import org.apache.parquet.hadoop.codec.CodecConfig
 import org.apache.parquet.hadoop.util.ContextUtil
-import org.apache.parquet.schema.MessageType
+import org.apache.parquet.schema.{GroupType, MessageType}
 
 import org.apache.spark.{SparkException, TaskContext}
 import org.apache.spark.internal.Logging
@@ -367,11 +368,32 @@ class ParquetFileFormat
 
       val sharedConf = broadcastedHadoopConf.value.value
 
-      lazy val footerFileMetaData =
+      val footerFileMetaData =
         ParquetFileReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData
+      val parquetSchema = footerFileMetaData.getSchema
+
+      def checkDuplicateFields(parquetRecord: GroupType): Unit = {
+        val fields = parquetRecord.getFields.asScala
+        val fieldMap = fields.groupBy(_.getName.toLowerCase(Locale.ROOT))
+        fieldMap.foreach { case (_, types) =>
+          if (types.size > 1) {
+            // Need to fail if there is ambiguity, i.e. more than one field is duplicate
+            val typesString = types.map(_.getName).mkString("[", ", ", "]")
+            throw new RuntimeException(s"Found duplicate field(s):" +
+              s"$typesString in case-insensitive mode")
+          }
+        }
+
+        fields.filter(!_.isPrimitive).foreach { groupField =>
+          checkDuplicateFields(groupField.asGroupType())
+        }
+      }
+      if (!isCaseSensitive) {
+        checkDuplicateFields(parquetSchema)
+      }
+
       // Try to push down filters when filter push-down is enabled.
       val pushed = if (enableParquetFilterPushDown) {
-        val parquetSchema = footerFileMetaData.getSchema
         val parquetFilters = new ParquetFilters(pushDownDate, pushDownTimestamp, pushDownDecimal,
           pushDownStringStartWith, pushDownInFilterThreshold, isCaseSensitive)
         filters

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

@@ -364,24 +364,16 @@ private[parquet] class ParquetFilters(
     // Here we don't flatten the fields in the nested schema but just look up through
     // root fields. Currently, accessing to nested fields does not push down filters
     // and it does not support to create filters for them.
-    val primitiveFields =
+    val primitiveFieldMap =
       dataType.getFields.asScala.filter(_.isPrimitive).map(_.asPrimitiveType()).map { f =>
         f.getName -> ParquetField(f.getName,
           ParquetSchemaType(f.getOriginalType,
             f.getPrimitiveTypeName, f.getTypeLength, f.getDecimalMetadata))
-      }
+      }.toMap
     if (caseSensitive) {
-      primitiveFields.toMap
+      primitiveFieldMap
     } else {
-      // Don't consider ambiguity here, i.e. more than one field is matched in case insensitive
-      // mode, just skip pushdown for these fields, they will trigger Exception when reading,
-      // See: SPARK-25132.
-      val dedupPrimitiveFields =
-        primitiveFields
-          .groupBy(_._1.toLowerCase(Locale.ROOT))
-          .filter(_._2.size == 1)
-          .mapValues(_.head._2)
-      CaseInsensitiveMap(dedupPrimitiveFields)
+      CaseInsensitiveMap(primitiveFieldMap)
     }
   }
 

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

@@ -30,6 +30,7 @@ import org.apache.parquet.schema.Type.Repetition
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
@@ -292,33 +293,14 @@ private[parquet] object ParquetReadSupport {
   private def clipParquetGroupFields(
       parquetRecord: GroupType, structType: StructType, caseSensitive: Boolean): Seq[Type] = {
     val toParquet = new SparkToParquetSchemaConverter(writeLegacyParquetFormat = false)
-    if (caseSensitive) {
-      val caseSensitiveParquetFieldMap =
-        parquetRecord.getFields.asScala.map(f => f.getName -> f).toMap
-      structType.map { f =>
-        caseSensitiveParquetFieldMap
-          .get(f.name)
-          .map(clipParquetType(_, f.dataType, caseSensitive))
-          .getOrElse(toParquet.convertField(f))
-      }
-    } else {
-      // Do case-insensitive resolution only if in case-insensitive mode
-      val caseInsensitiveParquetFieldMap =
-        parquetRecord.getFields.asScala.groupBy(_.getName.toLowerCase(Locale.ROOT))
-      structType.map { f =>
-        caseInsensitiveParquetFieldMap
-          .get(f.name.toLowerCase(Locale.ROOT))
-          .map { parquetTypes =>
-            if (parquetTypes.size > 1) {
-              // Need to fail if there is ambiguity, i.e. more than one field is matched
-              val parquetTypesString = parquetTypes.map(_.getName).mkString("[", ", ", "]")
-              throw new RuntimeException(s"""Found duplicate field(s) "${f.name}": """ +
-                s"$parquetTypesString in case-insensitive mode")
-            } else {
-              clipParquetType(parquetTypes.head, f.dataType, caseSensitive)
-            }
-          }.getOrElse(toParquet.convertField(f))
-      }
+    val fieldMap = parquetRecord.getFields.asScala.map(f => f.getName -> f).toMap
+    val finalParquetFieldMap =
+      if (caseSensitive) fieldMap else CaseInsensitiveMap(fieldMap)
+    structType.map { f =>
+      finalParquetFieldMap
+        .get(f.name)
+        .map(clipParquetType(_, f.dataType, caseSensitive))
+        .getOrElse(toParquet.convertField(f))
     }
   }
 

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

@@ -1106,7 +1106,7 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex
   }
 
   test("SPARK-25207: exception when duplicate fields in case-insensitive mode") {
-    withTempDir { dir =>
+    withTempPath { dir =>
       val tableName = "spark_25207"
       val tableDir = dir.getAbsoluteFile + "/table"
       withTable(tableName) {