[GLUTEN-11797][CORE] Improve the performance of getDistinctPartitionReadFileFormats for HiveTableScanExecTransformer

Author: beliefer

gluten-substrait/src/main/scala/org/apache/spark/sql/hive/HiveTableScanExecTransformer.scala

@@ -37,7 +37,7 @@ import org.apache.spark.util.Utils
 import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat
 import org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat
 import org.apache.hadoop.hive.ql.plan.TableDesc
-import org.apache.hadoop.mapred.TextInputFormat
+import org.apache.hadoop.mapred.{InputFormat, TextInputFormat}
 
 import java.net.URI
 
@@ -76,17 +76,46 @@ case class HiveTableScanExecTransformer(
   override def getPartitionWithReadFileFormats: Seq[(Partition, ReadFileFormat)] =
     partitionWithReadFileFormats
 
-  override def getDistinctPartitionReadFileFormats: Set[ReadFileFormat] =
-    if (
-      relation.isPartitioned &&
-      basePrunedPartitions.exists(_.getInputFormatClass != tableDesc.getInputFileFormatClass)
-    ) {
-      basePrunedPartitions.map {
-        partition => getReadFileFormat(HiveClientImpl.fromHivePartition(partition).storage)
-      }.toSet
-    } else {
+  // Only used for file format validation on the driver side. Must not trigger subquery execution.
+  override def getDistinctPartitionReadFileFormats: Set[ReadFileFormat] = {
+    if (!relation.isPartitioned) {
+      return Set(fileFormat)
+    }
+    // Single pass: use getInputFormatClass (cheap) to classify each partition.
+    // formatCache deduplicates by (InputFormatClass, Option[serdeClass]) so that
+    // HiveClientImpl.fromHivePartition is called at most once per distinct format combination.
+    // The serde is included in the key only for TextInputFormat, since its ReadFileFormat
+    // depends on both the InputFormat and the serde (e.g. JsonSerDe -> JsonReadFormat).
+    val tableInputFormatClass = tableDesc.getInputFileFormatClass
+    var hasTableFormatPartitions = false
+    val formatCache =
+      collection.mutable.Map[(Class[_ <: InputFormat[_, _]], Option[String]), ReadFileFormat]()
+    basePrunedPartitions.foreach {
+      partition =>
+        val cls = partition.getInputFormatClass
+        if (cls == tableInputFormatClass) {
+          hasTableFormatPartitions = true
+        } else {
+          val serdeKey =
+            if (TEXT_INPUT_FORMAT_CLASS.isAssignableFrom(cls)) {
+              Option(partition.getTPartition.getSd.getSerdeInfo.getSerializationLib)
+            } else {
+              None
+            }
+          formatCache.getOrElseUpdate(
+            (cls, serdeKey),
+            getReadFileFormat(HiveClientImpl.fromHivePartition(partition).storage))
+        }
+    }
+    val otherFormats = formatCache.values.toSet
+    if (otherFormats.isEmpty) {
       Set(fileFormat)
+    } else if (hasTableFormatPartitions) {
+      otherFormats + fileFormat
+    } else {
+      otherFormats
     }
+  }
 
   override def getPartitionSchema: StructType = relation.tableMeta.partitionSchema