Repository: prestodb/presto
Issue: Query optimizer generates incorrect plans for some logical expressions

Reproduce with:
```
select (a=8 or b=2) and (a=8 or c=3) or (b=9) from (values (1, 2, 3)) as t (a, b, c);
```

`(a=8 or b=2) and (a=8 or c=3)` should become `a=8 or (b=2 and c=3)`. The query optimizer turns it into `a=8 and (b=2 and c=3)`


Changed files:
presto-main/src/main/java/com/facebook/presto/sql/planner/optimizations/SimplifyExpressions.java
presto-main/src/test/java/com/facebook/presto/sql/planner/optimizations/TestSimplifyExpressions.java

diff --git a/presto-main/src/main/java/com/facebook/presto/sql/planner/optimizations/SimplifyExpressions.java b/presto-main/src/main/java/com/facebook/presto/sql/planner/optimizations/SimplifyExpressions.java
index 76c9e65ad1..6d318c880e 100644
--- a/presto-main/src/main/java/com/facebook/presto/sql/planner/optimizations/SimplifyExpressions.java
+++ b/presto-main/src/main/java/com/facebook/presto/sql/planner/optimizations/SimplifyExpressions.java
@@ -59,6 +59,7 @@ import static com.facebook.presto.util.ImmutableCollectors.toImmutableList;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptySet;
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toSet;
 
 public class SimplifyExpressions
@@ -212,11 +213,29 @@ public class SimplifyExpressions
         @Override
         public Expression rewriteLogicalBinaryExpression(LogicalBinaryExpression node, NodeContext context, ExpressionTreeRewriter<NodeContext> treeRewriter)
         {
-            List<Expression> predicates = extractPredicates(node.getType(), node).stream()
-                    .map(expression -> treeRewriter.rewrite(expression, NodeContext.NOT_ROOT_NODE))
-                    .collect(toImmutableList());
+            Expression expression = combinePredicates(
+                    node.getType(),
+                    extractPredicates(node.getType(), node).stream()
+                            .map(subExpression -> treeRewriter.rewrite(subExpression, NodeContext.NOT_ROOT_NODE))
+                            .collect(toImmutableList()));
+
+            if (!(expression instanceof LogicalBinaryExpression)) {
+                return expression;
+            }
+
+            Expression simplified = extractCommonPredicates((LogicalBinaryExpression) expression);
+
+            // Prefer AND LogicalBinaryExpression at the root if possible
+            if (context.isRootNode() && simplified instanceof LogicalBinaryExpression && ((LogicalBinaryExpression) simplified).getType() == OR) {
+                return distributeIfPossible((LogicalBinaryExpression) simplified);
+            }
+
+            return simplified;
+        }
 
