jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!1250 Fix snapshot scheduling bug causing queries to hang 

Merge pull request !1250 from Kevin Wan/hanging-fix
This commit is contained in:
i-robot 2021-11-25 16:12:10 +00:00 committed by Gitee
parent b311335c7e 1afc642499
commit efa2bc3ef5
2 changed files with 63 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
presto-main/src/main/java/io/prestosql/sql/planner/DistributedExecutionPlanner.java
View File

@ -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();

55
presto-main/src/test/java/io/prestosql/sql/planner/TestDistributedExecutionPlanner.java
View File

@ -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);
}