fix

Author: zml1206

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

@@ -1998,16 +1998,19 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
       cond: Seq[Expression],
       aliasMap: AttributeMap[Alias]): Seq[Expression] = {
     if (!SQLConf.get.getConf(SQLConf.ALWAYS_INLINE_COMMON_EXPR)) {
-      val replaceWithMap = cond
-        .flatMap(_.collect {case a: Attribute => a })
-        .groupBy(identity)
-        .transform((_, v) => v.size)
-        .filter(m => aliasMap.contains(m._1) && m._2 > 1)
-        .map(m => m._1 -> trimAliases(aliasMap.getOrElse(m._1, m._1)))
-        .filter(m => needRewriteByWith(m._2))
-      val defsMap = AttributeMap(replaceWithMap.map(m => m._1 -> CommonExpressionDef(m._2)))
-      val refsMap = AttributeMap(defsMap.map(m => m._1 -> new CommonExpressionRef(m._2)))
-      cond.map(rewriteByWith(_, defsMap, refsMap))
+      val canRewriteConf = cond.filter(canRewriteByWith)
+      if (canRewriteConf.nonEmpty) {
+        val replaceWithMap = canRewriteConf.reduce(And)
+          .collect { case a: Attribute => a }
+          .groupBy(identity)
+          .transform((_, v) => v.size)
+          .filter(m => aliasMap.contains(m._1) && m._2 > 1)
+          .map(m => m._1 -> trimAliases(aliasMap.getOrElse(m._1, m._1)))
+          .filter(m => !CollapseProject.isCheap(m._2))
+        val defsMap = AttributeMap(replaceWithMap.map(m => m._1 -> CommonExpressionDef(m._2)))
+        val refsMap = AttributeMap(defsMap.map(m => m._1 -> new CommonExpressionRef(m._2)))
+        cond.map(rewriteByWith(_, defsMap, refsMap))
+      } else cond
     } else cond
   }
 
@@ -2019,15 +2022,18 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
     } else cond
   }
 
-  // With does not support common expression inline subquery.
-  private def needRewriteByWith(commonExpr: Expression): Boolean = {
-    !(commonExpr.containsPattern(PLAN_EXPRESSION) || CollapseProject.isCheap(commonExpr))
+  // With does not support inline subquery
+  private def canRewriteByWith(expr: Expression): Boolean = {
+    !expr.containsPattern(PLAN_EXPRESSION)
   }
 
   private def rewriteByWith(
       expr: Expression,
       defsMap: AttributeMap[CommonExpressionDef],
       refsMap: AttributeMap[CommonExpressionRef]): Expression = {
+    if (!canRewriteByWith(expr)) {
+      return expr
+    }
     val defs = mutable.HashSet.empty[CommonExpressionDef]
     val replaced = expr.transform {
       case a: Attribute if refsMap.contains(a) =>