Repository: prestodb/presto
Issue: Potential deadlock when using phased scheduler with broadcast joins

``` sql
SELECT * FROM orders CROSS JOIN (VALUES 1);
```

```
 Fragment 0 [SINGLE]                                                                                                                                                                                                                
     Output layout: [orderkey, custkey, orderstatus, totalprice, orderdate, orderpriority, clerk, shippriority, comment, field]                                                                                                     
     Output partitioning: SINGLE []                                                                                                                                                                                                 
     - Output[orderkey, custkey, orderstatus, totalprice, orderdate, orderpriority, clerk, shippriority, comment, _col9] => [orderkey:bigint, custkey:bigint, orderstatus:varchar, totalprice:double, orderdate:date, orderpriority:
             _col9 := field                                                                                                                                                                                                         
         - RemoteSource[1] => [orderkey:bigint, custkey:bigint, orderstatus:varchar, totalprice:double, orderdate:date, orderpriority:varchar, clerk:varchar, shippriority:bigint, comment:varchar, field:bigint]                   

 Fragment 1 [SOURCE]                                                                                                                                                                                                                
     Output layout: [orderkey, custkey, orderstatus, totalprice, orderdate, orderpriority, clerk, shippriority, comment, field]                                                                                                     
     Output partitioning: SINGLE []                                                                                                                                                                                                 
     - InnerJoin[] => [orderkey:bigint, custkey:bigint, orderstatus:varchar, totalprice:double, orderdate:date, orderpriority:varchar, clerk:varchar, shippriority:bigint, comment:varchar, field:bigint]                           
         - TableScan[tpch:tpch:orders:sf1.0, originalConstraint = true] => [orderkey:bigint, custkey:bigint, orderstatus:varchar, totalprice:double, orderdate:date, orderpriority:varchar, clerk:varchar, shippriority:bigint, comm
                 orderkey := tpch:orderkey                                                                                                                                                                                          
                 custkey := tpch:custkey                                                                                                                                                                                            
                 orderstatus := tpch:orderstatus                                                                                                                                                                                    
                 totalprice := tpch:totalprice                                                                                                                                                                                      
                 orderdate := tpch:orderdate                                                                                                                                                                                        
                 orderpriority := tpch:orderpriority                                                                                                                                                                                
                 clerk := tpch:clerk                                                                                                                                                                                                
                 shippriority := tpch:shippriority                                                                                                                                                                                  
                 comment := tpch:comment                                                                                                                                                                                            
         - RemoteSource[2] => [field:bigint]                                                                                                                                                                                        

 Fragment 2 [SINGLE]                                                                                                                                                                                                                
     Output layout: [field]                                                                                                                                                                                                         
     Output partitioning: BROADCAST []                                                                                                                                                                                              
     - Values => [field:bigint]                                                                                                                                                                                                     
             (1)   
```

The scheduling dependency graph is missing an edge (2->1), which can cause the join stage to be scheduled independently (and before) the build stage.


Changed files:
presto-main/src/main/java/com/facebook/presto/execution/scheduler/PhasedExecutionSchedule.java
presto-main/src/test/java/com/facebook/presto/execution/scheduler/TestPhasedExecutionSchedule.java

diff --git a/presto-main/src/main/java/com/facebook/presto/execution/scheduler/PhasedExecutionSchedule.java b/presto-main/src/main/java/com/facebook/presto/execution/scheduler/PhasedExecutionSchedule.java
index 29c622be83..1c3f4c1df4 100644
--- a/presto-main/src/main/java/com/facebook/presto/execution/scheduler/PhasedExecutionSchedule.java
+++ b/presto-main/src/main/java/com/facebook/presto/execution/scheduler/PhasedExecutionSchedule.java
@@ -153,7 +153,12 @@ public class PhasedExecutionSchedule
         for (DefaultEdge edge : graph.edgeSet()) {
             PlanFragmentId source = graph.getEdgeSource(edge);
             PlanFragmentId target = graph.getEdgeTarget(edge);
-            componentGraph.addEdge(componentMembership.get(source), componentMembership.get(target));
+
+            Set<PlanFragmentId> from = componentMembership.get(source);
+            Set<PlanFragmentId> to = componentMembership.get(target);
+            if (!from.equals(to)) { // the topological order iterator below doesn't include vertices that have self-edges, so don't add them
+                componentGraph.addEdge(from, to);
+            }
         }
 
         List<Set<PlanFragmentId>> schedulePhases = ImmutableList.copyOf(new TopologicalOrderIterator<>(componentGraph));
@@ -274,7 +279,7 @@ public class PhasedExecutionSchedule
         {
             List<PlanNode> sources = node.getSources();
             if (sources.isEmpty()) {
-                return ImmutableSet.of();
+                return ImmutableSet.of(currentFragmentId);
             }
             if (sources.size() == 1) {
                 return sources.get(0).accept(this, currentFragmentId);
diff --git a/presto-main/src/test/java/com/facebook/presto/execution/scheduler/TestPhasedExecutionSchedule.java b/presto-main/src/test/java/com/facebook/presto/execution/scheduler/TestPhasedExecutionSchedule.java
index 45ec18c8ba..97ab602569 100644
--- a/presto-main/src/test/java/com/facebook/presto/execution/scheduler/TestPhasedExecutionSchedule.java
+++ b/presto-main/src/test/java/com/facebook/presto/execution/scheduler/TestPhasedExecutionSchedule.java
@@ -94,6 +94,17 @@ public class TestPhasedExecutionSchedule
         assertEquals(phases, ImmutableList.of(ImmutableSet.of(joinFragment.getId()), ImmutableSet.of(buildFragment.getId()), ImmutableSet.of(probeFragment.getId())));
     }
 
+    @Test
+    public void testBroadcastJoin()
+            throws Exception
+    {
+        PlanFragment buildFragment = createTableScanPlanFragment("build");
+        PlanFragment joinFragment = createBroadcastJoinPlanFragment("join", buildFragment);
+
+        List<Set<PlanFragmentId>> phases = PhasedExecutionSchedule.extractPhases(ImmutableList.of(joinFragment, buildFragment));
+        assertEquals(phases, ImmutableList.of(ImmutableSet.of(joinFragment.getId(), buildFragment.getId())));
+    }
+
     @Test
     public void testJoinWithDeepSources()
             throws Exception
@@ -150,6 +161,30 @@ public class TestPhasedExecutionSchedule
         return createFragment(planNode);
     }
 
+    private static PlanFragment createBroadcastJoinPlanFragment(String name, PlanFragment buildFragment)
+    {
+        Symbol symbol = new Symbol("column");
+        PlanNode tableScan = new TableScanNode(
+                new PlanNodeId(name),
+                new TableHandle("test", new TestingTableHandle()),
+                ImmutableList.of(symbol),
+                ImmutableMap.of(symbol, new TestingColumnHandle("column")),
+                Optional.empty(),
+                TupleDomain.all(),
+                null);
+
+        PlanNode join = new JoinNode(
+                new PlanNodeId(name + "_id"),
+                INNER,
+                tableScan,
+                new RemoteSourceNode(new PlanNodeId("build_id"), buildFragment.getId(), ImmutableList.of()),
+                ImmutableList.of(),
+                Optional.<Symbol>empty(),
+                Optional.<Symbol>empty());
+
+        return createFragment(join);
+    }
+
     private static PlanFragment createJoinPlanFragment(String name, PlanFragment buildFragment, PlanFragment probeFragment)
     {
         PlanNode planNode = new JoinNode(