fix

Author: zml1206

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -1756,7 +1756,6 @@ object PushDownPredicates extends Rule[LogicalPlan] {
 object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelper {
   def apply(plan: LogicalPlan): LogicalPlan = plan transform applyLocally
 
-  val maxExprReusedInPushPredicate = conf.getConf(SQLConf.MAX_EXPRESSION_REUSED_IN_PUSH_PREDICATE)
   val applyLocally: PartialFunction[LogicalPlan, LogicalPlan] = {
     // SPARK-13473: We can't push the predicate down when the underlying projection output non-
     // deterministic field(s).  Non-deterministic expressions are essentially stateful. This
@@ -1885,17 +1884,19 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
    */
   def canCollapseExpression(
       consumer: Expression,
-      producerMap: Map[Attribute, Expression]): Boolean = maxExprReusedInPushPredicate match {
-    case Int.MaxValue => true
-    case _ =>
-      CollapseProject.collectReferences(consumer)
-        .groupBy(identity)
-        .transform((_, v) => v.size)
-        .forall {
-          case (reference, count) =>
-            val producer = producerMap.getOrElse(reference, reference)
-            count <= maxExprReusedInPushPredicate || CollapseProject.isCheap(trimAliases(producer))
-        }
+      producerMap: Map[Attribute, Expression]): Boolean =
+    conf.getConf(SQLConf.MAX_EXPRESSION_REUSED_IN_PUSH_PREDICATE) match {
+      case Int.MaxValue => true
+      case maxReused =>
+        CollapseProject.collectReferences(consumer)
+          .groupBy(identity)
+          .transform((_, v) => v.size)
+          .forall {
+            case (reference, count) =>
+              val producer = producerMap.getOrElse(reference, reference)
+              count <= maxReused || CollapseProject.isCheap(trimAliases(producer))
+          }
+
   }
 
   def canPushThrough(p: UnaryNode): Boolean = p match {