Add additional runs of columnPruningRules and WindowFilterPushDown

Author: xin-zhang2

presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q58.plan.txt

@@ -2,14 +2,14 @@ local exchange (GATHER, SINGLE, [])
     remote exchange (GATHER, SINGLE, [])
         join (INNER, PARTITIONED):
             join (INNER, PARTITIONED):
-                final aggregation over (i_item_id)
+                final aggregation over (i_item_id_79)
                     local exchange (GATHER, SINGLE, [])
-                        remote exchange (REPARTITION, HASH, [i_item_id])
-                            partial aggregation over (i_item_id)
+                        remote exchange (REPARTITION, HASH, [i_item_id_79])
+                            partial aggregation over (i_item_id_79)
                                 semijoin (REPLICATED):
                                     join (INNER, REPLICATED):
                                         join (INNER, REPLICATED):
-                                            scan store_sales
+                                            scan catalog_sales
                                             local exchange (GATHER, SINGLE, [])
                                                 remote exchange (REPLICATE, BROADCAST, [])
                                                     scan date_dim
@@ -25,14 +25,14 @@ local exchange (GATHER, SINGLE, [])
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 scan date_dim
-                final aggregation over (i_item_id_79)
+                final aggregation over (i_item_id_210)
                     local exchange (GATHER, SINGLE, [])
-                        remote exchange (REPARTITION, HASH, [i_item_id_79])
-                            partial aggregation over (i_item_id_79)
+                        remote exchange (REPARTITION, HASH, [i_item_id_210])
+                            partial aggregation over (i_item_id_210)
                                 semijoin (REPLICATED):
                                     join (INNER, REPLICATED):
                                         join (INNER, REPLICATED):
-                                            scan catalog_sales
+                                            scan web_sales
                                             local exchange (GATHER, SINGLE, [])
                                                 remote exchange (REPLICATE, BROADCAST, [])
                                                     scan date_dim
@@ -48,14 +48,14 @@ local exchange (GATHER, SINGLE, [])
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 scan date_dim
-            final aggregation over (i_item_id_210)
+            final aggregation over (i_item_id)
                 local exchange (GATHER, SINGLE, [])
-                    remote exchange (REPARTITION, HASH, [i_item_id_210])
-                        partial aggregation over (i_item_id_210)
+                    remote exchange (REPARTITION, HASH, [i_item_id])
+                        partial aggregation over (i_item_id)
                             semijoin (REPLICATED):
                                 join (INNER, REPLICATED):
                                     join (INNER, REPLICATED):
-                                        scan web_sales
+                                        scan store_sales
                                         local exchange (GATHER, SINGLE, [])
                                             remote exchange (REPLICATE, BROADCAST, [])
                                                 scan date_dim

presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java

@@ -733,6 +733,14 @@ public PlanOptimizers(
         builder.add(new SimplifyPlanWithEmptyInput(),
                 new PruneUnreferencedOutputs());
 
+        // Run columnPruningRules one more time to optimize the ProjectNode generated in SimplifyPlanWithEmptyInput.
+        builder.add(new IterativeOptimizer(
+                metadata,
+                ruleStats,
+                statsCalculator,
+                estimatedExchangesCostCalculator,
+                columnPruningRules));
+
         builder.add(new IterativeOptimizer(
                         metadata,
                         ruleStats,
@@ -886,6 +894,7 @@ public PlanOptimizers(
                         // Run RemoveEmptyDelete and EliminateEmptyJoins after table scan is removed by PickTableLayout/AddExchanges
                         ImmutableSet.of(new RemoveEmptyDelete())));
         builder.add(predicatePushDown); // Run predicate push down one more time in case we can leverage new information from layouts' effective predicate
+        builder.add(new WindowFilterPushDown(metadata));
         builder.add(new RemoveUnsupportedDynamicFilters(metadata.getFunctionAndTypeManager()));
         builder.add(simplifyRowExpressionOptimizer); // Should be always run after PredicatePushDown
         builder.add(projectionPushDown);

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/WindowFilterPushDown.java

@@ -51,7 +51,6 @@
 import static com.facebook.presto.common.type.BigintType.BIGINT;
 import static com.facebook.presto.expressions.LogicalRowExpressions.TRUE_CONSTANT;
 import static com.facebook.presto.sql.planner.plan.ChildReplacer.replaceChildren;
-import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Verify.verify;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static java.lang.Math.toIntExact;
@@ -116,7 +115,6 @@ public boolean isPlanChanged()
         @Override
         public PlanNode visitWindow(WindowNode node, RewriteContext<Void> context)
         {
-            checkState(node.getWindowFunctions().size() == 1, "WindowFilterPushdown requires that WindowNodes contain exactly one window function");
             PlanNode rewrittenSource = context.rewrite(node.getSource());
 
             if (canReplaceWithRowNumber(node, metadata.getFunctionAndTypeManager())) {