-            List<List<Expression>> subPredicates = getSubPredicates(predicates);
+        private static Expression extractCommonPredicates(LogicalBinaryExpression node)
+        {
+            List<List<Expression>> subPredicates = getSubPredicates(node);
 
             Set<Expression> commonPredicates = ImmutableSet.copyOf(subPredicates.stream()
                     .map(ExtractCommonPredicatesExpressionRewriter::filterDeterministicPredicates)
@@ -234,26 +253,63 @@ public class SimplifyExpressions
                     .collect(toImmutableList());
             Expression combinedUncorrelatedPredicates = combinePredicates(node.getType(), uncorrelatedPredicates);
 
-            // Do not simplify top level conjuncts if it would result in top level disjuncts
-            // Conjuncts are easier to process when pushing down predicates.
-            if (context.isRootNode() && flippedNodeType == OR && !combinedUncorrelatedPredicates.equals(FALSE_LITERAL)) {
-                return combinePredicates(node.getType(), predicates);
-            }
-
             return combinePredicates(flippedNodeType, ImmutableList.<Expression>builder()
                     .addAll(commonPredicates)
                     .add(combinedUncorrelatedPredicates)
                     .build());
         }
 
-        private static List<List<Expression>> getSubPredicates(List<Expression> predicates)
+        private static List<List<Expression>> getSubPredicates(LogicalBinaryExpression expression)
         {
-            return predicates.stream()
+            return extractPredicates(expression.getType(), expression).stream()
                     .map(predicate -> predicate instanceof LogicalBinaryExpression ?
                             extractPredicates((LogicalBinaryExpression) predicate) : ImmutableList.of(predicate))
                     .collect(toImmutableList());
         }
 
+        /**
+         * Applies the boolean distributive property.
+         *
+         * For example:
+         * ( A & B ) | ( C & D ) => ( A | C ) & ( A | D ) & ( B | C ) & ( B | D )
+         *
+         * Returns the original expression if the expression is non-deterministic or if the distribution will
+         * expand the expression by too much.
+         */
+        private static Expression distributeIfPossible(LogicalBinaryExpression expression)
+        {
+            if (!DeterminismEvaluator.isDeterministic(expression)) {
+                // Do not distribute boolean expressions if there are any non-deterministic elements
+                // TODO: This can be optimized further if non-deterministic elements are not repeated
+                return expression;
+            }
+            List<Set<Expression>> subPredicates = getSubPredicates(expression).stream()
+                    .map(ImmutableSet::copyOf)
+                    .collect(toList());
+
+            int originalBaseExpressions = subPredicates.stream()
+                    .mapToInt(Set::size)
+                    .sum();
+            int newBaseExpressions = subPredicates.stream()
+                    .mapToInt(Set::size)
+                    .reduce(Math::multiplyExact)
+                    .getAsInt() * subPredicates.size();
+            if (newBaseExpressions > originalBaseExpressions * 2) {
+                // Do not distribute boolean expressions if it would create 2x more base expressions
+                // (e.g. A, B, C, D from the above example). This is just an arbitrary heuristic to
+                // avoid cross product expression explosion.
+                return expression;
+            }
+
+            Set<List<Expression>> crossProduct = Sets.cartesianProduct(subPredicates);
+
+            return combinePredicates(
+                    expression.getType().flip(),
+                    crossProduct.stream()
+                            .map(expressions -> combinePredicates(expression.getType(), expressions))
+                            .collect(toImmutableList()));
+        }
+
         private static Set<Expression> filterDeterministicPredicates(List<Expression> predicates)
         {
             return predicates.stream()
diff --git a/presto-main/src/test/java/com/facebook/presto/sql/planner/optimizations/TestSimplifyExpressions.java b/presto-main/src/test/java/com/facebook/presto/sql/planner/optimizations/TestSimplifyExpressions.java
index 07a68a85b7..c647f36e10 100644
--- a/presto-main/src/test/java/com/facebook/presto/sql/planner/optimizations/TestSimplifyExpressions.java
+++ b/presto-main/src/test/java/com/facebook/presto/sql/planner/optimizations/TestSimplifyExpressions.java
@@ -65,8 +65,10 @@ public class TestSimplifyExpressions
     }
 
     @Test
-    public void testExtractsCommonPredicate()
+    public void testExtractCommonPredicates()
     {
+        assertSimplifies("X AND Y", "X AND Y");
+        assertSimplifies("X OR Y", "X OR Y");
         assertSimplifies("X AND X", "X");
         assertSimplifies("X OR X", "X");
         assertSimplifies("(X OR Y) AND (X OR Y)", "X OR Y");
@@ -83,9 +85,18 @@ public class TestSimplifyExpressions
         assertSimplifies("((X OR V) AND V) OR ((X OR V) AND V)", "V");
         assertSimplifies("((X OR V) AND X) OR ((X OR V) AND V)", "X OR V");
 
-        assertSimplifies("((X OR V) AND Z) OR ((X OR V) AND V)", "(((X OR V) AND Z) OR V)");
+        assertSimplifies("((X OR V) AND Z) OR ((X OR V) AND V)", "(X OR V) AND (Z OR V)");
         assertSimplifies("X AND ((Y AND Z) OR (Y AND V) OR (Y AND X))", "X AND Y AND (Z OR V OR X)");
         assertSimplifies("(A AND B AND C AND D) OR (A AND B AND E) OR (A AND F)", "A AND ((B AND C AND D) OR (B AND E) OR F)");
+
+        assertSimplifies("((A AND B) OR (A AND C)) AND D", "A AND (B OR C) AND D");
+        assertSimplifies("((A OR B) AND (A OR C)) OR D", "(A OR B OR D) AND (A OR C OR D)");
+        assertSimplifies("(((A AND B) OR (A AND C)) AND D) OR E", "(A OR E) AND (B OR C OR E) AND (D OR E)");
+        assertSimplifies("(((A OR B) AND (A OR C)) OR D) AND E", "(A OR (B AND C) OR D) AND E");
+
+        assertSimplifies("(A AND B) OR (C AND D)", "(A OR C) AND (A OR D) AND (B OR C) AND (B OR D)");
+        // No distribution since it would add too many new terms
+        assertSimplifies("(A AND B) OR (C AND D) OR (E AND F)", "(A AND B) OR (C AND D) OR (E AND F)");
     }
 
     private static void assertSimplifies(String expression, String expected)