You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by sm...@apache.org on 2014/10/29 02:12:31 UTC
[2/2] git commit: DRILL-1595: Send intermediate fragments before leaf
fragments
DRILL-1595: Send intermediate fragments before leaf fragments
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/fdcedf97
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/fdcedf97
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/fdcedf97
Branch: refs/heads/master
Commit: fdcedf972d30a5d9fc3e1bcaa364d9634235b82b
Parents: 81e9325
Author: Steven Phillips <sp...@maprtech.com>
Authored: Mon Oct 27 23:29:18 2014 -0700
Committer: Steven Phillips <sp...@maprtech.com>
Committed: Tue Oct 28 16:40:55 2014 -0700
----------------------------------------------------------------------
.../drill/exec/work/foreman/QueryManager.java | 45 ++++++++++++++++----
1 file changed, 36 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/fdcedf97/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java
index 61f3d82..b60d8a9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java
@@ -23,6 +23,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.drill.common.exceptions.ExecutionSetupException;
@@ -121,20 +122,37 @@ public class QueryManager implements FragmentStatusListener{
}
}
- Multimap<DrillbitEndpoint, PlanFragment> fragmentMap = ArrayListMultimap.create();
+ Multimap<DrillbitEndpoint, PlanFragment> leafFragmentMap = ArrayListMultimap.create();
+ Multimap<DrillbitEndpoint, PlanFragment> intFragmentMap = ArrayListMultimap.create();
// record all fragments for status purposes.
for (PlanFragment f : nonRootFragments) {
logger.debug("Tracking intermediate remote node {} with data {}", f.getAssignment(), f.getFragmentJson());
status.add(new FragmentData(f.getHandle(), f.getAssignment(), false));
- fragmentMap.put(f.getAssignment(), f);
+ if (f.getLeafFragment()) {
+ leafFragmentMap.put(f.getAssignment(), f);
+ } else {
+ intFragmentMap.put(f.getAssignment(), f);
+ }
}
+ CountDownLatch latch = new CountDownLatch(intFragmentMap.keySet().size());
+
+ // send remote intermediate fragments
+ for (DrillbitEndpoint ep : intFragmentMap.keySet()) {
+ sendRemoteFragments(ep, intFragmentMap.get(ep), latch);
+ }
+ // wait for send complete
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ throw new ExecutionSetupException(e);
+ }
// send remote (leaf) fragments.
- for (DrillbitEndpoint ep : fragmentMap.keySet()) {
- sendRemoteFragments(ep, fragmentMap.get(ep));
+ for (DrillbitEndpoint ep : leafFragmentMap.keySet()) {
+ sendRemoteFragments(ep, leafFragmentMap.get(ep), null);
}
bee.getContext().getAllocator().resetFragmentLimits();
@@ -146,7 +164,7 @@ public class QueryManager implements FragmentStatusListener{
}
}
- private void sendRemoteFragments(DrillbitEndpoint assignment, Collection<PlanFragment> fragments){
+ private void sendRemoteFragments(DrillbitEndpoint assignment, Collection<PlanFragment> fragments, CountDownLatch latch){
InitializeFragments.Builder fb = InitializeFragments.newBuilder();
for(PlanFragment f : fragments){
fb.addFragment(f);
@@ -154,7 +172,7 @@ public class QueryManager implements FragmentStatusListener{
InitializeFragments initFrags = fb.build();
logger.debug("Sending remote fragments to node {} with data {}", assignment, initFrags);
- FragmentSubmitListener listener = new FragmentSubmitListener(assignment, initFrags);
+ FragmentSubmitListener listener = new FragmentSubmitListener(assignment, initFrags, latch);
controller.getTunnel(assignment).sendFragments(listener, initFrags);
}
@@ -269,16 +287,25 @@ public class QueryManager implements FragmentStatusListener{
}
- public RpcOutcomeListener<Ack> getSubmitListener(DrillbitEndpoint endpoint, InitializeFragments value){
- return new FragmentSubmitListener(endpoint, value);
+ public RpcOutcomeListener<Ack> getSubmitListener(DrillbitEndpoint endpoint, InitializeFragments value, CountDownLatch latch){
+ return new FragmentSubmitListener(endpoint, value, latch);
}
private class FragmentSubmitListener extends EndpointListener<Ack, InitializeFragments>{
- public FragmentSubmitListener(DrillbitEndpoint endpoint, InitializeFragments value) {
+ private CountDownLatch latch;
+
+ public FragmentSubmitListener(DrillbitEndpoint endpoint, InitializeFragments value, CountDownLatch latch) {
super(endpoint, value);
+ this.latch = latch;
}
+ @Override
+ public void success(Ack ack, ByteBuf byteBuf) {
+ if (latch != null) {
+ latch.countDown();
+ }
+ }
@Override
public void failed(RpcException ex) {