diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/DistributedExecutionPlanner.java b/presto-main/src/main/java/io/prestosql/sql/planner/DistributedExecutionPlanner.java
index a05cc553d..944eab604 100644
--- a/presto-main/src/main/java/io/prestosql/sql/planner/DistributedExecutionPlanner.java
+++ b/presto-main/src/main/java/io/prestosql/sql/planner/DistributedExecutionPlanner.java
@@ -678,7 +678,7 @@ public class DistributedExecutionPlanner
         private final long nextSnapshotId;
         // Which are the corresponding "left" sources from join nodes while visiting the "right" nodes.
         // Value can be SplitSource, ValuesNode, or RemoteSourceNode
-        private final Stack<Object> sourceStack;
+        private Stack<Object> sourceStack;
         private Object leftmostSource;
         private int pendingJoins;
 
@@ -714,10 +714,16 @@ public class DistributedExecutionPlanner
             unionSources.add(sources);
 
             ImmutableMap.Builder<PlanNodeId, SplitSource> result = ImmutableMap.builder();
+            boolean isLeft = true;
             for (PlanNode child : node.getSources()) {
                 // Backup the current left-most source
                 Object prevLeftmost = leftmostSource;
                 leftmostSource = null;
+                // Save our sourceStack and recurse with empty stack if this is not the left child
+                Stack<Object> savedSourceStack = sourceStack;
+                if (!isLeft) {
+                    sourceStack = new Stack<>();
+                }
                 result.putAll(child.accept(this, context));
                 // Collect left-most source of this union member
                 sources.add(leftmostSource);
@@ -725,6 +731,9 @@ public class DistributedExecutionPlanner
                 if (prevLeftmost != null) {
                     leftmostSource = prevLeftmost;
                 }
+                // Recover original stack
+                sourceStack = savedSourceStack;
+                isLeft = false;
             }
 
             return result.build();
diff --git a/presto-main/src/test/java/io/prestosql/sql/planner/TestDistributedExecutionPlanner.java b/presto-main/src/test/java/io/prestosql/sql/planner/TestDistributedExecutionPlanner.java
index c54b6fb08..2bae1a69e 100644
--- a/presto-main/src/test/java/io/prestosql/sql/planner/TestDistributedExecutionPlanner.java
+++ b/presto-main/src/test/java/io/prestosql/sql/planner/TestDistributedExecutionPlanner.java
@@ -285,6 +285,15 @@ public class TestDistributedExecutionPlanner
         assertTrue(wrong.isEmpty(), "Wrong dependency: " + wrong);
     }
 
+    private void testExact(SubPlan root, Multimap<String, String> expected)
+    {
+        planner.plan(root, session, SNAPSHOT, null, 0);
+        Multimap<String, String> missing = Multimaps.filterEntries(expected, e -> !dependencies.containsEntry(e.getKey(), e.getValue()));
+        assertTrue(missing.isEmpty(), "Missing dependency: " + missing);
+        Multimap<String, String> wrong = Multimaps.filterEntries(dependencies, e -> !expected.containsEntry(e.getKey(), e.getValue()));
+        assertTrue(wrong.isEmpty(), "Wrong dependency: " + wrong);
+    }
+
     @Test
     public void testExchangeUnion()
     {
@@ -365,6 +374,43 @@ public class TestDistributedExecutionPlanner
         assertEquals(b.getNextSnapshotId(), 7);
     }
 
+    @Test
+    public void testSimpleExchangeAndJoin()
+    {
+        SubPlan root = makePlan(1,
+                join(
+                        union(source("A"), source("B")),
+                        source("C")
+                ), ImmutableList.of());
+        test(root, ImmutableMultimap.of("A", "C"),
+                ImmutableMultimap.of(
+                        "A", "B",
+                        "B", "C"));
+    }
+
+    @Test
+    public void testComplexExchangeAndJoin()
+    {
+        SubPlan root = makePlan(1,
+                join(
+                        union(ImmutableList.of(
+                                join(source("A"), source("B")),
+                                join(
+                                        union(source("C"), source("D")),
+                                        source("E")),
+                                source("F"))),
+                        union(
+                                join(source("G"), source("H")),
+                                join(source("I"), source("J")))),
+                ImmutableList.of());
+        testExact(root, ImmutableMultimap.of(
+                "A", "B",
+                "A", "G",
+                "C", "E",
+                "G", "H",
+                "I", "J"));
+    }
+
     private SubPlan makePlan(int fragmentId, PlanNode node, List<SubPlan> children)
     {
         PlanFragment fragment = new PlanFragment(
@@ -429,14 +475,19 @@ public class TestDistributedExecutionPlanner
     }
 
     private ExchangeNode union(PlanNode left, PlanNode right)
+    {
+        return union(ImmutableList.of(left, right));
+    }
+
+    private ExchangeNode union(ImmutableList<PlanNode> subPlans)
     {
         return new ExchangeNode(
                 new PlanNodeId(String.valueOf(++nodeId)),
                 GATHER,
                 LOCAL,
                 new PartitioningScheme(Partitioning.create(SINGLE_DISTRIBUTION, ImmutableList.of()), ImmutableList.of(symbol)),
-                ImmutableList.of(left, right),
-                ImmutableList.of(ImmutableList.of(symbol), ImmutableList.of(symbol)),
+                subPlans,
+                subPlans.stream().map(p -> ImmutableList.of(symbol)).collect(ImmutableList.toImmutableList()),
                 Optional.empty(),
                 HASH);
     }