address comments

Author: gengliangwang

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

@@ -63,6 +63,9 @@ private[sql] object PruneFileSourcePartitions
     val extraPartitionFilters =
       remainingFilterInCnf.filter(f => f.references.subsetOf(partitionSet))
 
+    // For the filters that can't be used for partition pruning, we simply use `remainingFilters`
+    // instead of using the non-convertible part from `remainingFilterInCnf`. Otherwise, the
+    // result filters can be very long.
     (ExpressionSet(partitionFilters ++ extraPartitionFilters), remainingFilters)
   }
 

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/PruneHiveTablePartitions.scala

@@ -27,7 +27,6 @@ import org.apache.spark.sql.catalyst.planning.PhysicalOperation
 import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.datasources.DataSourceStrategy
-import org.apache.spark.sql.execution.datasources.PruneFileSourcePartitions.CNFConversion
 import org.apache.spark.sql.internal.SQLConf
 
 /**
@@ -110,9 +109,7 @@ private[sql] class PruneHiveTablePartitions(session: SparkSession)
   override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
     case op @ PhysicalOperation(projections, filters, relation: HiveTableRelation)
       if filters.nonEmpty && relation.isPartitioned && relation.prunedPartitions.isEmpty =>
-      val predicates = CNFConversion(filters.reduceLeft(And))
-      val finalPredicates = if (predicates.nonEmpty) predicates else filters
-      val partitionKeyFilters = getPartitionKeyFilters(finalPredicates, relation)
+      val partitionKeyFilters = getPartitionKeyFilters(filters, relation)
       if (partitionKeyFilters.nonEmpty) {
         val newPartitions = prunePartitions(relation, partitionKeyFilters)
         val newTableMeta = updateTableMeta(relation.tableMeta, newPartitions)

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/PrunePartitionSuiteBase.scala

@@ -67,7 +67,7 @@ abstract class PrunePartitionSuiteBase extends QueryTest with SQLTestUtils with
     }
   }
 
-  test("SPARK-32284: Avoid pushing down too many predicates in partition pruning") {
+  test("SPARK-32284: Avoid expanding too many CNF predicates in partition pruning") {
     withTempView("temp") {
       withTable("t") {
         sql(