fix

zml1206 · Jan 8, 2025 · d74dc60 · d74dc60
1 parent 76d9b6d
commit d74dc60
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 29 deletions.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -1281,8 +1281,8 @@ object CollapseProject extends Rule[LogicalPlan] with AliasHelper {
       } else {
         true
       }
-    // Alias and ExtractValue are very cheap.
-    case _: Alias | _: ExtractValue => e.children.forall(isCheap)
+    // Alias, ExtractValue and CreateNamedStruct are very cheap.
+    case _: Alias | _: ExtractValue | _: CreateNamedStruct => e.children.forall(isCheap)
     case _ => false
   }
 
@@ -1837,7 +1837,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
         case a: Alias if exprIdSet.contains(a.exprId) => a.toAttribute
         case e => e
       }
-      project.copy(child = Filter(newCondition, grandChild), projectList = newProjectList)
+      Project(newProjectList, Filter(newCondition, grandChild))
 
     // We can push down deterministic predicate through Aggregate, including throwable predicate.
     // If we can push down a filter through Aggregate, it means the filter only references the

diff --git a/...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala b/...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala
@@ -129,8 +129,8 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
               "Cannot rewrite canonicalized Common expression definitions")
           }
 
-          if (CollapseProject.isCheap(child) ||
-            (originAlias.isEmpty && !commonExprIdSet.contains(id))) {
+          if (originAlias.isEmpty &&
+            (CollapseProject.isCheap(child) || !commonExprIdSet.contains(id))) {
             refToExpr(id) = child
           } else {
             val childPlanIndex = inputPlans.indexWhere(

diff --git a/...catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala b/...catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
@@ -1557,30 +1557,30 @@ class FilterPushdownSuite extends PlanTest {
   }
 
   test("SPARK-50589: avoid extra expression duplication when push filter") {
-//    withSQLConf(SQLConf.USE_COMMON_EXPR_ID_FOR_ALIAS.key -> "false") {
-//      // through project
-//      val originalQuery1 = testRelation
-//        .select($"a" + $"b" as "add", $"a" - $"b" as "sub")
-//        .where($"add" < 10 && $"add" + $"add" > 10 && $"sub" > 0)
-//        .analyze
-//      val optimized1 = Optimize.execute(originalQuery1)
-//      comparePlans(optimized1, originalQuery1)
-//
-//      // through aggregate
-//      val originalQuery2 = testRelation
-//        .groupBy($"a")($"a", $"a" + $"a" as "add", abs($"a") as "abs", count(1) as "ct")
-//        .where($"add" < 10 && $"add" + $"add" > 10 && $"abs" > 5)
-//      val optimized2 = Optimize.execute(originalQuery2.analyze)
-//      val correctAnswer2 = testRelation
-//        .select($"a", $"a" + $"a" as "_common_expr_0")
-//        .where($"_common_expr_0" < 10 &&
-//          $"_common_expr_0" + $"_common_expr_0" > 10 &&
-//          abs($"a") > 5)
-//        .select($"a")
-//        .groupBy($"a")($"a", $"a" + $"a" as "add", abs($"a") as "abs", count(1) as "ct")
-//        .analyze
-//      comparePlans(optimized2, correctAnswer2)
-//    }
+    withSQLConf(SQLConf.USE_COMMON_EXPR_ID_FOR_ALIAS.key -> "false") {
+      // through project
+      val originalQuery1 = testRelation
+        .select($"a" + $"b" as "add", $"a" - $"b" as "sub")
+        .where($"add" < 10 && $"add" + $"add" > 10 && $"sub" > 0)
+        .analyze
+      val optimized1 = Optimize.execute(originalQuery1)
+      comparePlans(optimized1, originalQuery1)
+
+      // through aggregate
+      val originalQuery2 = testRelation
+        .groupBy($"a")($"a", $"a" + $"a" as "add", abs($"a") as "abs", count(1) as "ct")
+        .where($"add" < 10 && $"add" + $"add" > 10 && $"abs" > 5)
+      val optimized2 = Optimize.execute(originalQuery2.analyze)
+      val correctAnswer2 = testRelation
+        .select($"a", $"a" + $"a" as "_common_expr_0")
+        .where($"_common_expr_0" < 10 &&
+          $"_common_expr_0" + $"_common_expr_0" > 10 &&
+          abs($"a") > 5)
+        .select($"a")
+        .groupBy($"a")($"a", $"a" + $"a" as "add", abs($"a") as "abs", count(1) as "ct")
+        .analyze
+      comparePlans(optimized2, correctAnswer2)
+    }
     withSQLConf(SQLConf.USE_COMMON_EXPR_ID_FOR_ALIAS.key -> "false") {
       // partial push down
       val originalQuery3 = testRelation