Add ReorderJoins rule to pick the best join order

Add a rule to enumerate join order possibilities for a join graph and
choose the least cost option.

This does a minimal form of cross join elimination, by only partitioning
nodes into groups that have at least one edge between them, which
eliminates some unnecessary cross joins from consideration.  It also
means that necessary cross joins will always be executed as late as
possible in the plan (which may be worse).

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

@@ -55,6 +55,7 @@
 import com.facebook.presto.sql.planner.iterative.rule.RemoveEmptyDelete;
 import com.facebook.presto.sql.planner.iterative.rule.RemoveFullSample;
 import com.facebook.presto.sql.planner.iterative.rule.RemoveRedundantIdentityProjections;
+import com.facebook.presto.sql.planner.iterative.rule.ReorderJoins;
 import com.facebook.presto.sql.planner.iterative.rule.SimplifyCountOverConstant;
 import com.facebook.presto.sql.planner.iterative.rule.SingleMarkDistinctToGroupBy;
 import com.facebook.presto.sql.planner.iterative.rule.SwapAdjacentWindowsBySpecifications;
@@ -322,13 +323,32 @@ public PlanOptimizers(
                         ImmutableList.of(new com.facebook.presto.sql.planner.optimizations.EliminateCrossJoins()), // This can pull up Filter and Project nodes from between Joins, so we need to push them down again
                         ImmutableSet.of(new EliminateCrossJoins())
                 ),
+
                 new PredicatePushDown(metadata, sqlParser),
                 new IterativeOptimizer(
                         stats,
                         statsCalculator,
                         estimatedExchangesCostCalculator,
                         ImmutableSet.of(new PushDownTableConstraints(metadata, sqlParser))),
-                projectionPushDown);
+                projectionPushDown,
+                new PruneUnreferencedOutputs(),
+                new IterativeOptimizer(
+                        stats,
+                        statsCalculator,
+                        estimatedExchangesCostCalculator,
+                        ImmutableSet.of(new RemoveRedundantIdentityProjections())
+                ),
+
+                // Because ReorderJoins runs only once,
+                // PredicatePushDown, PruneUnreferenedOutputpus and RemoveRedundantIdentityProjections
+                // need to run beforehand in order to produce an optimal join order
+                // It also needs to run after EliminateCrossJoins so that its chosen order doesn't get undone.
+                new IterativeOptimizer(
+                        stats,
+                        statsCalculator,
+                        estimatedExchangesCostCalculator,
+                        ImmutableSet.of(new ReorderJoins(costComparator))
+                ));
 
         if (featuresConfig.isOptimizeSingleDistinct()) {
             builder.add(

presto-main/src/main/java/com/facebook/presto/sql/planner/iterative/rule/MultiJoinNode.java

@@ -0,0 +1,116 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.facebook.presto.sql.planner.iterative.rule;
+
+import com.facebook.presto.sql.planner.Symbol;
+import com.facebook.presto.sql.planner.iterative.Lookup;
+import com.facebook.presto.sql.planner.plan.JoinNode;
+import com.facebook.presto.sql.planner.plan.PlanNode;
+import com.facebook.presto.sql.tree.Expression;
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static com.facebook.presto.sql.ExpressionUtils.and;
+import static com.facebook.presto.sql.planner.DeterminismEvaluator.isDeterministic;
+import static com.facebook.presto.sql.planner.plan.JoinNode.Type.INNER;
+import static com.facebook.presto.sql.tree.BooleanLiteral.TRUE_LITERAL;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This class represents a set of inner joins that can be executed in any order.
+ */
+class MultiJoinNode
+{
+    private static final int JOIN_LIMIT = 10;
+
+    private final List<PlanNode> sources;
+    private final Expression filter;
+    private final List<Symbol> outputSymbols;
+
+    public MultiJoinNode(List<PlanNode> sources, Expression filter, List<Symbol> outputSymbols)
+    {
+        this.sources = ImmutableList.copyOf(requireNonNull(sources, "sources is null"));
+        this.filter = requireNonNull(filter, "filter is null");
+        this.outputSymbols = ImmutableList.copyOf(requireNonNull(outputSymbols, "outputSymbols is null"));
+
+        List<Symbol> inputSymbols = sources.stream().flatMap(source -> source.getOutputSymbols().stream()).collect(toImmutableList());
+        checkArgument(inputSymbols.containsAll(outputSymbols), "inputs do not contain all output symbols");
+    }
+
+    public Expression getFilter()
+    {
+        return filter;
+    }
+
+    public List<PlanNode> getSources()
+    {
+        return sources;
+    }
+
+    public List<Symbol> getOutputSymbols()
+    {
+        return outputSymbols;
+    }
+
+    static MultiJoinNode toMultiJoinNode(JoinNode joinNode, Lookup lookup)
+    {
+        return new MultiJoinNodeBuilder(joinNode, lookup).toMultiJoinNode();
+    }
+
+    private static class MultiJoinNodeBuilder
+    {
+        private final List<PlanNode> sources = new ArrayList<>();
+        private final List<Expression> filters = new ArrayList<>();
+        private final List<Symbol> outputSymbols;
+        private final Lookup lookup;
+
+        MultiJoinNodeBuilder(JoinNode node, Lookup lookup)
+        {
+            requireNonNull(node, "node is null");
+            checkState(node.getType() == INNER, "join type must be INNER");
+            this.outputSymbols = node.getOutputSymbols();
+            this.lookup = requireNonNull(lookup, "lookup is null");
+            flattenNode(node);
+        }
+
+        private void flattenNode(PlanNode node)
+        {
+            PlanNode resolved = lookup.resolve(node);
+            if (resolved instanceof JoinNode && sources.size() < JOIN_LIMIT) {
+                JoinNode joinNode = (JoinNode) resolved;
+                if (joinNode.getType() == INNER && isDeterministic(joinNode.getFilter().orElse(TRUE_LITERAL))) {
+                    flattenNode(joinNode.getLeft());
+                    flattenNode(joinNode.getRight());
+                    joinNode.getCriteria().stream()
+                            .map(JoinNode.EquiJoinClause::toExpression)
+                            .forEach(filters::add);
+                    joinNode.getFilter().ifPresent(filters::add);
+                    return;
+                }
+            }
+            sources.add(node);
+        }
+
+        MultiJoinNode toMultiJoinNode()
+        {
+            return new MultiJoinNode(sources, and(filters), outputSymbols);
+        }
+    }
+}