[23/52] [abbrv] [partial] DRILL-165: Reorganize directories (moves only)

[ja...@apache.org]

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java
new file mode 100644
index 0000000..de8d31c
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java
@@ -0,0 +1,169 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.drill.exec.cache.DistributedCache;
+import org.apache.drill.exec.coord.ClusterCoordinator;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.NamedThreadFactory;
+import org.apache.drill.exec.rpc.bit.BitCom;
+import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.work.batch.BitComHandler;
+import org.apache.drill.exec.work.batch.BitComHandlerImpl;
+import org.apache.drill.exec.work.foreman.Foreman;
+import org.apache.drill.exec.work.fragment.IncomingFragmentHandler;
+import org.apache.drill.exec.work.user.UserWorker;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
+
+public class WorkManager implements Closeable{
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(WorkManager.class);
+  
+  private Set<IncomingFragmentHandler> incomingFragments = Collections.newSetFromMap(Maps.<IncomingFragmentHandler, Boolean> newConcurrentMap());
+
+  private PriorityBlockingQueue<Runnable> pendingTasks = Queues.newPriorityBlockingQueue();
+  
+  private Map<FragmentHandle, FragmentRunner> runningFragments = Maps.newConcurrentMap();
+  
+  private ConcurrentMap<QueryId, Foreman> queries = Maps.newConcurrentMap();
+
+  private BootStrapContext bContext;
+  private DrillbitContext dContext;
+
+  private final BitComHandler bitComWorker;
+  private final UserWorker userWorker;
+  private final WorkerBee bee;
+  private Executor executor = Executors.newFixedThreadPool(4, new NamedThreadFactory("WorkManager-"));
+  private final EventThread eventThread;
+  
+  public WorkManager(BootStrapContext context){
+    this.bee = new WorkerBee();
+    this.bContext = context;
+    this.bitComWorker = new BitComHandlerImpl(bee);
+    this.userWorker = new UserWorker(bee);
+    this.eventThread = new EventThread();
+    
+  }
+  
+  public void start(DrillbitEndpoint endpoint, DistributedCache cache, BitCom com, ClusterCoordinator coord){
+    this.dContext = new DrillbitContext(endpoint, bContext, coord, com, cache);
+    eventThread.start();
+  }
+  
+  public BitComHandler getBitComWorker(){
+    return bitComWorker;
+  }
+
+  public UserWorker getUserWorker(){
+    return userWorker;
+  }
+  
+  @Override
+  public void close() throws IOException {
+  }
+  
+
+  public DrillbitContext getContext() {
+    return dContext;
+  }
+
+  // create this so items can see the data here whether or not they are in this package.
+  public class WorkerBee{
+
+    public void addFragmentRunner(FragmentRunner runner){
+      pendingTasks.add(runner);
+    }
+    
+    public void addNewForeman(Foreman foreman){
+      pendingTasks.add(foreman);
+    }
+
+
+    public void addFragmentPendingRemote(IncomingFragmentHandler handler){
+      incomingFragments.add(handler);
+    }
+    
+    public void startFragmentPendingRemote(IncomingFragmentHandler handler){
+      incomingFragments.remove(handler);
+      pendingTasks.add(handler.getRunnable());
+    }
+    
+    public FragmentRunner getFragmentRunner(FragmentHandle handle){
+      return runningFragments.get(handle);
+    }
+    
+    public Foreman getForemanForQueryId(QueryId queryId){
+      return queries.get(queryId);
+    }
+    
+    public void retireForeman(Foreman foreman){
+      queries.remove(foreman.getQueryId(), foreman);
+    }
+
+    public DrillbitContext getContext() {
+      return dContext;
+    }
+
+  }
+
+
+
+ private class EventThread extends Thread{
+   public EventThread(){
+     this.setDaemon(true);
+     this.setName("WorkManager Event Thread");
+   }
+
+  @Override
+  public void run() {
+    try {
+    while(true){
+//      logger.debug("Polling for pending work tasks.");
+      Runnable r = pendingTasks.poll(10, TimeUnit.SECONDS);
+      if(r != null){
+        logger.debug("Starting pending task {}", r);
+        executor.execute(r);  
+      }
+      
+    }
+    } catch (InterruptedException e) {
+      logger.info("Work Manager stopping as it was interrupted.");
+    }
+  }
+   
+   
+ }
+
+  
+  
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractFragmentCollector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractFragmentCollector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractFragmentCollector.java
new file mode 100644
index 0000000..9e420c1
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractFragmentCollector.java
@@ -0,0 +1,94 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicIntegerArray;
+
+import org.apache.drill.exec.physical.base.Receiver;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle;
+
+import com.google.common.base.Preconditions;
+
+public abstract class AbstractFragmentCollector implements BatchCollector{
+
+  private final List<DrillbitEndpoint> incoming;
+  private final int oppositeMajorFragmentId;
+  private final AtomicIntegerArray remainders;
+  private final AtomicInteger remainingRequired;
+  protected final RawBatchBuffer[] buffers;
+  private final AtomicInteger parentAccounter;
+  private final AtomicInteger finishedStreams = new AtomicInteger();
+  
+  public AbstractFragmentCollector(AtomicInteger parentAccounter, Receiver receiver, int minInputsRequired) {
+    Preconditions.checkArgument(minInputsRequired > 0);
+    Preconditions.checkNotNull(receiver);
+    Preconditions.checkNotNull(parentAccounter);
+
+    this.parentAccounter = parentAccounter;
+    this.incoming = receiver.getProvidingEndpoints();
+    this.remainders = new AtomicIntegerArray(incoming.size());
+    this.oppositeMajorFragmentId = receiver.getOppositeMajorFragmentId();
+    this.buffers = new RawBatchBuffer[minInputsRequired];
+    for(int i = 0; i < buffers.length; i++){
+      buffers[i] = new UnlimitedRawBatchBuffer();
+    }
+    if (receiver.supportsOutOfOrderExchange()) {
+      this.remainingRequired = new AtomicInteger(1);
+    } else {
+      this.remainingRequired = new AtomicInteger(minInputsRequired);
+    }
+  }
+
+  public int getOppositeMajorFragmentId() {
+    return oppositeMajorFragmentId;
+  }
+
+  public RawBatchBuffer[] getBuffers(){
+    return buffers;
+  }
+  
+  public abstract void streamFinished(int minorFragmentId);
+  
+  public boolean batchArrived(ConnectionThrottle throttle, int minorFragmentId, RawFragmentBatch batch) {
+    boolean decremented = false;
+    if (remainders.compareAndSet(minorFragmentId, 0, 1)) {
+      int rem = remainingRequired.decrementAndGet();
+      if (rem == 0) {
+        parentAccounter.decrementAndGet();
+        decremented = true;
+      }
+    }
+    if(batch.getHeader().getIsLastBatch()){
+      streamFinished(minorFragmentId);
+    }
+    getBuffer(minorFragmentId).enqueue(throttle, batch);
+    return decremented;
+  }
+
+
+  @Override
+  public int getTotalIncomingFragments() {
+    return incoming.size();
+  }
+
+  protected abstract RawBatchBuffer getBuffer(int minorFragmentId);
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BatchCollector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BatchCollector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BatchCollector.java
new file mode 100644
index 0000000..89a889c
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BatchCollector.java
@@ -0,0 +1,31 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle;
+
+
+interface BatchCollector {
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BatchCollector.class);
+  public boolean batchArrived(ConnectionThrottle throttle, int minorFragmentId, RawFragmentBatch batch);
+  public int getOppositeMajorFragmentId();
+  public RawBatchBuffer[] getBuffers();
+  public int getTotalIncomingFragments();
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BitComHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BitComHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BitComHandler.java
new file mode 100644
index 0000000..97064e3
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BitComHandler.java
@@ -0,0 +1,41 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.PlanFragment;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.bit.BitConnection;
+import org.apache.drill.exec.work.fragment.IncomingFragmentHandler;
+
+public interface BitComHandler {
+
+  public abstract Response handle(BitConnection connection, int rpcType, ByteBuf pBody, ByteBuf dBody)
+      throws RpcException;
+
+  public abstract void startNewRemoteFragment(PlanFragment fragment);
+
+  public abstract Ack cancelFragment(FragmentHandle handle);
+
+  public abstract void registerIncomingFragmentHandler(IncomingFragmentHandler handler);
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BitComHandlerImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BitComHandlerImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BitComHandlerImpl.java
new file mode 100644
index 0000000..8cba493
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/BitComHandlerImpl.java
@@ -0,0 +1,211 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+import static org.apache.drill.exec.rpc.RpcBus.get;
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.impl.ImplCreator;
+import org.apache.drill.exec.physical.impl.RootExec;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.FragmentRecordBatch;
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus;
+import org.apache.drill.exec.proto.ExecProtos.PlanFragment;
+import org.apache.drill.exec.proto.ExecProtos.RpcType;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.Acks;
+import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.RpcConstants;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.bit.BitConnection;
+import org.apache.drill.exec.rpc.bit.BitRpcConfig;
+import org.apache.drill.exec.rpc.bit.BitTunnel;
+import org.apache.drill.exec.work.FragmentRunner;
+import org.apache.drill.exec.work.RemoteFragmentRunnerListener;
+import org.apache.drill.exec.work.WorkManager.WorkerBee;
+import org.apache.drill.exec.work.fragment.IncomingFragmentHandler;
+import org.apache.drill.exec.work.fragment.RemoteFragmentHandler;
+
+import com.google.common.collect.Maps;
+
+public class BitComHandlerImpl implements BitComHandler {
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitComHandlerImpl.class);
+  
+  private ConcurrentMap<FragmentHandle, IncomingFragmentHandler> handlers = Maps.newConcurrentMap();
+  private final WorkerBee bee;
+  
+  public BitComHandlerImpl(WorkerBee bee) {
+    super();
+    this.bee = bee;
+  }
+
+  /* (non-Javadoc)
+   * @see org.apache.drill.exec.work.batch.BitComHandler#handle(org.apache.drill.exec.rpc.bit.BitConnection, int, io.netty.buffer.ByteBuf, io.netty.buffer.ByteBuf)
+   */
+  @Override
+  public Response handle(BitConnection connection, int rpcType, ByteBuf pBody, ByteBuf dBody) throws RpcException {
+    if(RpcConstants.EXTRA_DEBUGGING) logger.debug("Received bit com message of type {}", rpcType);
+
+    switch (rpcType) {
+    
+    case RpcType.REQ_CANCEL_FRAGMENT_VALUE:
+      FragmentHandle handle = get(pBody, FragmentHandle.PARSER);
+      cancelFragment(handle);
+      return BitRpcConfig.OK;
+
+    case RpcType.REQ_FRAGMENT_STATUS_VALUE:
+      connection.getListenerPool().status( get(pBody, FragmentStatus.PARSER));
+      // TODO: Support a type of message that has no response.
+      return BitRpcConfig.OK;
+
+    case RpcType.REQ_INIATILIZE_FRAGMENT_VALUE:
+      PlanFragment fragment = get(pBody, PlanFragment.PARSER);
+      startNewRemoteFragment(fragment);
+      return BitRpcConfig.OK;
+      
+    case RpcType.REQ_RECORD_BATCH_VALUE:
+      try {
+        FragmentRecordBatch header = get(pBody, FragmentRecordBatch.PARSER);
+        incomingRecordBatch(connection, header, dBody);
+        return BitRpcConfig.OK;
+      } catch (FragmentSetupException e) {
+        throw new RpcException("Failure receiving record batch.", e);
+      }
+
+    default:
+      throw new RpcException("Not yet supported.");
+    }
+
+  }
+  
+  
+  
+  /* (non-Javadoc)
+   * @see org.apache.drill.exec.work.batch.BitComHandler#startNewRemoteFragment(org.apache.drill.exec.proto.ExecProtos.PlanFragment)
+   */
+  @Override
+  public void startNewRemoteFragment(PlanFragment fragment){
+    logger.debug("Received remote fragment start instruction", fragment);
+    FragmentContext context = new FragmentContext(bee.getContext(), fragment.getHandle(), null, null,new FunctionImplementationRegistry(bee.getContext().getConfig()));
+    BitTunnel tunnel = bee.getContext().getBitCom().getTunnel(fragment.getForeman());
+    RemoteFragmentRunnerListener listener = new RemoteFragmentRunnerListener(context, tunnel);
+    try{
+      FragmentRoot rootOperator = bee.getContext().getPlanReader().readFragmentOperator(fragment.getFragmentJson());
+      RootExec exec = ImplCreator.getExec(context, rootOperator);
+      FragmentRunner fr = new FragmentRunner(context, exec, listener);
+      bee.addFragmentRunner(fr);
+
+    }catch(IOException e){
+      listener.fail(fragment.getHandle(), "Failure while parsing fragment execution plan.", e);
+    }catch(ExecutionSetupException e){
+      listener.fail(fragment.getHandle(), "Failure while setting up execution plan.", e);
+    } catch (Exception e) {
+      listener.fail(fragment.getHandle(), "Failure due to uncaught exception", e);
+    } catch (OutOfMemoryError t) {
+      if(t.getMessage().startsWith("Direct buffer")){
+        listener.fail(fragment.getHandle(), "Failure due to error", t);  
+      }else{
+        throw t;
+      }
+      
+    }
+    
+  }
+  
+  /* (non-Javadoc)
+   * @see org.apache.drill.exec.work.batch.BitComHandler#cancelFragment(org.apache.drill.exec.proto.ExecProtos.FragmentHandle)
+   */
+  @Override
+  public Ack cancelFragment(FragmentHandle handle){
+    IncomingFragmentHandler handler = handlers.get(handle);
+    if(handler != null){
+      // try remote fragment cancel.
+      handler.cancel();
+    }else{
+      // then try local cancel.
+      FragmentRunner runner = bee.getFragmentRunner(handle);
+      if(runner != null) runner.cancel();
+    }
+    
+    return Acks.OK;
+  }
+  
+  
+  /**
+   * Returns a positive Ack if this fragment is accepted.  
+   */
+  private Ack incomingRecordBatch(RemoteConnection connection, FragmentRecordBatch fragmentBatch, ByteBuf body) throws FragmentSetupException{
+    FragmentHandle handle = fragmentBatch.getHandle();
+    IncomingFragmentHandler handler = handlers.get(handle);
+
+    // Create a handler if there isn't already one.
+    if(handler == null){
+      
+      PlanFragment fragment = bee.getContext().getCache().getFragment(handle);
+      if(fragment == null){
+        logger.error("Received batch where fragment was not in cache.");
+        return Acks.FAIL;
+      }
+
+      IncomingFragmentHandler newHandler = new RemoteFragmentHandler(fragment, bee.getContext(), bee.getContext().getBitCom().getTunnel(fragment.getForeman()));
+      
+      // since their could be a race condition on the check, we'll use putIfAbsent so we don't have two competing handlers.
+      handler = handlers.putIfAbsent(fragment.getHandle(), newHandler);
+          
+      if(handler == null){
+        // we added a handler, inform the bee that we did so.  This way, the foreman can track status. 
+        bee.addFragmentPendingRemote(newHandler);
+        handler = newHandler;
+      }
+    }
+    
+    boolean canRun = handler.handle(connection.getConnectionThrottle(), new RawFragmentBatch(fragmentBatch, body));
+    if(canRun){
+      logger.debug("Arriving batch means local batch can run, starting local batch.");
+      // if we've reached the canRun threshold, we'll proceed.  This expects handler.handle() to only return a single true.
+      bee.startFragmentPendingRemote(handler);
+    }
+    if(fragmentBatch.getIsLastBatch() && !handler.isWaiting()){
+      logger.debug("Removing handler.  Is Last Batch {}.  Is Waiting for more {}", fragmentBatch.getIsLastBatch(), handler.isWaiting());
+      handlers.remove(handler.getHandle());
+    }
+    
+    return Acks.OK;
+  }
+  
+  /* (non-Javadoc)
+   * @see org.apache.drill.exec.work.batch.BitComHandler#registerIncomingFragmentHandler(org.apache.drill.exec.work.fragment.IncomingFragmentHandler)
+   */
+  @Override
+  public void registerIncomingFragmentHandler(IncomingFragmentHandler handler){
+    IncomingFragmentHandler old = handlers.putIfAbsent(handler.getHandle(), handler);
+    assert old == null : "You can only register a fragment handler if one hasn't been registered already.";
+  }
+  
+  
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java
new file mode 100644
index 0000000..7f14069
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java
@@ -0,0 +1,122 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Receiver;
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+/**
+ * Determines when a particular fragment has enough data for each of its receiving exchanges to commence execution.  Also monitors whether we've collected all incoming data.
+ */
+public class IncomingBuffers {
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(IncomingBuffers.class);
+
+  private final AtomicInteger streamsRemaining = new AtomicInteger(0);
+  private final AtomicInteger remainingRequired = new AtomicInteger(0);
+  private final Map<Integer, BatchCollector> fragCounts;
+
+  public IncomingBuffers(PhysicalOperator root) {
+    Map<Integer, BatchCollector> counts = Maps.newHashMap();
+    CountRequiredFragments reqFrags = new CountRequiredFragments();
+    root.accept(reqFrags, counts);
+    
+    logger.debug("Came up with a list of {} required fragments.  Fragments {}", remainingRequired.get(), counts);
+    fragCounts = ImmutableMap.copyOf(counts);
+
+    // Determine the total number of incoming streams that will need to be completed before we are finished.
+    int totalStreams = 0;
+    for(BatchCollector bc : fragCounts.values()){
+      totalStreams += bc.getTotalIncomingFragments();
+    }
+    assert totalStreams >= remainingRequired.get() : String.format("Total Streams %d should be more than the minimum number of streams to commence (%d).  It isn't.", totalStreams, remainingRequired.get());
+    streamsRemaining.set(totalStreams);
+  }
+
+  public boolean batchArrived(ConnectionThrottle throttle, RawFragmentBatch batch) throws FragmentSetupException {
+    // no need to do anything if we've already enabled running.
+    logger.debug("New Batch Arrived {}", batch);
+    if(batch.getHeader().getIsLastBatch()){
+      streamsRemaining.decrementAndGet();
+    }
+    int sendMajorFragmentId = batch.getHeader().getSendingMajorFragmentId();
+    BatchCollector fSet = fragCounts.get(sendMajorFragmentId);
+    if (fSet == null) throw new FragmentSetupException(String.format("We received a major fragment id that we were not expecting.  The id was %d.", sendMajorFragmentId));
+    boolean decremented = fSet.batchArrived(throttle, batch.getHeader().getSendingMinorFragmentId(), batch);
+    
+    // we should only return true if remaining required has been decremented and is currently equal to zero.
+    return decremented && remainingRequired.get() == 0;
+  }
+
+  public int getRemainingRequired() {
+    int rem = remainingRequired.get();
+    if (rem < 0) return 0;
+    return rem;
+  }
+
+  public RawBatchBuffer[] getBuffers(int senderMajorFragmentId){
+    return fragCounts.get(senderMajorFragmentId).getBuffers();
+  }
+  
+  
+  /**
+   * Designed to setup initial values for arriving fragment accounting.
+   */
+  public class CountRequiredFragments extends AbstractPhysicalVisitor<Void, Map<Integer, BatchCollector>, RuntimeException> {
+    
+    @Override
+    public Void visitReceiver(Receiver receiver, Map<Integer, BatchCollector> counts) throws RuntimeException {
+      BatchCollector set;
+      if (receiver.supportsOutOfOrderExchange()) {
+        set = new MergingCollector(remainingRequired, receiver);
+      } else {
+        set = new PartitionedCollector(remainingRequired, receiver);
+      }
+      
+      counts.put(set.getOppositeMajorFragmentId(), set);
+      remainingRequired.incrementAndGet();
+      return null;
+    }
+
+    
+    @Override
+    public Void visitOp(PhysicalOperator op, Map<Integer, BatchCollector> value) throws RuntimeException {
+      for(PhysicalOperator o : op){
+        o.accept(this, value);
+      }
+      return null;
+    }
+
+
+  }
+
+  public boolean isDone(){
+    return streamsRemaining.get() < 1;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/MergingCollector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/MergingCollector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/MergingCollector.java
new file mode 100644
index 0000000..93868a7
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/MergingCollector.java
@@ -0,0 +1,46 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.drill.exec.physical.base.Receiver;
+
+public class MergingCollector extends AbstractFragmentCollector{
+
+  private AtomicInteger streamsRunning;
+  
+  public MergingCollector(AtomicInteger parentAccounter, Receiver receiver) {
+    super(parentAccounter, receiver, 1);
+    streamsRunning = new AtomicInteger(receiver.getProvidingEndpoints().size());
+  }
+
+  @Override
+  protected RawBatchBuffer getBuffer(int minorFragmentId) {
+    return buffers[0];
+  }
+
+  
+  public void streamFinished(int minorFragmentId) {
+    if(streamsRunning.decrementAndGet() == 0) buffers[0].finished();
+  }
+
+  
+  
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/PartitionedCollector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/PartitionedCollector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/PartitionedCollector.java
new file mode 100644
index 0000000..25b5884
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/PartitionedCollector.java
@@ -0,0 +1,43 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.drill.exec.physical.base.Receiver;
+
+public class PartitionedCollector extends AbstractFragmentCollector{
+
+  public PartitionedCollector(AtomicInteger parentAccounter, Receiver receiver) {
+    super(parentAccounter, receiver, receiver.getProvidingEndpoints().size());
+  }
+
+  @Override
+  protected RawBatchBuffer getBuffer(int minorFragmentId) {
+    return buffers[minorFragmentId];
+  }
+
+  @Override
+  public void streamFinished(int minorFragmentId) {
+    buffers[minorFragmentId].finished();
+  }
+
+  
+  
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java
new file mode 100644
index 0000000..0f10e26
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java
@@ -0,0 +1,33 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.record.RawFragmentBatchProvider;
+import org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle;
+
+public interface RawBatchBuffer extends RawFragmentBatchProvider{
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RawBatchBuffer.class);
+  
+  public void enqueue(ConnectionThrottle throttle, RawFragmentBatch batch);
+  
+  /**
+   * Inform the buffer that no more records are expected.
+   */
+  public void finished();
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java
new file mode 100644
index 0000000..faff6c7
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java
@@ -0,0 +1,73 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.batch;
+
+import java.util.Iterator;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle;
+
+import com.google.common.collect.Queues;
+
+public class UnlimitedRawBatchBuffer implements RawBatchBuffer{
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnlimitedRawBatchBuffer.class);
+
+  private final LinkedBlockingDeque<RawFragmentBatch> buffer = Queues.newLinkedBlockingDeque();
+  private volatile boolean finished = false;
+  
+  @Override
+  public void enqueue(ConnectionThrottle throttle, RawFragmentBatch batch) {
+    buffer.add(batch);
+  }
+
+//  @Override
+//  public RawFragmentBatch dequeue() {
+//    return buffer.poll();
+//  }
+
+  @Override
+  public void kill(FragmentContext context) {
+    // TODO: Pass back or kill handler?
+  }
+
+  
+  @Override
+  public void finished() {
+    finished = true;
+  }
+
+  @Override
+  public RawFragmentBatch getNext(){
+    
+    RawFragmentBatch b = buffer.poll();
+    if(b == null && !finished){
+      try {
+        return buffer.take();
+      } catch (InterruptedException e) {
+        return null;
+      }
+    }
+    
+    return b;
+    
+  }
+
+  
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/ErrorHelper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/ErrorHelper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/ErrorHelper.java
new file mode 100644
index 0000000..72c5f34
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/ErrorHelper.java
@@ -0,0 +1,62 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.foreman;
+
+import java.util.UUID;
+
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.slf4j.Logger;
+
+
+public class ErrorHelper {
+  
+  public static DrillPBError logAndConvertError(DrillbitEndpoint endpoint, String message, Throwable t, Logger logger){
+    String id = UUID.randomUUID().toString();
+    DrillPBError.Builder builder = DrillPBError.newBuilder();
+    builder.setEndpoint(endpoint);
+    builder.setErrorId(id);
+    StringBuilder sb = new StringBuilder();
+    if(message != null){
+      sb.append(message);
+    }
+
+    while (true) {
+      sb.append(" < ");
+      sb.append(t.getClass().getSimpleName());
+      if(t.getMessage() != null){
+        sb.append(":[ ");
+        sb.append(t.getMessage());
+        sb.append(" ]");
+      }
+      if (t.getCause() == null || t.getCause() == t) break;
+      t = t.getCause();
+    }
+    
+    builder.setMessage(sb.toString());
+    
+
+    builder.setErrorType(0);
+    
+    // record the error to the log for later reference.
+    logger.error("Error {}: {}", id, message, t);
+    
+    
+    return builder.build();
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
new file mode 100644
index 0000000..26f0192
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
@@ -0,0 +1,283 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.foreman;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.logical.LogicalPlan;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.exception.OptimizerException;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.opt.BasicOptimizer;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.fragment.PlanningSet;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.StatsCollector;
+import org.apache.drill.exec.proto.ExecProtos.PlanFragment;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserProtos.QueryResult;
+import org.apache.drill.exec.proto.UserProtos.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.RequestResults;
+import org.apache.drill.exec.proto.UserProtos.RunQuery;
+import org.apache.drill.exec.rpc.BaseRpcOutcomeListener;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.util.AtomicState;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.WorkManager.WorkerBee;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Lists;
+
+/**
+ * Foreman manages all queries where this is the driving/root node.
+ */
+public class Foreman implements Runnable, Closeable, Comparable<Object>{
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Foreman.class);
+
+  private QueryId queryId;
+  private RunQuery queryRequest;
+  private QueryContext context;
+  private RunningFragmentManager fragmentManager;
+  private WorkerBee bee;
+  private UserClientConnection initiatingClient;
+  private final AtomicState<QueryState> state;
+  
+  
+  public Foreman(WorkerBee bee, DrillbitContext dContext, UserClientConnection connection, QueryId queryId,
+      RunQuery queryRequest) {
+    this.queryId = queryId;
+    this.queryRequest = queryRequest;
+    this.context = new QueryContext(queryId, dContext);
+    this.initiatingClient = connection;
+    this.fragmentManager = new RunningFragmentManager(new ForemanManagerListener(), new TunnelManager(dContext.getBitCom()));
+    this.bee = bee;
+    
+    this.state = new AtomicState<QueryState>(QueryState.PENDING) {
+      protected QueryState getStateFromNumber(int i) {
+        return QueryState.valueOf(i);
+      }
+    };
+  }
+  
+  private boolean isFinished(){
+    switch(state.getState()){
+    case PENDING:
+    case RUNNING:
+      return false;
+    default:
+      return true;
+    }
+    
+  }
+
+  private void fail(String message, Throwable t) {
+    if(isFinished()){
+      logger.error("Received a failure message query finished of: {}", message, t);
+    }
+    DrillPBError error = ErrorHelper.logAndConvertError(context.getCurrentEndpoint(), message, t, logger);
+    QueryResult result = QueryResult //
+        .newBuilder() //
+        .addError(error) //
+        .setIsLastChunk(true) //
+        .setQueryState(QueryState.FAILED) //
+        .setQueryId(queryId) //
+        .build();
+    cleanupAndSendResult(result);
+  }
+
+  
+  public void cancel() {
+    if(isFinished()){
+      return;
+    }
+    
+    // cancel remote fragments.
+    fragmentManager.cancel();
+    
+    QueryResult result = QueryResult.newBuilder().setQueryState(QueryState.CANCELED).setIsLastChunk(true).setQueryId(queryId).build();
+    cleanupAndSendResult(result);
+  }
+  
+  void cleanupAndSendResult(QueryResult result){
+    bee.retireForeman(this);
+    initiatingClient.sendResult(new ResponseSendListener(), new QueryWritableBatch(result));
+  }
+
+  private class ResponseSendListener extends BaseRpcOutcomeListener<Ack> {
+    @Override
+    public void failed(RpcException ex) {
+      logger.info(
+              "Failure while trying communicate query result to initating client.  This would happen if a client is disconnected before response notice can be sent.",
+              ex);
+    }
+  }
+  
+
+
+  /**
+   * Called by execution pool to do foreman setup. Actual query execution is a separate phase (and can be scheduled).
+   */
+  public void run() {
+    // convert a run query request into action
+    try{
+      switch (queryRequest.getType()) {
+      
+      case LOGICAL:
+        parseAndRunLogicalPlan(queryRequest.getPlan());
+        break;
+      case PHYSICAL:
+        parseAndRunPhysicalPlan(queryRequest.getPlan());
+        break;
+      case SQL:
+        runSQL(queryRequest.getPlan());
+        break;
+      default:
+        throw new UnsupportedOperationException();
+      }
+    }catch(Exception ex){
+      fail("Failure while setting up Foreman.", ex);
+    }
+  }
+
+  private void parseAndRunLogicalPlan(String json) {
+    
+    try {
+      LogicalPlan logicalPlan = context.getPlanReader().readLogicalPlan(json);
+      if(logger.isDebugEnabled()) logger.debug("Logical {}", logicalPlan.unparse(context.getConfig()));
+      PhysicalPlan physicalPlan = convert(logicalPlan);
+      if(logger.isDebugEnabled()) logger.debug("Physical {}", context.getConfig().getMapper().writeValueAsString(physicalPlan));
+      runPhysicalPlan(physicalPlan);
+    } catch (IOException e) {
+      fail("Failure while parsing logical plan.", e);
+    } catch (OptimizerException e) {
+      fail("Failure while converting logical plan to physical plan.", e);
+    }
+  }
+
+  private void parseAndRunPhysicalPlan(String json) {
+    try {
+      PhysicalPlan plan = context.getPlanReader().readPhysicalPlan(json);
+      runPhysicalPlan(plan);
+    } catch (IOException e) {
+      fail("Failure while parsing physical plan.", e);
+    }
+  }
+
+  private void runPhysicalPlan(PhysicalPlan plan) {
+
+    PhysicalOperator rootOperator = plan.getSortedOperators(false).iterator().next();
+    MakeFragmentsVisitor makeFragmentsVisitor = new MakeFragmentsVisitor();
+    Fragment rootFragment;
+    try {
+      rootFragment = rootOperator.accept(makeFragmentsVisitor, null);
+    } catch (FragmentSetupException e) {
+      fail("Failure while fragmenting query.", e);
+      return;
+    }
+    
+    
+
+    
+    PlanningSet planningSet = StatsCollector.collectStats(rootFragment);
+    SimpleParallelizer parallelizer = new SimpleParallelizer();
+
+    try {
+      QueryWorkUnit work = parallelizer.getFragments(context.getCurrentEndpoint(), queryId, context.getActiveEndpoints(), context.getPlanReader(), rootFragment, planningSet, 10);
+
+      this.context.getBitCom().getListeners().addFragmentStatusListener(work.getRootFragment().getHandle(), fragmentManager);
+      List<PlanFragment> leafFragments = Lists.newArrayList();
+      List<PlanFragment> intermediateFragments = Lists.newArrayList();
+
+      // store fragments in distributed grid.
+      for (PlanFragment f : work.getFragments()) {
+        if (f.getLeafFragment()) {
+          leafFragments.add(f);
+        } else {
+          intermediateFragments.add(f);
+          context.getCache().storeFragment(f);
+        }
+      }
+
+      fragmentManager.runFragments(bee, work.getRootFragment(), work.getRootOperator(), initiatingClient, leafFragments, intermediateFragments);
+
+    
+    } catch (ExecutionSetupException | RpcException e) {
+      fail("Failure while setting up query.", e);
+    }
+
+  }
+
+  private void runSQL(String json) {
+    throw new UnsupportedOperationException();
+  }
+
+  private PhysicalPlan convert(LogicalPlan plan) throws OptimizerException {
+    return new BasicOptimizer(DrillConfig.create(), context).optimize(new BasicOptimizer.BasicOptimizationContext(), plan);
+  }
+
+  public QueryResult getResult(UserClientConnection connection, RequestResults req) {
+    throw new UnsupportedOperationException();
+  }
+
+
+  public QueryId getQueryId() {
+    return queryId;
+  }
+
+  @Override
+  public void close() throws IOException {
+  }
+  
+  QueryState getQueryState(){
+    return this.state.getState();
+  }
+
+  
+  class ForemanManagerListener{
+    void fail(String message, Throwable t) {
+      ForemanManagerListener.this.fail(message, t);
+    }
+    
+    void cleanupAndSendResult(QueryResult result){
+      Foreman.this.cleanupAndSendResult(result);
+    }
+    
+  }
+
+
+
+  @Override
+  public int compareTo(Object o) {
+    return o.hashCode() - o.hashCode();
+  }
+  
+  
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/FragmentStatusListener.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/FragmentStatusListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/FragmentStatusListener.java
new file mode 100644
index 0000000..d906ba2
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/FragmentStatusListener.java
@@ -0,0 +1,26 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.foreman;
+
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus;
+
+public interface FragmentStatusListener {
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FragmentStatusListener.class);
+  
+  public void statusUpdate(FragmentStatus status);
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/RunningFragmentManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/RunningFragmentManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/RunningFragmentManager.java
new file mode 100644
index 0000000..9d9aca6
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/RunningFragmentManager.java
@@ -0,0 +1,283 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.foreman;
+
+import io.netty.buffer.ByteBuf;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.impl.ImplCreator;
+import org.apache.drill.exec.physical.impl.RootExec;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus;
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus.FragmentState;
+import org.apache.drill.exec.proto.ExecProtos.PlanFragment;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserProtos.QueryResult;
+import org.apache.drill.exec.proto.UserProtos.QueryResult.QueryState;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.RpcOutcomeListener;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.work.AbstractFragmentRunnerListener;
+import org.apache.drill.exec.work.EndpointListener;
+import org.apache.drill.exec.work.FragmentRunner;
+import org.apache.drill.exec.work.WorkManager.WorkerBee;
+import org.apache.drill.exec.work.batch.IncomingBuffers;
+import org.apache.drill.exec.work.foreman.Foreman.ForemanManagerListener;
+import org.apache.drill.exec.work.fragment.LocalFragmentHandler;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Each Foreman holds its own fragment manager.  This manages the events associated with execution of a particular query across all fragments.  
+ */
+class RunningFragmentManager implements FragmentStatusListener{
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RunningFragmentManager.class);
+  
+  public Map<FragmentHandle, FragmentData> map = Maps.newHashMap(); // doesn't need to be
+  private final TunnelManager tun;
+  private ForemanManagerListener foreman;
+  private AtomicInteger remainingFragmentCount;
+  private FragmentRunner rootRunner;
+  private volatile QueryId queryId;
+  
+  public RunningFragmentManager(ForemanManagerListener foreman, TunnelManager tun) {
+    super();
+    this.foreman = foreman;
+    this.tun = tun;
+    this.remainingFragmentCount = new AtomicInteger(0);
+    
+  }
+
+  public void runFragments(WorkerBee bee, PlanFragment rootFragment, FragmentRoot rootOperator, UserClientConnection rootClient, List<PlanFragment> leafFragments, List<PlanFragment> intermediateFragments) throws ExecutionSetupException{
+    remainingFragmentCount.set(leafFragments.size()+1);
+    queryId = rootFragment.getHandle().getQueryId();
+
+    // set up the root fragment first so we'll have incoming buffers available.
+    {
+      IncomingBuffers buffers = new IncomingBuffers(rootOperator);
+      
+      FragmentContext rootContext = new FragmentContext(bee.getContext(), rootFragment.getHandle(), rootClient, buffers, new FunctionImplementationRegistry(bee.getContext().getConfig()));
+      RootExec rootExec = ImplCreator.getExec(rootContext, rootOperator);
+      // add fragment to local node.
+      map.put(rootFragment.getHandle(), new FragmentData(rootFragment.getHandle(), null, true));
+      rootRunner = new FragmentRunner(rootContext, rootExec, new RootFragmentManager(rootContext, rootFragment));
+      LocalFragmentHandler handler = new LocalFragmentHandler(rootFragment.getHandle(), buffers, rootRunner);
+      if(buffers.isDone()){
+        bee.addFragmentRunner(handler.getRunnable());
+      }else{
+        bee.getContext().getBitCom().registerIncomingBatchHandler(handler);
+      }
+      
+    }
+
+    // keep track of intermediate fragments (not root or leaf)
+    for (PlanFragment f : intermediateFragments) {
+      logger.debug("Tracking intermediate remote node {} with data {}", f.getAssignment(), f.getFragmentJson());
+      map.put(f.getHandle(), new FragmentData(f.getHandle(), f.getAssignment(), false));
+    }
+
+    // send remote (leaf) fragments.
+    for (PlanFragment f : leafFragments) {
+      sendRemoteFragment(f);
+    }
+    
+  }
+    
+  private void sendRemoteFragment(PlanFragment fragment){
+    logger.debug("Sending remote fragment to node {} with data {}", fragment.getAssignment(), fragment.getFragmentJson());
+    map.put(fragment.getHandle(), new FragmentData(fragment.getHandle(), fragment.getAssignment(), false));
+    FragmentSubmitListener listener = new FragmentSubmitListener(fragment.getAssignment(), fragment);
+    tun.get(fragment.getAssignment()).sendFragment(listener, fragment);
+  }
+  
+  
+  @Override
+  public void statusUpdate(FragmentStatus status) {
+    logger.debug("New fragment status was provided to Foreman of {}", status);
+    switch(status.getState()){
+    case AWAITING_ALLOCATION:
+      updateStatus(status);
+      break;
+    case CANCELLED:
+      // we don't care about cancellation messages since we're the only entity that should drive cancellations.
+      break;
+    case FAILED:
+      fail(status);
+      break;
+    case FINISHED:
+      finished(status);
+      break;
+    case RUNNING:
+      updateStatus(status);
+      break;
+    default:
+      throw new UnsupportedOperationException();
+    }
+  }
+  
+  private void updateStatus(FragmentStatus status){
+    map.get(status.getHandle()).setStatus(status);
+  }
+  
+  private void finished(FragmentStatus status){
+    updateStatus(status);
+    int remaining = remainingFragmentCount.decrementAndGet();
+    if(remaining == 0){
+      QueryResult result = QueryResult.newBuilder() //
+              .setQueryState(QueryState.COMPLETED) //
+              .setQueryId(queryId) //
+              .build();
+      foreman.cleanupAndSendResult(result);
+    }
+  }
+  
+  private void fail(FragmentStatus status){
+    updateStatus(status);
+    stopQuery();
+    QueryResult result = QueryResult.newBuilder().setQueryId(queryId).setQueryState(QueryState.FAILED).addError(status.getError()).build();
+    foreman.cleanupAndSendResult(result);
+  }
+ 
+  
+  private void stopQuery(){
+    // Stop all queries with a currently active status.
+//    for(FragmentData data: map.values()){
+//      FragmentHandle handle = data.getStatus().getHandle();
+//      switch(data.getStatus().getState()){
+//      case SENDING:
+//      case AWAITING_ALLOCATION:
+//      case RUNNING:
+//        if(data.isLocal()){
+//          rootRunner.cancel();
+//        }else{
+//          tun.get(data.getEndpoint()).cancelFragment(handle).addLightListener(new CancelListener(data.endpoint, handle));
+//        }
+//        break;
+//      default:
+//        break;
+//      }
+//    }
+  }
+  
+  public void cancel(){
+    stopQuery();
+  }
+
+  private class CancelListener extends EndpointListener<Ack, FragmentHandle>{
+    
+    public CancelListener(DrillbitEndpoint endpoint, FragmentHandle handle) {
+      super(endpoint, handle);
+    }
+
+    @Override
+    public void failed(RpcException ex) {
+      logger.error("Failure while attempting to cancel fragment {} on endpoint {}.", value, endpoint, ex);
+    }
+
+    @Override
+    public void success(Ack value, ByteBuf buf) {
+      if(!value.getOk()){
+        logger.warn("Remote node {} responded negative on cancellation request for fragment {}.", endpoint, value);
+      }
+      // do nothing.
+    }
+
+  };
+  
+  public RpcOutcomeListener<Ack> getSubmitListener(DrillbitEndpoint endpoint, PlanFragment value){
+    return new FragmentSubmitListener(endpoint, value);
+  }
+  
+  
+  
+  private class FragmentSubmitListener extends EndpointListener<Ack, PlanFragment>{
+    
+    public FragmentSubmitListener(DrillbitEndpoint endpoint, PlanFragment value) {
+      super(endpoint, value);
+    }
+
+    @Override
+    public void failed(RpcException ex) {
+      logger.debug("Failure while sending fragment.  Stopping query.", ex);
+      stopQuery();
+    }
+
+  }
+  
+  
+  private class FragmentData{
+    private final boolean isLocal;
+    private volatile FragmentStatus status;
+    private volatile long lastStatusUpdate = 0;
+    private final DrillbitEndpoint endpoint;
+    
+    public FragmentData(FragmentHandle handle, DrillbitEndpoint endpoint, boolean isLocal) {
+      super();
+      this.status = FragmentStatus.newBuilder().setHandle(handle).setState(FragmentState.SENDING).build();
+      this.endpoint = endpoint;
+      this.isLocal = isLocal;
+    }
+    
+    public void setStatus(FragmentStatus status){
+      this.status = status;
+      lastStatusUpdate = System.currentTimeMillis();
+    }
+
+    public FragmentStatus getStatus() {
+      return status;
+    }
+
+    public boolean isLocal() {
+      return isLocal;
+    }
+
+    public long getLastStatusUpdate() {
+      return lastStatusUpdate;
+    }
+
+    public DrillbitEndpoint getEndpoint() {
+      return endpoint;
+    }
+    
+    
+  }
+
+  private class RootFragmentManager extends AbstractFragmentRunnerListener{
+
+    private RootFragmentManager(FragmentContext context, PlanFragment fragment){
+      super(context);
+    }
+
+    @Override
+    protected void statusChange(FragmentHandle handle, FragmentStatus status) {
+      RunningFragmentManager.this.statusUpdate(status);
+    }
+
+    
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/TunnelManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/TunnelManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/TunnelManager.java
new file mode 100644
index 0000000..ad3534c
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/TunnelManager.java
@@ -0,0 +1,53 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.foreman;
+
+import java.util.Map;
+
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.bit.BitCom;
+import org.apache.drill.exec.rpc.bit.BitTunnel;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Keeps a local list of tunnels associated with a particular Foreman.
+ */
+public class TunnelManager {
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TunnelManager.class);
+  
+  private final BitCom com;
+  private Map<DrillbitEndpoint, BitTunnel> tunnels = Maps.newHashMap();
+  
+  public TunnelManager(BitCom com){
+    this.com = com;
+  }
+  
+  public BitTunnel get(DrillbitEndpoint ep){
+    BitTunnel bt = tunnels.get(ep);
+    if(bt == null){
+      bt = com.getTunnel(ep);
+      tunnels.put(ep, bt);
+    }
+    
+    return bt;
+  }
+  
+   
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/IncomingFragmentHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/IncomingFragmentHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/IncomingFragmentHandler.java
new file mode 100644
index 0000000..b23f003
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/IncomingFragmentHandler.java
@@ -0,0 +1,49 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.fragment;
+
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle;
+import org.apache.drill.exec.work.FragmentRunner;
+
+/**
+ * Handles incoming fragments as they arrive, routing them as apporpriate. 
+ */
+public interface IncomingFragmentHandler {
+
+  /**
+   * Handle the next incoming fragment.
+   * @param throttle
+   * @param batch
+   * @return True if the fragment has enough incoming data to be able to be run.
+   * @throws FragmentSetupException
+   */
+  public abstract boolean handle(ConnectionThrottle throttle, RawFragmentBatch batch) throws FragmentSetupException;
+
+  /**
+   * Get the fragment runner for this incoming fragment.  Note, this can only be requested once.
+   * @return
+   */
+  public abstract FragmentRunner getRunnable();
+
+  public abstract void cancel();
+  public boolean isWaiting();
+  public abstract FragmentHandle getHandle();
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/LocalFragmentHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/LocalFragmentHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/LocalFragmentHandler.java
new file mode 100644
index 0000000..5ffd09a
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/LocalFragmentHandler.java
@@ -0,0 +1,69 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.fragment;
+
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle;
+import org.apache.drill.exec.work.FragmentRunner;
+import org.apache.drill.exec.work.batch.IncomingBuffers;
+
+public class LocalFragmentHandler implements IncomingFragmentHandler{
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(LocalFragmentHandler.class);
+
+  private final IncomingBuffers buffers;
+  private final FragmentRunner runner;
+  private final FragmentHandle handle;
+  private volatile boolean cancel = false;
+  
+  public LocalFragmentHandler(FragmentHandle handle, IncomingBuffers buffers, FragmentRunner runner) {
+    super();
+    this.handle = handle;
+    this.buffers = buffers;
+    this.runner = runner;
+  }
+
+  @Override
+  public boolean handle(ConnectionThrottle throttle, RawFragmentBatch batch) throws FragmentSetupException {
+    return buffers.batchArrived(throttle, batch);
+  }
+
+  @Override
+  public FragmentRunner getRunnable() {
+    return runner;
+  }
+
+  
+  public FragmentHandle getHandle() {
+    return handle;
+  }
+
+  @Override
+  public void cancel() {
+    cancel = true;
+  }
+
+  @Override
+  public boolean isWaiting() {
+    return !buffers.isDone() && !cancel;
+  }
+  
+  
+  
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RemoteFragmentHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RemoteFragmentHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RemoteFragmentHandler.java
new file mode 100644
index 0000000..d63b4f4
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RemoteFragmentHandler.java
@@ -0,0 +1,124 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.fragment;
+
+import java.io.IOException;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.FragmentLeaf;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.impl.ImplCreator;
+import org.apache.drill.exec.physical.impl.RootExec;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.PlanFragment;
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle;
+import org.apache.drill.exec.rpc.bit.BitTunnel;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.work.FragmentRunner;
+import org.apache.drill.exec.work.FragmentRunnerListener;
+import org.apache.drill.exec.work.RemoteFragmentRunnerListener;
+import org.apache.drill.exec.work.batch.IncomingBuffers;
+
+/**
+ * This handler receives all incoming traffic for a particular FragmentHandle.  It will monitor the state of incoming batches
+ */
+public class RemoteFragmentHandler implements IncomingFragmentHandler {
+  private final PlanFragment fragment;
+  private FragmentLeaf root;
+  private final IncomingBuffers buffers;
+  private final FragmentRunnerListener runnerListener;
+  private volatile FragmentRunner runner;
+  private volatile boolean cancel = false;
+  private final FragmentContext context;
+  private final PhysicalPlanReader reader;
+  
+  public RemoteFragmentHandler(PlanFragment fragment, DrillbitContext context, BitTunnel foremanTunnel) throws FragmentSetupException{
+    try{
+      this.fragment = fragment;
+      this.root = context.getPlanReader().readFragmentOperator(fragment.getFragmentJson());
+      this.buffers = new IncomingBuffers(root);
+      this.context = new FragmentContext(context, fragment.getHandle(), null, buffers, new FunctionImplementationRegistry(context.getConfig()));
+      this.runnerListener = new RemoteFragmentRunnerListener(this.context, foremanTunnel);
+      this.reader = context.getPlanReader();
+      
+    }catch(IOException e){
+      throw new FragmentSetupException("Failure while decoding fragment.", e);
+    }
+  }
+
+  /* (non-Javadoc)
+   * @see org.apache.drill.exec.work.fragment.FragmentHandler#handle(org.apache.drill.exec.rpc.RemoteConnection.ConnectionThrottle, org.apache.drill.exec.record.RawFragmentBatch)
+   */
+  @Override
+  public boolean handle(ConnectionThrottle throttle, RawFragmentBatch batch) throws FragmentSetupException {
+    return buffers.batchArrived(throttle, batch);
+  }
+
+  /* (non-Javadoc)
+   * @see org.apache.drill.exec.work.fragment.FragmentHandler#getRunnable()
+   */
+  @Override
+  public FragmentRunner getRunnable(){
+    synchronized(this){
+      if(runner != null) throw new IllegalStateException("Get Runnable can only be run once.");
+      if(cancel) return null;
+      try {
+        FragmentRoot fragRoot = reader.readFragmentOperator(fragment.getFragmentJson());
+        RootExec exec = ImplCreator.getExec(context, fragRoot);
+        runner = new FragmentRunner(context, exec, runnerListener);
+        return this.runner;
+      } catch (IOException | ExecutionSetupException e) {
+        runnerListener.fail(fragment.getHandle(), "Failure while setting up remote fragment.", e);
+        return null;
+      }
+    }
+    
+  }
+
+  /* (non-Javadoc)
+   * @see org.apache.drill.exec.work.fragment.FragmentHandler#cancel()
+   */
+  @Override
+  public void cancel(){
+    synchronized(this){
+      cancel = true;
+      if(runner != null){
+        runner.cancel();
+      }
+    }
+  }
+
+  @Override
+  public FragmentHandle getHandle() {
+    return fragment.getHandle();
+  }
+
+  @Override
+  public boolean isWaiting() {
+    return !buffers.isDone() && !cancel;
+  }
+  
+  
+
+  
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/work/user/UserWorker.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/user/UserWorker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/user/UserWorker.java
new file mode 100644
index 0000000..621c7cb
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/user/UserWorker.java
@@ -0,0 +1,72 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.drill.exec.work.user;
+
+import java.util.UUID;
+
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserProtos.QueryResult;
+import org.apache.drill.exec.proto.UserProtos.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.RequestResults;
+import org.apache.drill.exec.proto.UserProtos.RunQuery;
+import org.apache.drill.exec.rpc.Acks;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.work.FragmentRunner;
+import org.apache.drill.exec.work.WorkManager.WorkerBee;
+import org.apache.drill.exec.work.foreman.Foreman;
+
+public class UserWorker{
+  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserWorker.class);
+  
+  private final WorkerBee bee;
+    
+  public UserWorker(WorkerBee bee) {
+    super();
+    this.bee = bee;
+  }
+
+  public QueryId submitWork(UserClientConnection connection, RunQuery query){
+    UUID uuid = UUID.randomUUID();
+    QueryId id = QueryId.newBuilder().setPart1(uuid.getMostSignificantBits()).setPart2(uuid.getLeastSignificantBits()).build();
+    Foreman foreman = new Foreman(bee, bee.getContext(), connection, id, query);
+    bee.addNewForeman(foreman);
+    return id;
+  }
+  
+  public QueryResult getResult(UserClientConnection connection, RequestResults req){
+    Foreman foreman = bee.getForemanForQueryId(req.getQueryId());
+    if(foreman == null) return QueryResult.newBuilder().setQueryState(QueryState.UNKNOWN_QUERY).build();
+    return foreman.getResult(connection, req);
+  }
+
+  public Ack cancelQuery(QueryId query){
+    Foreman foreman = bee.getForemanForQueryId(query);
+    if(foreman != null){
+      foreman.cancel();
+    }
+    return Acks.OK;
+  }
+  
+  public Ack cancelFragment(FragmentHandle handle){
+    FragmentRunner runner = bee.getFragmentRunner(handle);
+    if(runner != null) runner.cancel();
+    return Acks.OK;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/parquet/hadoop/CodecFactoryExposer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/parquet/hadoop/CodecFactoryExposer.java b/exec/java-exec/src/main/java/parquet/hadoop/CodecFactoryExposer.java
new file mode 100644
index 0000000..4998888
--- /dev/null
+++ b/exec/java-exec/src/main/java/parquet/hadoop/CodecFactoryExposer.java
@@ -0,0 +1,42 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package parquet.hadoop;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+
+import parquet.bytes.BytesInput;
+import parquet.hadoop.metadata.CompressionCodecName;
+
+public class CodecFactoryExposer{
+
+  private CodecFactory codecFactory;
+
+  public CodecFactoryExposer(Configuration config){
+    codecFactory = new CodecFactory(config);
+  }
+
+  public CodecFactory getCodecFactory() {
+    return codecFactory;
+  }
+
+  public BytesInput decompress(BytesInput bytes, int uncompressedSize, CompressionCodecName codecName) throws IOException {
+    return codecFactory.getDecompressor(codecName).decompress(bytes, uncompressedSize);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/protobuf/Coordination.proto
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/protobuf/Coordination.proto b/exec/java-exec/src/main/protobuf/Coordination.proto
new file mode 100644
index 0000000..5cc5cab
--- /dev/null
+++ b/exec/java-exec/src/main/protobuf/Coordination.proto
@@ -0,0 +1,26 @@
+package exec;
+
+option java_package = "org.apache.drill.exec.proto";
+option java_outer_classname = "CoordinationProtos";
+option optimize_for = SPEED;
+
+message DrillbitEndpoint{
+  optional string address = 1;
+  optional int32 user_port = 2;
+  optional int32 bit_port = 3;
+  optional Roles roles = 4;
+}
+
+message DrillServiceInstance{
+  optional string id = 1;
+  optional int64 registrationTimeUTC = 2;
+  optional DrillbitEndpoint endpoint = 3;
+}
+
+message Roles{
+	optional bool sql_query = 1 [default = true];
+	optional bool logical_plan = 2 [default = true];
+	optional bool physical_plan = 3 [default = true];
+	optional bool java_executor = 4 [default = true];
+	optional bool distributed_cache = 5 [default = true];
+}