diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
index 237aba1..fdbb971 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
@@ -93,6 +93,8 @@
 import org.apache.drill.shaded.guava.com.google.common.annotations.VisibleForTesting;
 import org.apache.drill.shaded.guava.com.google.common.base.Strings;
 import org.apache.drill.shaded.guava.com.google.common.util.concurrent.SettableFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import io.netty.channel.EventLoopGroup;
 
@@ -101,16 +103,15 @@
  * String into ByteBuf.
  */
 public class DrillClient implements Closeable, ConnectionThrottle {
+  private static Logger logger = LoggerFactory.getLogger(DrillClient.class);
   public static final String DEFAULT_CLIENT_NAME = "Apache Drill Java client";
-
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillClient.class);
-
   private static final ObjectMapper objectMapper = new ObjectMapper();
+
   private final DrillConfig config;
   private UserClient client;
   private DrillProperties properties;
   private volatile ClusterCoordinator clusterCoordinator;
-  private volatile boolean connected = false;
+  private volatile boolean connected;
   private final BufferAllocator allocator;
   private final int reconnectTimes;
   private final int reconnectDelay;
@@ -199,10 +200,12 @@
   }
 
   /**
-   * Sets whether the application is willing to accept complex types (Map, Arrays) in the returned result set.
-   * Default is {@code true}. If set to {@code false}, the complex types are returned as JSON encoded VARCHAR type.
+   * Sets whether the application is willing to accept complex types (Map,
+   * Arrays) in the returned result set. Default is {@code true}. If set to
+   * {@code false}, the complex types are returned as JSON encoded VARCHAR type.
    *
-   * @throws IllegalStateException if called after a connection has been established.
+   * @throws IllegalStateException
+   *           if called after a connection has been established.
    */
   public void setSupportComplexTypes(boolean supportComplexTypes) {
     if (connected) {
@@ -468,10 +471,10 @@
    */
   @Override
   public void close() {
-    if (this.client != null) {
-      this.client.close();
+    if (client != null) {
+      client.close();
     }
-    if (this.ownsAllocator && allocator != null) {
+    if (ownsAllocator && allocator != null) {
       DrillAutoCloseables.closeNoChecked(allocator);
     }
     if (ownsZkConnection) {
@@ -486,19 +489,17 @@
     }
     if (eventLoopGroup != null) {
       eventLoopGroup.shutdownGracefully();
+      eventLoopGroup = null;
     }
 
     if (executor != null) {
       executor.shutdownNow();
+      executor = null;
     }
 
-    // TODO:  Did DRILL-1735 changes cover this TODO?:
-    // TODO: fix tests that fail when this is called.
-    //allocator.close();
     connected = false;
   }
 
-
   /**
    * Return the server infos. Only available after connecting
    *
@@ -585,11 +586,15 @@
 
   /**
    * API to just plan a query without execution
+   *
    * @param type
    * @param query
-   * @param isSplitPlan - option to tell whether to return single or split plans for a query
-   * @return list of PlanFragments that can be used later on in {@link #runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType, java.util.List, org.apache.drill.exec.rpc.user.UserResultsListener)}
-   * to run a query without additional planning
+   * @param isSplitPlan
+   *          option to tell whether to return single or split plans for a query
+   * @return list of PlanFragments that can be used later on in
+   *         {@link #runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType,
+   *         java.util.List, org.apache.drill.exec.rpc.user.UserResultsListener)}
+   *         to run a query without additional planning
    */
   public DrillRpcFuture<QueryPlanFragments> planQuery(QueryType type, String query, boolean isSplitPlan) {
     GetQueryPlanFragments runQuery = GetQueryPlanFragments.newBuilder().setQuery(query).setType(type).setSplitPlan(isSplitPlan).build();
@@ -632,7 +637,7 @@
     client.submitQuery(resultsListener, query);
   }
 
-  /*
+  /**
    * Helper method to generate the UserCredentials message from the properties.
    */
   private UserBitShared.UserCredentials getUserCredentials() {
@@ -660,10 +665,10 @@
   }
 
   /**
-   * Get the list of catalogs in <code>INFORMATION_SCHEMA.CATALOGS</code> table satisfying the given filters.
+   * Get the list of catalogs in {@code INFORMATION_SCHEMA.CATALOGS} table satisfying the given filters.
    *
-   * @param catalogNameFilter Filter on <code>catalog name</code>. Pass null to apply no filter.
-   * @return The list of catalogs in <code>INFORMATION_SCHEMA.CATALOGS</code> table satisfying the given filters.
+   * @param catalogNameFilter Filter on {@code catalog name}. Pass null to apply no filter.
+   * @return The list of catalogs in {@code INFORMATION_SCHEMA.CATALOGS} table satisfying the given filters.
    */
   public DrillRpcFuture<GetCatalogsResp> getCatalogs(LikeFilter catalogNameFilter) {
     final GetCatalogsReq.Builder reqBuilder = GetCatalogsReq.newBuilder();
@@ -675,11 +680,11 @@
   }
 
   /**
-   * Get the list of schemas in <code>INFORMATION_SCHEMA.SCHEMATA</code> table satisfying the given filters.
+   * Get the list of schemas in {@code INFORMATION_SCHEMA.SCHEMATA} table satisfying the given filters.
    *
-   * @param catalogNameFilter Filter on <code>catalog name</code>. Pass null to apply no filter.
-   * @param schemaNameFilter Filter on <code>schema name</code>. Pass null to apply no filter.
-   * @return The list of schemas in <code>INFORMATION_SCHEMA.SCHEMATA</code> table satisfying the given filters.
+   * @param catalogNameFilter Filter on {@code catalog name}. Pass null to apply no filter.
+   * @param schemaNameFilter Filter on {@code schema name}. Pass null to apply no filter.
+   * @return The list of schemas in {@code INFORMATION_SCHEMA.SCHEMATA} table satisfying the given filters.
    */
   public DrillRpcFuture<GetSchemasResp> getSchemas(LikeFilter catalogNameFilter, LikeFilter schemaNameFilter) {
     final GetSchemasReq.Builder reqBuilder = GetSchemasReq.newBuilder();
@@ -695,13 +700,13 @@
   }
 
   /**
-   * Get the list of tables in <code>INFORMATION_SCHEMA.TABLES</code> table satisfying the given filters.
+   * Get the list of tables in {@code INFORMATION_SCHEMA.TABLES} table satisfying the given filters.
    *
-   * @param catalogNameFilter Filter on <code>catalog name</code>. Pass null to apply no filter.
-   * @param schemaNameFilter Filter on <code>schema name</code>. Pass null to apply no filter.
-   * @param tableNameFilter Filter in <code>table name</code>. Pass null to apply no filter.
-   * @param tableTypeFilter Filter in <code>table type</code>. Pass null to apply no filter
-   * @return The list of tables in <code>INFORMATION_SCHEMA.TABLES</code> table satisfying the given filters.
+   * @param catalogNameFilter Filter on {@code catalog name}. Pass null to apply no filter.
+   * @param schemaNameFilter Filter on {@code schema name}. Pass null to apply no filter.
+   * @param tableNameFilter Filter in {@code table name}. Pass null to apply no filter.
+   * @param tableTypeFilter Filter in {@code table type}. Pass null to apply no filter
+   * @return The list of tables in {@code INFORMATION_SCHEMA.TABLES} table satisfying the given filters.
    */
   public DrillRpcFuture<GetTablesResp> getTables(LikeFilter catalogNameFilter, LikeFilter schemaNameFilter,
       LikeFilter tableNameFilter, List<String> tableTypeFilter) {
@@ -726,13 +731,13 @@
   }
 
   /**
-   * Get the list of columns in <code>INFORMATION_SCHEMA.COLUMNS</code> table satisfying the given filters.
+   * Get the list of columns in {@code INFORMATION_SCHEMA.COLUMNS} table satisfying the given filters.
    *
-   * @param catalogNameFilter Filter on <code>catalog name</code>. Pass null to apply no filter.
-   * @param schemaNameFilter Filter on <code>schema name</code>. Pass null to apply no filter.
-   * @param tableNameFilter Filter in <code>table name</code>. Pass null to apply no filter.
-   * @param columnNameFilter Filter in <code>column name</code>. Pass null to apply no filter.
-   * @return The list of columns in <code>INFORMATION_SCHEMA.COLUMNS</code> table satisfying the given filters.
+   * @param catalogNameFilter Filter on {@code catalog name}. Pass null to apply no filter.
+   * @param schemaNameFilter Filter on {@code schema name}. Pass null to apply no filter.
+   * @param tableNameFilter Filter in {@code table name}. Pass null to apply no filter.
+   * @param columnNameFilter Filter in {@code column name}. Pass null to apply no filter.
+   * @return The list of columns in {@code INFORMATION_SCHEMA.COLUMNS} table satisfying the given filters.
    */
   public DrillRpcFuture<GetColumnsResp> getColumns(LikeFilter catalogNameFilter, LikeFilter schemaNameFilter,
       LikeFilter tableNameFilter, LikeFilter columnNameFilter) {
@@ -757,10 +762,10 @@
   }
 
   /**
-   * Create a prepared statement for given the <code>query</code>.
+   * Create a prepared statement for given the {@code query}.
    *
    * @param query
-   * @return The prepared statement for given the <code>query</code>.
+   * @return The prepared statement for given the {@code query}.
    */
   public DrillRpcFuture<CreatePreparedStatementResp> createPreparedStatement(final String query) {
     final CreatePreparedStatementReq req =
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java
index 47a364a..c751897 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DumpCat.java
@@ -250,7 +250,7 @@
       System.out.println(getBatchMetaInfo(vcSerializable).toString());
 
       System.out.println("Schema Information");
-      for (final VectorWrapper w : vectorContainer) {
+      for (final VectorWrapper<?> w : vectorContainer) {
         final MaterializedField field = w.getValueVector().getField();
         System.out.println (String.format("name : %s, minor_type : %s, data_mode : %s",
                                           field.getName(),
@@ -279,7 +279,7 @@
       selectedRows = vcSerializable.getSv2().getCount();
     }
 
-    for (final VectorWrapper w : vectorContainer) {
+    for (final VectorWrapper<?> w : vectorContainer) {
        totalDataSize += w.getValueVector().getBufferSize();
     }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/InvalidConnectionInfoException.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/InvalidConnectionInfoException.java
index 19e72ff..e90cfae 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/InvalidConnectionInfoException.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/InvalidConnectionInfoException.java
@@ -22,6 +22,7 @@
 /**
  * Exception for malformed connection string from client
  */
+@SuppressWarnings("serial")
 public class InvalidConnectionInfoException extends NonTransientRpcException {
 
   public InvalidConnectionInfoException(String message) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/LoggingResultsListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/LoggingResultsListener.java
index 25e472f..25331da 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/LoggingResultsListener.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/LoggingResultsListener.java
@@ -36,11 +36,13 @@
 import org.apache.drill.exec.util.VectorUtil;
 
 import org.apache.drill.shaded.guava.com.google.common.base.Stopwatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import io.netty.buffer.DrillBuf;
 
 public class LoggingResultsListener implements UserResultsListener {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(LoggingResultsListener.class);
+  private static Logger logger = LoggerFactory.getLogger(LoggingResultsListener.class);
 
   private final AtomicInteger count = new AtomicInteger();
   private final Stopwatch w = Stopwatch.createUnstarted();
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
index d60f6ae..5a5f725 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
@@ -201,7 +201,5 @@
       watch.reset();
     }
     return 0;
-
   }
-
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/AccountingUserConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/AccountingUserConnection.java
index c0c1e0b..7510cf4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/AccountingUserConnection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/AccountingUserConnection.java
@@ -17,7 +17,7 @@
  */
 package org.apache.drill.exec.ops;
 
-import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
+import org.apache.drill.exec.physical.impl.materialize.QueryDataPackage;
 import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
 import org.apache.drill.exec.rpc.RpcOutcomeListener;
 import org.apache.drill.exec.rpc.UserClientConnection;
@@ -31,14 +31,15 @@
   private final SendingAccountor sendingAccountor;
   private final RpcOutcomeListener<Ack> statusHandler;
 
-  public AccountingUserConnection(UserClientConnection connection, SendingAccountor sendingAccountor, RpcOutcomeListener<Ack> statusHandler) {
+  public AccountingUserConnection(UserClientConnection connection, SendingAccountor sendingAccountor,
+      RpcOutcomeListener<Ack> statusHandler) {
     this.connection = connection;
     this.sendingAccountor = sendingAccountor;
     this.statusHandler = statusHandler;
   }
 
-  public void sendData(QueryWritableBatch batch) {
+  public void sendData(QueryDataPackage data) {
     sendingAccountor.increment();
-    connection.sendData(statusHandler, batch);
+    connection.sendData(statusHandler, data);
   }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java
index 0f7ca13..3f71cbd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java
@@ -19,17 +19,14 @@
 
 import java.util.List;
 
-import org.apache.drill.exec.exception.OutOfMemoryException;
 import org.apache.drill.exec.ops.AccountingUserConnection;
 import org.apache.drill.exec.ops.ExecutorFragmentContext;
 import org.apache.drill.exec.ops.MetricDef;
 import org.apache.drill.exec.ops.RootFragmentContext;
 import org.apache.drill.exec.physical.config.Screen;
-import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
-import org.apache.drill.exec.physical.impl.materialize.RecordMaterializer;
+import org.apache.drill.exec.physical.impl.materialize.QueryDataPackage.DataPackage;
+import org.apache.drill.exec.physical.impl.materialize.QueryDataPackage.EmptyResultsPackage;
 import org.apache.drill.exec.physical.impl.materialize.VectorRecordMaterializer;
-import org.apache.drill.exec.proto.UserBitShared.QueryData;
-import org.apache.drill.exec.proto.UserBitShared.RecordBatchDef;
 import org.apache.drill.exec.record.RecordBatch;
 import org.apache.drill.exec.record.RecordBatch.IterOutcome;
 import org.apache.drill.exec.testing.ControlsInjector;
@@ -49,13 +46,18 @@
     return new ScreenRoot(context, children.iterator().next(), config);
   }
 
+  /**
+   * Transfer batches to a user connection. The user connection is typically a
+   * network connection, but may be internal for a web or REST client. Data is
+   * sent as a "package", allowing the network client to request serialization,
+   * and the internal client to just transfer buffer ownership.
+   */
   public static class ScreenRoot extends BaseRootExec {
     private static final Logger logger = LoggerFactory.getLogger(ScreenRoot.class);
     private final RecordBatch incoming;
     private final RootFragmentContext context;
     private final AccountingUserConnection userConnection;
-    private RecordMaterializer materializer;
-
+    private DataPackage dataPackage;
     private boolean firstBatch = true;
 
     public enum Metric implements MetricDef {
@@ -67,15 +69,11 @@
       }
     }
 
-    public ScreenRoot(RootFragmentContext context, RecordBatch incoming, Screen config) throws OutOfMemoryException {
+    public ScreenRoot(RootFragmentContext context, RecordBatch incoming, Screen config) {
       super(context, config);
       this.context = context;
       this.incoming = incoming;
-      userConnection = context.getUserDataTunnel();
-    }
-
-    public RootFragmentContext getContext() {
-      return context;
+      this.userConnection = context.getUserDataTunnel();
     }
 
     @Override
@@ -85,53 +83,40 @@
       switch (outcome) {
         case NONE:
           if (firstBatch) {
-            // this is the only data message sent to the client and may contain the schema
-            QueryWritableBatch batch;
-            QueryData header = QueryData.newBuilder()
-              .setQueryId(context.getHandle().getQueryId())
-              .setRowCount(0)
-              .setDef(RecordBatchDef.getDefaultInstance())
-              .build();
-            batch = new QueryWritableBatch(header);
 
             stats.startWait();
             try {
-              userConnection.sendData(batch);
+              // This is the only data message sent to the client and does not contain the schema
+              userConnection.sendData(new EmptyResultsPackage(context.getHandle().getQueryId()));
             } finally {
               stats.stopWait();
             }
             firstBatch = false; // we don't really need to set this. But who knows!
           }
-
           return false;
+
         case OK_NEW_SCHEMA:
-          materializer = new VectorRecordMaterializer(context, oContext, incoming);
+          dataPackage = new DataPackage(new VectorRecordMaterializer(context, oContext, incoming), stats);
           //$FALL-THROUGH$
         case OK:
           injector.injectPause(context.getExecutionControls(), "sending-data", logger);
-          final QueryWritableBatch batch = materializer.convertNext();
-          updateStats(batch);
           stats.startWait();
           try {
-            userConnection.sendData(batch);
+            // Stats updated if connection serializes the batch
+            userConnection.sendData(dataPackage);
           } finally {
             stats.stopWait();
           }
           firstBatch = false;
-
           return true;
+
         default:
           throw new UnsupportedOperationException(outcome.name());
       }
     }
 
-    public void updateStats(QueryWritableBatch queryBatch) {
-      stats.addLongStat(Metric.BYTES_SENT, queryBatch.getByteCount());
-    }
-
-    RecordBatch getIncoming() {
-      return incoming;
-    }
+    public RootFragmentContext getContext() { return context; }
+    protected RecordBatch getIncoming() { return incoming; }
 
     @Override
     public void close() throws Exception {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java
index a3d03c2..ec22632 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java
@@ -24,9 +24,12 @@
 import org.apache.drill.exec.record.TransferPair;
 
 public interface Filterer {
-  TemplateClassDefinition<Filterer> TEMPLATE_DEFINITION2 = new TemplateClassDefinition<Filterer>(Filterer.class, FilterTemplate2.class);
-  TemplateClassDefinition<Filterer> TEMPLATE_DEFINITION4 = new TemplateClassDefinition<Filterer>(Filterer.class, FilterTemplate4.class);
+  TemplateClassDefinition<Filterer> TEMPLATE_DEFINITION2 =
+      new TemplateClassDefinition<Filterer>(Filterer.class, FilterTemplate2.class);
+  TemplateClassDefinition<Filterer> TEMPLATE_DEFINITION4 =
+      new TemplateClassDefinition<Filterer>(Filterer.class, FilterTemplate4.class);
 
-  void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, TransferPair[] transfers) throws SchemaChangeException;
+  void setup(FragmentContext context, RecordBatch incoming,
+      RecordBatch outgoing, TransferPair[] transfers) throws SchemaChangeException;
   void filterBatch(int recordCount) throws SchemaChangeException;
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/QueryDataPackage.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/QueryDataPackage.java
new file mode 100644
index 0000000..f9a933d
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/QueryDataPackage.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.physical.impl.materialize;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.drill.exec.ops.OperatorStats;
+import org.apache.drill.exec.physical.impl.ScreenCreator.ScreenRoot.Metric;
+import org.apache.drill.exec.proto.UserBitShared.QueryData;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.RecordBatchDef;
+import org.apache.drill.exec.proto.UserBitShared.SerializedField;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.VectorWrapper;
+
+/**
+ * Packages a batch from the Screen operator to send to its
+ * user connection. In the original Drill, that connection was always a
+ * network connection, and so the outgoing batch is serialized to a set
+ * of buffers ready to send. However, the REST server runs in the same process.
+ * The original REST query implementation serialized the data to buffers, then
+ * copied the data to a big buffer to be deserialized, causing significant memory
+ * pressure. This version allows the user connection to elect for serialization,
+ * or just to access the original source batch.
+ */
+public interface QueryDataPackage {
+
+  QueryId queryId();
+  QueryWritableBatch toWritableBatch();
+  VectorContainer batch();
+  List<SerializedField> fields();
+
+  /**
+   * Package that contains only a query ID. Send for a query that
+   * finishes with no data. The results are null: no data, no schema.
+   */
+  public static class EmptyResultsPackage implements QueryDataPackage {
+
+    private final QueryId queryId;
+
+    public EmptyResultsPackage(QueryId queryId) {
+      this.queryId = queryId;
+    }
+
+    @Override
+    public QueryId queryId() { return queryId; }
+
+    /**
+     * Creates a message that sends only the query ID to the
+     * client.
+     */
+    @Override
+    public QueryWritableBatch toWritableBatch() {
+      QueryData header = QueryData.newBuilder()
+        .setQueryId(queryId)
+        .setRowCount(0)
+        .setDef(RecordBatchDef.getDefaultInstance())
+        .build();
+      return new QueryWritableBatch(header);
+    }
+
+    @Override
+    public VectorContainer batch() { return null; }
+
+    @Override
+    public List<SerializedField> fields() {
+      return Collections.emptyList();
+    }
+  }
+
+  /**
+   * Represents a batch of data with a schema.
+   */
+  public static class DataPackage implements QueryDataPackage {
+    private final RecordMaterializer materializer;
+    private final OperatorStats stats;
+
+    public DataPackage(RecordMaterializer materializer, OperatorStats stats) {
+      this.materializer = materializer;
+      this.stats = stats;
+    }
+
+    @Override
+    public QueryId queryId() { return materializer.queryId(); }
+
+    @Override
+    public QueryWritableBatch toWritableBatch() {
+      QueryWritableBatch batch = materializer.convertNext();
+      stats.addLongStat(Metric.BYTES_SENT, batch.getByteCount());
+      return batch;
+    }
+
+    @Override
+    public VectorContainer batch() {
+      return materializer.incoming();
+    }
+
+    @Override
+    public List<SerializedField> fields() {
+      List<SerializedField> metadata = new ArrayList<>();
+      for (VectorWrapper<?> vw : batch()) {
+        metadata.add(vw.getValueVector().getMetadata());
+      }
+      return metadata;
+    }
+  }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/QueryWritableBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/QueryWritableBatch.java
index e69bd51..f9537b0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/QueryWritableBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/QueryWritableBatch.java
@@ -24,7 +24,6 @@
 import org.apache.drill.exec.proto.UserBitShared.QueryData;
 
 public class QueryWritableBatch {
-//  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryWritableBatch.class);
 
   private final QueryData header;
   private final ByteBuf[] buffers;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/RecordMaterializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/RecordMaterializer.java
index 75de592..9c2f7ca 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/RecordMaterializer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/RecordMaterializer.java
@@ -17,9 +17,14 @@
  */
 package org.apache.drill.exec.physical.impl.materialize;
 
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.record.VectorContainer;
 
 public interface RecordMaterializer {
 
   public QueryWritableBatch convertNext();
 
+  public QueryId queryId();
+
+  public VectorContainer incoming();
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/VectorRecordMaterializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/VectorRecordMaterializer.java
index 7cdf9b3..c294774 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/VectorRecordMaterializer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/materialize/VectorRecordMaterializer.java
@@ -25,16 +25,16 @@
 import org.apache.drill.exec.proto.UserBitShared.QueryData;
 import org.apache.drill.exec.record.BatchSchema;
 import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.VectorContainer;
 import org.apache.drill.exec.record.WritableBatch;
-import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.server.options.OptionSet;
 
 public class VectorRecordMaterializer implements RecordMaterializer {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(VectorRecordMaterializer.class);
 
-  private QueryId queryId;
-  private RecordBatch batch;
-  private BufferAllocator allocator;
-  private OptionManager options;
+  private final QueryId queryId;
+  private final RecordBatch batch;
+  private final BufferAllocator allocator;
+  private final boolean resultResultsForDDL;
 
   public VectorRecordMaterializer(FragmentContext context, OperatorContext oContext, RecordBatch batch) {
     this.queryId = context.getHandle().getQueryId();
@@ -42,19 +42,27 @@
     this.allocator = oContext.getAllocator();
     BatchSchema schema = batch.getSchema();
     assert schema != null : "Schema must be defined.";
-    options = context.getOptions();
+    OptionSet options = context.getOptions();
+    this.resultResultsForDDL = options.getBoolean(ExecConstants.RETURN_RESULT_SET_FOR_DDL);
   }
 
+  @Override
   public QueryWritableBatch convertNext() {
     WritableBatch w = batch.getWritableBatch().transfer(allocator);
     QueryData.Builder builder = QueryData.newBuilder()
         .setQueryId(queryId)
         .setRowCount(batch.getRecordCount())
         .setDef(w.getDef());
-    if (!options.getBoolean(ExecConstants.RETURN_RESULT_SET_FOR_DDL)) {
+    if (!resultResultsForDDL) {
       int count = w.getDef().getAffectedRowsCount();
       builder.setAffectedRowsCount(count == -1 ? 0 : count);
     }
     return new QueryWritableBatch(builder.build(), w.getBuffers());
   }
+
+  @Override
+  public QueryId queryId() { return queryId; }
+
+  @Override
+  public VectorContainer incoming() { return batch.getContainer(); }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
index cb835e7..d4087fb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
@@ -25,7 +25,7 @@
  * Implements an AbstractUnaryRecordBatch where the incoming record batch is
  * known at the time of creation
  *
- * @param <T>
+ * @param <T> the plan definition of the operator
  */
 public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> extends AbstractUnaryRecordBatch<T> {
 
@@ -43,12 +43,15 @@
   }
 
   /**
-   * Based on lastKnownOutcome and if there are more records to be output for current record boundary detected by
-   * EMIT outcome, this method returns EMIT or OK outcome.
+   * Based on lastKnownOutcome and if there are more records to be output for
+   * current record boundary detected by EMIT outcome, this method returns EMIT
+   * or OK outcome.
+   *
    * @param hasMoreRecordInBoundary
-   * @return - EMIT - If the lastknownOutcome was EMIT and output records corresponding to all the incoming records in
-   * current record boundary is already produced.
-   *         - OK - otherwise
+   * @return EMIT - If the lastknownOutcome was EMIT and output records
+   *         corresponding to all the incoming records in current record
+   *         boundary is already produced.
+   *         OK - otherwise
    */
   protected IterOutcome getFinalOutcome(boolean hasMoreRecordInBoundary) {
     final IterOutcome lastOutcome = getLastKnownOutcome();
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
index 6d8865d..2ee1047 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
@@ -30,7 +30,6 @@
  * {@link org.apache.drill.exec.record.metadata.TupleMetadata} instead.
  */
 public class BatchSchema implements Iterable<MaterializedField> {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BatchSchema.class);
 
   private final SelectionVectorMode selectionVectorMode;
   private final List<MaterializedField> fields;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java
index 064c601..1eb2ac0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchLoader.java
@@ -51,7 +51,7 @@
   private final static Logger logger = LoggerFactory.getLogger(RecordBatchLoader.class);
 
   private final BufferAllocator allocator;
-  private VectorContainer container = new VectorContainer();
+  private VectorContainer container;
   private int valueCount;
   private BatchSchema schema;
 
@@ -60,6 +60,7 @@
    */
   public RecordBatchLoader(BufferAllocator allocator) {
     this.allocator = Preconditions.checkNotNull(allocator);
+    this.container = new VectorContainer(allocator);
   }
 
   public BufferAllocator allocator() { return allocator; }
@@ -116,12 +117,10 @@
           vector = TypeHelper.getNewVector(fieldDef, allocator);
 
         // If the field is a map or a dict, check if the schema changed.
-
         } else if ((vector.getField().getType().getMinorType() == MinorType.MAP || vector.getField().getType().getMinorType() == MinorType.DICT) &&
                    ! isSameSchema(vector.getField().getChildren(), field.getChildList())) {
 
           // The schema changed. Discard the old one and create a new one.
-
           schemaChanged = true;
           vector.clear();
           vector = TypeHelper.getNewVector(fieldDef, allocator);
@@ -155,8 +154,8 @@
       container = newVectors;
       container.setRecordCount(valueCount);
     } catch (final Throwable cause) {
-      // We have to clean up new vectors created here and pass over the actual cause. It is upper layer who should
-      // adjudicate to call upper layer specific clean up logic.
+      // We have to clean up new vectors created here and pass over the actual cause.
+      // It is upper layer who should adjudicate to call upper layer specific clean up logic.
       VectorAccessibleUtilities.clear(newVectors);
       throw cause;
     } finally {
@@ -190,7 +189,6 @@
 
     // Column order can permute (see DRILL-5828). So, use a map
     // for matching.
-
     Map<String, MaterializedField> childMap = CaseInsensitiveMap.newHashMap();
     for (MaterializedField currentChild : currentChildren) {
       childMap.put(currentChild.getName(), currentChild);
@@ -199,13 +197,11 @@
       MaterializedField currentChild = childMap.get(newChild.getNamePart().getName());
 
       // New map member?
-
       if (currentChild == null) {
         return false;
       }
 
       // Changed data type?
-
       if (! currentChild.getType().equals(newChild.getMajorType())) {
         return false;
       }
@@ -223,7 +219,6 @@
     }
 
     // Everything matches.
-
     return true;
   }
 
@@ -232,20 +227,6 @@
     return container.getValueVectorId(path);
   }
 
-//
-//  @SuppressWarnings("unchecked")
-//  public <T extends ValueVector> T getValueVectorId(int fieldId, Class<?> clazz) {
-//    ValueVector v = container.get(fieldId);
-//    assert v != null;
-//    if (v.getClass() != clazz){
-//      logger.warn(String.format(
-//          "Failure while reading vector.  Expected vector class of %s but was holding vector class %s.",
-//          clazz.getCanonicalName(), v.getClass().getCanonicalName()));
-//      return null;
-//    }
-//    return (T) v;
-//  }
-
   @Override
   public int getRecordCount() { return valueCount; }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
index 966ade7..1b30d78 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
@@ -19,6 +19,7 @@
 
 import io.netty.buffer.DrillBuf;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.drill.exec.memory.BufferAllocator;
@@ -28,7 +29,6 @@
 import org.apache.drill.exec.vector.ValueVector;
 
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
-import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 
 /**
  * A specialized version of record batch that can moves out buffers and preps
@@ -52,7 +52,7 @@
   }
 
   public WritableBatch transfer(BufferAllocator allocator) {
-    List<DrillBuf> newBuffers = Lists.newArrayList();
+    List<DrillBuf> newBuffers = new ArrayList<>();
     for (DrillBuf buf : buffers) {
       int writerIndex = buf.writerIndex();
       DrillBuf newBuf = buf.transferOwnership(allocator).buffer;
@@ -135,7 +135,7 @@
   }
 
   public static WritableBatch getBatchNoHVWrap(int recordCount, Iterable<VectorWrapper<?>> vws, boolean isSV2) {
-    List<ValueVector> vectors = Lists.newArrayList();
+    List<ValueVector> vectors = new ArrayList<>();
     for (VectorWrapper<?> vw : vws) {
       Preconditions.checkArgument(!vw.isHyper());
       vectors.add(vw.getValueVector());
@@ -144,8 +144,8 @@
   }
 
   public static WritableBatch getBatchNoHV(int recordCount, Iterable<ValueVector> vectors, boolean isSV2) {
-    List<DrillBuf> buffers = Lists.newArrayList();
-    List<SerializedField> metadata = Lists.newArrayList();
+    List<DrillBuf> buffers = new ArrayList<>();
+    List<SerializedField> metadata = new ArrayList<>();
 
     for (ValueVector vv : vectors) {
       metadata.add(vv.getMetadata());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractDisposableUserClientConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractDisposableUserClientConnection.java
index 2934c34..0fd2862 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractDisposableUserClientConnection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/AbstractDisposableUserClientConnection.java
@@ -18,6 +18,8 @@
 package org.apache.drill.exec.rpc;
 
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
 import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
 import org.apache.drill.exec.proto.UserBitShared.QueryId;
@@ -29,13 +31,14 @@
 import java.util.concurrent.TimeUnit;
 
 /**
- * Helps to run a query and await on the results. All the inheriting sub-class manages the session/connection
- * state and submits query with respect to that state. The subclass instance lifetime is per query lifetime
- * and is not re-used.
+ * Helps to run a query and await on the results. All the inheriting sub-class
+ * manages the session/connection state and submits query with respect to that
+ * state. The subclass instance lifetime is per query lifetime and is not
+ * re-used.
  */
 public abstract class AbstractDisposableUserClientConnection implements UserClientConnection {
-  private static final org.slf4j.Logger logger =
-      org.slf4j.LoggerFactory.getLogger(AbstractDisposableUserClientConnection.class);
+  private static final Logger logger =
+      LoggerFactory.getLogger(AbstractDisposableUserClientConnection.class);
 
   protected final CountDownLatch latch = new CountDownLatch(1);
 
@@ -72,7 +75,8 @@
     final QueryId queryId = result.getQueryId();
 
     if (logger.isDebugEnabled()) {
-      logger.debug("Result arrived for QueryId: {} with QueryState: {}", QueryIdHelper.getQueryId(queryId), state);
+      logger.debug("Result arrived for QueryId: {} with QueryState: {}",
+          QueryIdHelper.getQueryId(queryId), state);
     }
 
     switch (state) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/UserClientConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/UserClientConnection.java
index 1372b29..179cc7c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/UserClientConnection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/UserClientConnection.java
@@ -18,7 +18,8 @@
 package org.apache.drill.exec.rpc;
 
 import io.netty.channel.ChannelFuture;
-import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
+
+import org.apache.drill.exec.physical.impl.materialize.QueryDataPackage;
 import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
 import org.apache.drill.exec.proto.UserBitShared.QueryResult;
 import org.apache.drill.exec.rpc.user.UserSession;
@@ -26,12 +27,14 @@
 import java.net.SocketAddress;
 
 /**
- * Interface for getting user session properties and interacting with user connection. Separating this interface from
- * {@link AbstractRemoteConnection} implementation for user connection:
+ * Interface for getting user session properties and interacting with user
+ * connection. Separating this interface from {@link AbstractRemoteConnection}
+ * implementation for user connection:
  * <p><ul>
- * <li> Connection is passed to Foreman and Screen operators. Instead passing this interface exposes few details.
- * <li> Makes it easy to have wrappers around user connection which can be helpful to tap the messages and data
- * going to the actual client.
+ * <li>Connection is passed to Foreman and Screen operators. Instead passing
+ * this interface exposes few details.
+ * <li>Makes it easy to have wrappers around user connection which can be
+ * helpful to tap the messages and data going to the actual client.
  * </ul>
  */
 public interface UserClientConnection {
@@ -41,7 +44,7 @@
   UserSession getSession();
 
   /**
-   * Send query result outcome to client. Outcome is returned through <code>listener</code>
+   * Send query result outcome to client. Outcome is returned through {@code listener}.
    *
    * @param listener
    * @param result
@@ -49,12 +52,12 @@
   void sendResult(RpcOutcomeListener<Ack> listener, QueryResult result);
 
   /**
-   * Send query data to client. Outcome is returned through <code>listener</code>
+   * Send query data to client. Outcome is returned through {@code listener}.
    *
    * @param listener
    * @param result
    */
-  void sendData(RpcOutcomeListener<Ack> listener, QueryWritableBatch result);
+  void sendData(RpcOutcomeListener<Ack> listener, QueryDataPackage data);
 
   /**
    * Returns the {@link ChannelFuture} which will be notified when this
@@ -66,4 +69,4 @@
    * @return Return the client node address.
    */
   SocketAddress getRemoteAddress();
-}
\ No newline at end of file
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/AwaitableUserResultsListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/AwaitableUserResultsListener.java
index 461d8aa..25f1c68 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/AwaitableUserResultsListener.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/AwaitableUserResultsListener.java
@@ -78,5 +78,4 @@
     }
     return count.get();
   }
-
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
index 99b0723..460dbb7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
@@ -120,13 +120,13 @@
 
     try {
       if (isFailureResult) {
-        // Failure case--pass on via submissionFailed(...).
 
+        // Failure case--pass on via submissionFailed(...).
         resultsListener.submissionFailed(new UserRemoteException(queryResult.getError(0)));
         // Note: Listener is removed in finally below.
       } else if (isTerminalResult) {
-        // A successful completion/canceled case--pass on via resultArrived
 
+        // A successful completion/canceled case--pass on via resultArrived
         try {
           resultsListener.queryCompleted(queryState);
         } catch (Exception e) {
@@ -189,9 +189,9 @@
     UserResultsListener resultsListener = queryIdToResultsListenersMap.get(queryId);
     logger.trace("For QueryId [{}], retrieved results listener {}", queryId, resultsListener);
     if (null == resultsListener) {
+
       // WHO?? didn't get query ID response and set submission listener yet,
       // so install a buffering listener for now
-
       BufferingResultsListener bl = new BufferingResultsListener();
       resultsListener = queryIdToResultsListenersMap.putIfAbsent(queryId, bl);
       // If we had a successful insertion, use that reference.  Otherwise, just
@@ -272,8 +272,7 @@
     }
 
     @Override
-    public void queryIdArrived(QueryId queryId) {
-    }
+    public void queryIdArrived(QueryId queryId) { }
   }
 
   private class SubmissionListener extends BaseRpcOutcomeListener<QueryId> {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserResultsListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserResultsListener.java
index 1d864f2..190e9ed 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserResultsListener.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserResultsListener.java
@@ -25,14 +25,21 @@
 public interface UserResultsListener {
 
   /**
-   * QueryId is available. Called when a query is successfully submitted to the server.
-   * @param queryId sent by the server along {@link org.apache.drill.exec.rpc.Acks.OK Acks.OK}
+   * QueryId is available. Called when a query is successfully submitted to the
+   * server.
+   *
+   * @param queryId
+   *          sent by the server along {@link org.apache.drill.exec.rpc.Acks.OK
+   *          Acks.OK}
    */
   void queryIdArrived(QueryId queryId);
 
   /**
-   * The query has failed. Most likely called when the server returns a FAILED query state. Can also be called if
-   * {@link #dataArrived(QueryDataBatch, ConnectionThrottle) dataArrived()} throws an exception
+   * The query has failed. Most likely called when the server returns a FAILED
+   * query state. Can also be called if
+   * {@link #dataArrived(QueryDataBatch, ConnectionThrottle) dataArrived()}
+   * throws an exception
+   *
    * @param ex exception describing the cause of the failure
    */
   void submissionFailed(UserException ex);
@@ -45,10 +52,9 @@
   void dataArrived(QueryDataBatch result, ConnectionThrottle throttle);
 
   /**
-   * The query has completed (successsful completion or cancellation). The listener will not receive any other
-   * data or result message. Called when the server returns a terminal-non failing- state (COMPLETED or CANCELLED)
-   * @param state
+   * The query has completed (successful completion or cancellation). The
+   * listener will not receive any other data or result message. Called when the
+   * server returns a terminal-non failing- state (COMPLETED or CANCELLED)
    */
   void queryCompleted(QueryState state);
-
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
index 276758e..cb1db13 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
@@ -31,6 +31,7 @@
 import org.apache.drill.common.exceptions.DrillException;
 import org.apache.drill.exec.exception.DrillbitStartupException;
 import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.physical.impl.materialize.QueryDataPackage;
 import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
 import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
 import org.apache.drill.exec.proto.GeneralRPCProtos.RpcMode;
@@ -63,6 +64,7 @@
 import org.apache.hadoop.security.HadoopKerberosName;
 import org.joda.time.DateTime;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.net.ssl.SSLEngine;
 import javax.security.sasl.SaslException;
@@ -75,7 +77,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 public class UserServer extends BasicServer<RpcType, BitToUserConnection> {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserServer.class);
+  private static final Logger logger = LoggerFactory.getLogger(UserServer.class);
   private static final String SERVER_NAME = "Apache Drill Server";
 
   private final UserConnectionConfig config;
@@ -84,7 +86,7 @@
   private final UserWorker userWorker;
   private static final ConcurrentHashMap<BitToUserConnection, BitToUserConnectionConfig> userConnectionMap;
 
-  //Initializing the singleton map during startup
+  // Initialize the singleton map during startup
   static {
     userConnectionMap = new ConcurrentHashMap<>();
   }
@@ -195,11 +197,13 @@
   }
 
   /**
-   * It represents a client connection accepted by Foreman Drillbit's UserServer from a DrillClient. This connection
-   * is used to get hold of {@link UserSession} which stores all session related information like session options
-   * changed over the lifetime of this connection. There is a 1:1 mapping between a BitToUserConnection and a
-   * UserSession. This connection object is also used to send query data and result back to the client submitted as part
-   * of the session tied to this connection.
+   * Represents a client connection accepted by Foreman Drillbit's UserServer
+   * from a DrillClient. This connection is used to get hold of
+   * {@link UserSession} which stores all session related information like
+   * session options changed over the lifetime of this connection. There is a
+   * 1:1 mapping between a BitToUserConnection and a UserSession. This
+   * connection object is also used to send query data and result back to the
+   * client submitted as part of the session tied to this connection.
    */
   public class BitToUserConnection extends AbstractServerConnection<BitToUserConnection>
       implements UserClientConnection {
@@ -240,7 +244,6 @@
      * Sets the user on the session, and finalizes the session.
      *
      * @param userName user name to set on the session
-     *
      */
     void finalizeSession(String userName) {
       // create a session
@@ -262,9 +265,10 @@
     }
 
     @Override
-    public UserSession getSession(){
-      return session;
-    }
+    public UserSession getSession() { return session; }
+
+    @Override
+    protected Logger getLogger() { return logger; }
 
     @Override
     public void sendResult(final RpcOutcomeListener<Ack> listener, final QueryResult result) {
@@ -273,17 +277,13 @@
     }
 
     @Override
-    public void sendData(final RpcOutcomeListener<Ack> listener, final QueryWritableBatch result) {
+    public void sendData(final RpcOutcomeListener<Ack> listener, final QueryDataPackage data) {
+      QueryWritableBatch result = data.toWritableBatch();
       logger.trace("Sending data to client with {}", result);
       send(listener, this, RpcType.QUERY_DATA, result.getHeader(), Ack.class, false, result.getBuffers());
     }
 
     @Override
-    protected Logger getLogger() {
-      return logger;
-    }
-
-    @Override
     public ChannelFuture getChannelClosureFuture() {
       return getChannel().closeFuture()
           .addListener(new GenericFutureListener<Future<? super Void>>() {
@@ -504,10 +504,10 @@
    * User Connection's config for System Table access
    */
   public class BitToUserConnectionConfig {
-    private DateTime established;
-    private boolean isAuthEnabled;
-    private boolean isEncryptionEnabled;
-    private boolean isSSLEnabled;
+    private final DateTime established;
+    private final boolean isAuthEnabled;
+    private final boolean isEncryptionEnabled;
+    private final boolean isSSLEnabled;
 
     public BitToUserConnectionConfig() {
       established = new DateTime(); //Current Joda-based Time
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
index 0798dea..a7594f6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserSession.java
@@ -49,11 +49,13 @@
 import org.apache.drill.exec.store.dfs.WorkspaceSchemaFactory;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class UserSession implements AutoCloseable {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserSession.class);
+  private static final Logger logger = LoggerFactory.getLogger(UserSession.class);
 
-  private boolean supportComplexTypes = false;
+  private boolean supportComplexTypes;
   private UserCredentials credentials;
   private DrillProperties properties;
   private SessionOptionManager sessionOptions;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/BaseWebUserConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/BaseWebUserConnection.java
new file mode 100644
index 0000000..42d93f2
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/BaseWebUserConnection.java
@@ -0,0 +1,57 @@
+/*
+ * 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.server.rest;
+
+import java.net.SocketAddress;
+
+import org.apache.drill.exec.rpc.AbstractDisposableUserClientConnection;
+import org.apache.drill.exec.rpc.ConnectionThrottle;
+import org.apache.drill.exec.rpc.user.UserSession;
+
+import io.netty.channel.ChannelFuture;
+
+public abstract class BaseWebUserConnection extends AbstractDisposableUserClientConnection implements ConnectionThrottle {
+
+  protected WebSessionResources webSessionResources;
+
+  public BaseWebUserConnection(WebSessionResources webSessionResources) {
+    this.webSessionResources = webSessionResources;
+  }
+
+  @Override
+  public UserSession getSession() {
+    return webSessionResources.getSession();
+  }
+
+  @Override
+  public ChannelFuture getChannelClosureFuture() {
+    return webSessionResources.getCloseFuture();
+  }
+
+  @Override
+  public SocketAddress getRemoteAddress() {
+    return webSessionResources.getRemoteAddress();
+  }
+
+  @Override
+  public void setAutoRead(boolean enableAutoRead) { }
+
+  public WebSessionResources resources() {
+    return webSessionResources;
+  }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRestServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRestServer.java
index 141c027..c345571 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRestServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRestServer.java
@@ -237,9 +237,7 @@
     }
 
     @Override
-    public void dispose(WebUserConnection instance) {
-
-    }
+    public void dispose(WebUserConnection instance) { }
   }
 
   public static class AnonWebUserConnectionProvider implements Factory<WebUserConnection> {
@@ -300,14 +298,14 @@
     }
 
     @Override
-    public void dispose(WebUserConnection instance) {
-
-    }
+    public void dispose(WebUserConnection instance) { }
 
     /**
-     * Creates session user principal. If impersonation is enabled without authentication and User-Name header is present and valid,
-     * will create session user principal with provided user name, otherwise anonymous user name will be used.
-     * In both cases session user principal will have admin rights.
+     * Creates session user principal. If impersonation is enabled without
+     * authentication and User-Name header is present and valid, will create
+     * session user principal with provided user name, otherwise anonymous user
+     * name will be used. In both cases session user principal will have admin
+     * rights.
      *
      * @param config drill config
      * @param request client request
@@ -322,10 +320,12 @@
       }
       return new AnonDrillUserPrincipal();
     }
-
   }
 
-  // Provider which injects DrillUserPrincipal directly instead of getting it from SecurityContext and typecasting
+  /**
+   * Provider which injects DrillUserPrincipal directly instead of getting it
+   * from SecurityContext and typecasting
+   */
   public static class DrillUserPrincipalProvider implements Factory<DrillUserPrincipal> {
 
     @Inject HttpServletRequest request;
@@ -336,9 +336,7 @@
     }
 
     @Override
-    public void dispose(DrillUserPrincipal principal) {
-      // No-Op
-    }
+    public void dispose(DrillUserPrincipal principal) { }
   }
 
   // Provider which creates and cleanups DrillUserPrincipal for anonymous (auth disabled) mode
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/LogsResources.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/LogsResources.java
index 4aa2061..e29050a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/LogsResources.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/LogsResources.java
@@ -150,7 +150,7 @@
   private File getFileByName(File folder, final String name) {
     File[] files = folder.listFiles((dir, fileName) -> fileName.equals(name));
     if (files.length == 0) {
-      throw new DrillRuntimeException (name + " doesn't exist");
+      throw new DrillRuntimeException(name + " doesn't exist");
     }
     return files[0];
   }
@@ -159,9 +159,9 @@
   @XmlRootElement
   public class Log implements Comparable<Log> {
 
-    private String name;
-    private long size;
-    private DateTime lastModified;
+    private final String name;
+    private final long size;
+    private final DateTime lastModified;
 
     @JsonCreator
     public Log (@JsonProperty("name") String name, @JsonProperty("size") long size, @JsonProperty("lastModified") long lastModified) {
@@ -190,9 +190,9 @@
 
   @XmlRootElement
   public class LogContent {
-    private String name;
-    private Collection<String> lines;
-    private int maxLines;
+    private final String name;
+    private final Collection<String> lines;
+    private final int maxLines;
 
     @JsonCreator
     public LogContent (@JsonProperty("name") String name, @JsonProperty("lines") Collection<String> lines, @JsonProperty("maxLines") int maxLines) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/QueryWrapper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/QueryWrapper.java
index 94fcea9..be4c331 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/QueryWrapper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/QueryWrapper.java
@@ -22,6 +22,7 @@
 import javax.xml.bind.annotation.XmlRootElement;
 
 import org.apache.drill.common.PlanStringBuilder;
+import org.apache.drill.exec.proto.UserBitShared.QueryType;
 import org.apache.drill.shaded.guava.com.google.common.base.CharMatcher;
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
 import org.apache.parquet.Strings;
@@ -98,7 +99,7 @@
   public static final class RestQueryBuilder {
 
     private String query;
-    private String queryType = "SQL";
+    private String queryType = QueryType.SQL.name();
     private int rowLimit;
     private String userName;
     private String defaultSchema;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/RestQueryRunner.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/RestQueryRunner.java
index e944386..69d81ae 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/RestQueryRunner.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/RestQueryRunner.java
@@ -62,7 +62,7 @@
     this.options = webUserConnection.getSession().getOptions();
   }
 
-  public RestQueryRunner.QueryResult run() throws Exception {
+  public QueryResult run() throws Exception {
     applyUserName();
     applyOptions();
     applyDefaultSchema();
@@ -131,7 +131,7 @@
     return maxRows;
   }
 
-  public RestQueryRunner.QueryResult submitQuery(int maxRows) {
+  public QueryResult submitQuery(int maxRows) {
     final RunQuery runQuery = RunQuery.newBuilder()
         .setType(QueryType.valueOf(query.getQueryType()))
         .setPlan(query.getQuery())
@@ -161,7 +161,7 @@
       }
     } while (!isComplete && !nearlyOutOfHeapSpace);
 
-    //Fail if nearly out of heap space
+    // Fail if nearly out of heap space
     if (nearlyOutOfHeapSpace) {
       UserException almostOutOfHeapException = UserException.resourceError()
           .message("There is not enough heap memory to run this query using the web interface. ")
@@ -185,7 +185,7 @@
     return new QueryResult(queryId, webUserConnection, webUserConnection.results);
   }
 
-  //Detect possible excess heap
+  // Detect possible excess heap
   private float getHeapUsage() {
     return (float) memMXBean.getHeapMemoryUsage().getUsed() / memMXBean.getHeapMemoryUsage().getMax();
   }
@@ -198,15 +198,16 @@
     public final String queryState;
     public final int attemptedAutoLimit;
 
-    //DRILL-6847:  Modified the constructor so that the method has access to all the properties in webUserConnection
+    // DRILL-6847:  Modified the constructor so that the method has access
+    // to all the properties in webUserConnection
     public QueryResult(QueryId queryId, WebUserConnection webUserConnection, List<Map<String, String>> rows) {
-        this.queryId = QueryIdHelper.getQueryId(queryId);
-        this.columns = webUserConnection.columns;
-        this.metadata = webUserConnection.metadata;
-        this.queryState = webUserConnection.getQueryState();
-        this.rows = rows;
-        this.attemptedAutoLimit = webUserConnection.getAutoLimitRowCount();
-      }
+      this.queryId = QueryIdHelper.getQueryId(queryId);
+      this.columns = webUserConnection.columns;
+      this.metadata = webUserConnection.metadata;
+      this.queryState = webUserConnection.getQueryState();
+      this.rows = rows;
+      this.attemptedAutoLimit = webUserConnection.getAutoLimitRowCount();
+    }
 
     public String getQueryId() {
       return queryId;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
index 6819457..e00cb48 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
@@ -262,12 +262,11 @@
     responseHeadersSettingFilter.setInitParameters(ResponseHeadersSettingFilter.retrieveResponseHeaders(config));
     servletContextHandler.addFilter(responseHeadersSettingFilter, "/*", EnumSet.of(DispatcherType.REQUEST));
 
-
     return servletContextHandler;
   }
 
   /**
-   * It creates A {@link SessionHandler} which contains a {@link HashSessionManager}
+   * Create a {@link SessionHandler} which contains a {@link HashSessionManager}
    *
    * @param securityHandler Set of init parameters that are used by the Authentication
    * @return session handler
@@ -279,9 +278,7 @@
     sessionManager.getSessionCookieConfig().setHttpOnly(true);
     sessionManager.addEventListener(new HttpSessionListener() {
       @Override
-      public void sessionCreated(HttpSessionEvent se) {
-
-      }
+      public void sessionCreated(HttpSessionEvent se) { }
 
       @Override
       public void sessionDestroyed(HttpSessionEvent se) {
@@ -338,8 +335,9 @@
   }
 
   /**
-   * Create an HTTPS connector for given jetty server instance. If the admin has specified keystore/truststore settings
-   * they will be used else a self-signed certificate is generated and used.
+   * Create an HTTPS connector for given jetty server instance. If the admin has
+   * specified keystore/truststore settings they will be used else a self-signed
+   * certificate is generated and used.
    *
    * @return Initialized {@link ServerConnector} for HTTPS connections.
    */
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebSessionResources.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebSessionResources.java
index 016278f..c06770e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebSessionResources.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebSessionResources.java
@@ -22,22 +22,23 @@
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.rpc.ChannelClosedException;
 import org.apache.drill.exec.rpc.user.UserSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.net.SocketAddress;
 
 /**
- * Class holding all the resources required for Web User Session. This class is responsible for the proper cleanup of
- * all the resources.
+ * Holds the resources required for Web User Session. This class is responsible
+ * for the proper cleanup of all the resources.
  */
 public class WebSessionResources implements AutoCloseable {
+  private static final Logger logger = LoggerFactory.getLogger(WebSessionResources.class);
 
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(WebSessionResources.class);
-
-  private BufferAllocator allocator;
+  private final BufferAllocator allocator;
 
   private final SocketAddress remoteAddress;
 
-  private UserSession webUserSession;
+  private final UserSession webUserSession;
 
   private ChannelPromise closeFuture;
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebUserConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebUserConnection.java
index 0ca6abe..9427fc3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebUserConnection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebUserConnection.java
@@ -17,177 +17,128 @@
  */
 package org.apache.drill.exec.server.rest;
 
-import org.apache.drill.exec.ExecConstants;
 import org.apache.drill.exec.util.ValueVectorElementFormatter;
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
 import org.apache.drill.shaded.guava.com.google.common.collect.Sets;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.DrillBuf;
-import io.netty.channel.ChannelFuture;
-import org.apache.drill.common.exceptions.UserException;
+
 import org.apache.drill.common.types.TypeProtos;
-import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
+import org.apache.drill.exec.physical.impl.materialize.QueryDataPackage;
 import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
-import org.apache.drill.exec.record.RecordBatchLoader;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.VectorContainer;
 import org.apache.drill.exec.record.VectorWrapper;
-import org.apache.drill.exec.rpc.AbstractDisposableUserClientConnection;
 import org.apache.drill.exec.rpc.Acks;
-import org.apache.drill.exec.rpc.ConnectionThrottle;
 import org.apache.drill.exec.rpc.RpcOutcomeListener;
-import org.apache.drill.exec.rpc.user.UserSession;
 import org.apache.drill.exec.vector.ValueVector.Accessor;
+import org.apache.drill.exec.record.BatchSchema;
 import org.apache.drill.exec.record.MaterializedField;
 
-import java.net.SocketAddress;
 import java.util.List;
 import java.util.Map;
 import java.util.ArrayList;
 import java.util.Set;
 
 /**
- * WebUserConnectionWrapper which represents the UserClientConnection between WebServer and Foreman, for the WebUser
- * submitting the query. It provides access to the UserSession executing the query. There is no actual physical
+ * {@code WebUserConnectionWrapper} which represents the {@code UserClientConnection} between
+ * WebServer and Foreman, for the WebUser submitting the query. It provides
+ * access to the {@code UserSession} executing the query. There is no actual physical
  * channel corresponding to this connection wrapper.
  *
- * It returns a close future with no actual underlying {@link io.netty.channel.Channel} associated with it but do have an
- * EventExecutor out of BitServer EventLoopGroup. Since there is no actual connection established using this class,
- * hence the close event will never be fired by underlying layer and close future is set only when the
+ * It returns a close future with no actual underlying
+ * {@link io.netty.channel.Channel} associated with it but do have an
+ * {@code EventExecutor} out of BitServer EventLoopGroup. Since there is no actual
+ * connection established using this class, hence the close event will never be
+ * fired by underlying layer and close future is set only when the
  * {@link WebSessionResources} are closed.
  */
-
-public class WebUserConnection extends AbstractDisposableUserClientConnection implements ConnectionThrottle {
-
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(WebUserConnection.class);
-
-  protected WebSessionResources webSessionResources;
+public class WebUserConnection extends BaseWebUserConnection {
 
   public final List<Map<String, String>> results = Lists.newArrayList();
-
   public final Set<String> columns = Sets.newLinkedHashSet();
-
   public final List<String> metadata = new ArrayList<>();
-
   private int autoLimitRowCount;
+  private int rowCount;
 
   WebUserConnection(WebSessionResources webSessionResources) {
-    this.webSessionResources = webSessionResources;
+    super(webSessionResources);
   }
 
   @Override
-  public UserSession getSession() {
-    return webSessionResources.getSession();
+  public void sendData(RpcOutcomeListener<Ack> listener, QueryDataPackage data) {
+    processBatch(data.batch());
+    listener.success(Acks.OK, null);
   }
 
-  @Override
-  public void sendData(RpcOutcomeListener<Ack> listener, QueryWritableBatch result) {
-    // There can be overflow here but DrillBuf doesn't support allocating with
-    // bytes in long. Hence we are just preserving the earlier behavior and logging debug log for the case.
-    final int dataByteCount = (int) result.getByteCount();
-
-    if (dataByteCount < 0) {
-      if (logger.isDebugEnabled()) {
-        logger.debug("There is BufferOverflow in dataByteCount: {}",
-            dataByteCount);
-      }
-      listener.success(Acks.OK, null);
+  private void processBatch(VectorContainer batch) {
+    if (batch == null) {
+      // Empty query: no data, no schema.
       return;
     }
 
-    // Create a ByteBuf with all the data in it.
-    final int rows = result.getHeader().getRowCount();
-    final BufferAllocator allocator = webSessionResources.getAllocator();
-    final DrillBuf bufferWithData = allocator.buffer(dataByteCount);
-    try {
-      final ByteBuf[] resultDataBuffers = result.getBuffers();
+    // Build metadata only on first batch, or if the schema changes
+    if (metadata.isEmpty() || batch.isSchemaChanged()) {
+      columns.clear();
+      metadata.clear();
+      buildMetadata(batch.getSchema());
+    }
+    addResults(batch.getRecordCount(), batch);
+    batch.zeroVectors();
+  }
 
-      for (final ByteBuf buffer : resultDataBuffers) {
-        bufferWithData.writeBytes(buffer);
-        buffer.release();
-      }
+  private void buildMetadata(BatchSchema schema) {
+    for (int i = 0; i < schema.getFieldCount(); ++i) {
+      // DRILL-6847:  This section adds query metadata to the REST results
+      MaterializedField col = schema.getColumn(i);
+      columns.add(col.getName());
+      StringBuilder dataType = new StringBuilder(col.getType().getMinorType().name());
 
-      final RecordBatchLoader loader = new RecordBatchLoader(allocator);
-      try {
-        loader.load(result.getHeader().getDef(), bufferWithData);
-        // TODO:  Clean:  DRILL-2933:  That load(...) no longer throws
-        // SchemaChangeException, so check/clean catch clause below.
-        for (int i = 0; i < loader.getSchema().getFieldCount(); ++i) {
-          //DRILL-6847:  This section adds query metadata to the REST results
-          MaterializedField col = loader.getSchema().getColumn(i);
-          columns.add(col.getName());
-          StringBuilder dataType = new StringBuilder(col.getType().getMinorType().name());
+      // For DECIMAL type
+      if (col.getType().hasPrecision()) {
+        dataType.append("(");
+        dataType.append(col.getType().getPrecision());
 
-          //For DECIMAL type
-          if (col.getType().hasPrecision()) {
-            dataType.append("(");
-            dataType.append(col.getType().getPrecision());
-
-            if (col.getType().hasScale()) {
-              dataType.append(", ");
-              dataType.append(col.getType().getScale());
-            }
-
-            dataType.append(")");
-          } else if (col.getType().hasWidth()) {
-            //Case for VARCHAR columns with specified width
-            dataType.append("(");
-            dataType.append(col.getType().getWidth());
-            dataType.append(")");
-          }
-          metadata.add(dataType.toString());
+        if (col.getType().hasScale()) {
+          dataType.append(", ");
+          dataType.append(col.getType().getScale());
         }
-        ValueVectorElementFormatter formatter = new ValueVectorElementFormatter(webSessionResources.getSession().getOptions());
-        for (int i = 0; i < rows; ++i) {
-          final Map<String, String> record = Maps.newHashMap();
-          for (VectorWrapper<?> vw : loader) {
-            final String field = vw.getValueVector().getMetadata().getNamePart().getName();
-            final TypeProtos.MinorType fieldMinorType = vw.getValueVector().getMetadata().getMajorType().getMinorType();
-            final Accessor accessor = vw.getValueVector().getAccessor();
-            final Object value = i < accessor.getValueCount() ? accessor.getObject(i) : null;
-            final String display = value == null ? null : formatter.format(value, fieldMinorType);
-            record.put(field, display);
-          }
-          results.add(record);
-        }
-      } finally {
-        loader.clear();
+
+        dataType.append(")");
+      } else if (col.getType().hasWidth()) {
+        // Case for VARCHAR columns with specified width
+        dataType.append("(");
+        dataType.append(col.getType().getWidth());
+        dataType.append(")");
       }
-    } catch (Exception e) {
-      boolean verbose = webSessionResources.getSession().getOptions().getBoolean(ExecConstants.ENABLE_VERBOSE_ERRORS_KEY);
-      // Wrapping the exception into UserException and then into DrillPBError.
-      // It will be thrown as exception in QueryWrapper class.
-      // It's verbosity depends on system option "exec.errors.verbose".
-      error = UserException.systemError(e).build(logger).getOrCreatePBError(verbose);
-    } finally {
-      // Notify the listener with ACK.OK both in error/success case because data was send successfully from Drillbit.
-      bufferWithData.release();
-      listener.success(Acks.OK, null);
+      metadata.add(dataType.toString());
     }
   }
 
-  @Override
-  public ChannelFuture getChannelClosureFuture() {
-    return webSessionResources.getCloseFuture();
-  }
-
-  @Override
-  public SocketAddress getRemoteAddress() {
-    return webSessionResources.getRemoteAddress();
-  }
-
-  @Override
-  public void setAutoRead(boolean enableAutoRead) {
-    // no-op
+  private void addResults(int rows, VectorAccessible batch) {
+    ValueVectorElementFormatter formatter = new ValueVectorElementFormatter(webSessionResources.getSession().getOptions());
+    if (autoLimitRowCount > 0) {
+      rows = Math.max(0, Math.min(rows, autoLimitRowCount - rowCount));
+    }
+    for (int i = 0; i < rows; ++i) {
+      rowCount++;
+      final Map<String, String> record = Maps.newHashMap();
+      for (VectorWrapper<?> vw : batch) {
+        final String field = vw.getValueVector().getMetadata().getNamePart().getName();
+        final TypeProtos.MinorType fieldMinorType = vw.getValueVector().getMetadata().getMajorType().getMinorType();
+        final Accessor accessor = vw.getValueVector().getAccessor();
+        final Object value = i < accessor.getValueCount() ? accessor.getObject(i) : null;
+        final String display = value == null ? null : formatter.format(value, fieldMinorType);
+        record.put(field, display);
+      }
+      results.add(record);
+    }
   }
 
   /**
    * For authenticated WebUser no cleanup of {@link WebSessionResources} is done since it's re-used
    * for all the queries until lifetime of the web session.
    */
-  public void cleanupSession() {
-    // no-op
-  }
+  public void cleanupSession() { }
 
   public static class AnonWebUserConnection extends WebUserConnection {
 
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
index ac1b4cc..de49e28 100644
--- 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
@@ -21,6 +21,8 @@
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.apache.drill.common.SelfCleaningRunnable;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.exec.coord.ClusterCoordinator;
@@ -60,11 +62,11 @@
 import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * Manages the running fragments in a Drillbit. Periodically requests run-time stats updates from fragments
- * running elsewhere.
+ * Manages the running fragments in a Drillbit. Periodically requests run-time
+ * stats updates from fragments running elsewhere.
  */
 public class WorkManager implements AutoCloseable {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(WorkManager.class);
+  private static final Logger logger = LoggerFactory.getLogger(WorkManager.class);
 
   private static final int EXIT_TIMEOUT_MS = 5000;
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/prepare/PreparedStatementProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/prepare/PreparedStatementProvider.java
index ef018f6..5de560a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/prepare/PreparedStatementProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/prepare/PreparedStatementProvider.java
@@ -18,14 +18,14 @@
 package org.apache.drill.exec.work.prepare;
 
 import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableMap;
-import io.netty.buffer.ByteBuf;
+
 import io.netty.channel.ChannelFuture;
 import org.apache.drill.common.exceptions.ErrorHelper;
 import org.apache.drill.common.types.TypeProtos.DataMode;
 import org.apache.drill.common.types.TypeProtos.MajorType;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
-import org.apache.drill.exec.physical.impl.materialize.QueryWritableBatch;
+import org.apache.drill.exec.physical.impl.materialize.QueryDataPackage;
 import org.apache.drill.exec.proto.ExecProtos.ServerPreparedStatementState;
 import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
 import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
@@ -43,6 +43,7 @@
 import org.apache.drill.exec.proto.UserProtos.ResultColumnMetadata;
 import org.apache.drill.exec.proto.UserProtos.RpcType;
 import org.apache.drill.exec.proto.UserProtos.RunQuery;
+import org.apache.drill.exec.record.VectorContainer;
 import org.apache.drill.exec.rpc.AbstractDisposableUserClientConnection;
 import org.apache.drill.exec.rpc.Acks;
 import org.apache.drill.exec.rpc.Response;
@@ -53,6 +54,8 @@
 import org.apache.drill.exec.store.ischema.InfoSchemaConstants;
 import org.apache.drill.exec.work.user.UserWorker;
 import org.joda.time.Period;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.math.BigDecimal;
 import java.net.SocketAddress;
@@ -72,11 +75,11 @@
  * Contains worker {@link Runnable} for creating a prepared statement and helper methods.
  */
 public class PreparedStatementProvider {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PreparedStatementProvider.class);
+  private static final Logger logger = LoggerFactory.getLogger(PreparedStatementProvider.class);
 
   /**
-   * Static list of mappings from {@link MinorType} to JDBC ResultSet class name (to be returned through
-   * {@link ResultSetMetaData#getColumnClassName(int)}.
+   * Static list of mappings from {@link MinorType} to JDBC ResultSet class name
+   * (to be returned through {@link ResultSetMetaData#getColumnClassName(int)}.
    */
   private static final Map<MinorType, String> DRILL_TYPE_TO_JDBC_CLASSNAME = ImmutableMap.<MinorType, String>builder()
       .put(MinorType.INT, Integer.class.getName())
@@ -186,9 +189,10 @@
   }
 
   /**
-   * Helper method to create {@link DrillPBError} and set it in <code>respBuilder</code>
+   * Helper method to create {@link DrillPBError} and set it in {@code respBuilder}
    */
-  private static void setErrorHelper(final CreatePreparedStatementResp.Builder respBuilder, final RequestStatus status,
+  private static void setErrorHelper(final CreatePreparedStatementResp.Builder respBuilder,
+      final RequestStatus status,
       final Throwable ex, final String message, final ErrorType errorType) {
     respBuilder.setStatus(status);
     final String errorId = UUID.randomUUID().toString();
@@ -211,7 +215,7 @@
   }
 
   /**
-   * Helper method to log error and set given {@link DrillPBError} in <code>respBuilder</code>
+   * Helper method to log error and set given {@link DrillPBError} in {@code respBuilder}
    */
   private static void setErrorHelper(final CreatePreparedStatementResp.Builder respBuilder, final DrillPBError error,
       final String message) {
@@ -250,14 +254,14 @@
     }
 
     @Override
-    public void sendData(RpcOutcomeListener<Ack> listener, QueryWritableBatch result) {
+    public void sendData(RpcOutcomeListener<Ack> listener, QueryDataPackage data) {
       // Save the query results schema and release the buffers.
-      if (fields == null) {
-        fields = result.getHeader().getDef().getFieldList();
-      }
-
-      for (ByteBuf buf : result.getBuffers()) {
-        buf.release();
+      VectorContainer batch = data.batch();
+      if (batch != null) {
+        if (fields == null) {
+          fields = data.fields();
+        }
+        batch.zeroVectors();
       }
 
       listener.success(Acks.OK, null);
@@ -287,16 +291,18 @@
     builder.setCatalogName(InfoSchemaConstants.IS_CATALOG_NAME);
 
     /**
-     * Designated column's schema name. Empty string if not applicable. Initial implementation defaults to empty string
-     * as we use LIMIT 0 queries to get the schema and schema info is lost. If we derive the schema from plan, we may
-     * get the right value.
+     * Designated column's schema name. Empty string if not applicable. Initial
+     * implementation defaults to empty string as we use LIMIT 0 queries to get
+     * the schema and schema info is lost. If we derive the schema from plan, we
+     * may get the right value.
      */
     builder.setSchemaName("");
 
     /**
-     * Designated column's table name. Not set if not applicable. Initial implementation defaults to empty string as
-     * we use LIMIT 0 queries to get the schema and table info is lost. If we derive the table from plan, we may get
-     * the right value.
+     * Designated column's table name. Not set if not applicable. Initial
+     * implementation defaults to empty string as we use LIMIT 0 queries to get
+     * the schema and table info is lost. If we derive the table from plan, we
+     * may get the right value.
      */
     builder.setTableName("");
 
@@ -327,7 +333,8 @@
     builder.setPrecision(Types.getPrecision(field.getMajorType()));
 
     /**
-     * Column's number of digits to right of the decimal point. 0 is returned for types where the scale is not applicable
+     * Column's number of digits to right of the decimal point. 0 is returned
+     * for types where the scale is not applicable
      */
     builder.setScale(Types.getScale(majorType));
 
@@ -342,8 +349,8 @@
     builder.setDisplaySize(Types.getJdbcDisplaySize(majorType));
 
     /**
-     * Is the column an aliased column. Initial implementation defaults to true as we derive schema from LIMIT 0 query and
-     * not plan
+     * Is the column an aliased column. Initial implementation defaults to true
+     * as we derive schema from LIMIT 0 query and not plan
      */
     builder.setIsAliased(true);
 
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
index 9c32b56..976820d 100644
--- 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
@@ -41,9 +41,11 @@
 import org.apache.drill.exec.work.metadata.MetadataProvider;
 import org.apache.drill.exec.work.metadata.ServerMetaProvider.ServerMetaWorker;
 import org.apache.drill.exec.work.prepare.PreparedStatementProvider.PreparedStatementWorker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class UserWorker{
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UserWorker.class);
+  static final Logger logger = LoggerFactory.getLogger(UserWorker.class);
 
   private final WorkerBee bee;
   private final QueryCountIncrementer incrementer = new QueryCountIncrementer() {
diff --git a/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java b/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java
index 3fc5f7c..9e90d06 100644
--- a/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java
+++ b/exec/memory/base/src/main/java/io/netty/buffer/DrillBuf.java
@@ -42,8 +42,9 @@
 import io.netty.util.internal.PlatformDependent;
 
 /**
- * Drill data structure for accessing and manipulating data buffers. This class is integrated with the
- * Drill memory management layer for quota enforcement and buffer sharing.
+ * Drill data structure for accessing and manipulating data buffers. This class
+ * is integrated with the Drill memory management layer for quota enforcement
+ * and buffer sharing.
  */
 @SuppressWarnings("unused")
 public final class DrillBuf extends AbstractByteBuf implements AutoCloseable {
@@ -121,18 +122,22 @@
   }
 
   /**
-   * Create a new DrillBuf that is associated with an alternative allocator for the purposes of memory ownership and
-   * accounting. This has no impact on the reference counting for the current DrillBuf except in the situation where the
-   * passed in Allocator is the same as the current buffer.
+   * Create a new DrillBuf that is associated with an alternative allocator for
+   * the purposes of memory ownership and accounting. This has no impact on the
+   * reference counting for the current DrillBuf except in the situation where
+   * the passed in Allocator is the same as the current buffer.
    *
-   * This operation has no impact on the reference count of this DrillBuf. The newly created DrillBuf with either have a
-   * reference count of 1 (in the case that this is the first time this memory is being associated with the new
-   * allocator) or the current value of the reference count + 1 for the other AllocationManager/BufferLedger combination
-   * in the case that the provided allocator already had an association to this underlying memory.
+   * This operation has no impact on the reference count of this DrillBuf. The
+   * newly created DrillBuf with either have a reference count of 1 (in the case
+   * that this is the first time this memory is being associated with the new
+   * allocator) or the current value of the reference count + 1 for the other
+   * AllocationManager/BufferLedger combination in the case that the provided
+   * allocator already had an association to this underlying memory.
    *
    * @param target
    *          The target allocator to create an association with.
-   * @return A new DrillBuf which shares the same underlying memory as this DrillBuf.
+   * @return A new DrillBuf which shares the same underlying memory as this
+   *         DrillBuf.
    */
   public DrillBuf retain(BufferAllocator target) {
 
@@ -148,28 +153,35 @@
   }
 
   /**
-   * Transfer the memory accounting ownership of this DrillBuf to another allocator. This will generate a new DrillBuf
-   * that carries an association with the underlying memory of this DrillBuf. If this DrillBuf is connected to the
-   * owning BufferLedger of this memory, that memory ownership/accounting will be transferred to the target allocator. If
-   * this DrillBuf does not currently own the memory underlying it (and is only associated with it), this does not
-   * transfer any ownership to the newly created DrillBuf.
+   * Transfer the memory accounting ownership of this DrillBuf to another
+   * allocator. This will generate a new DrillBuf that carries an association
+   * with the underlying memory of this DrillBuf. If this DrillBuf is connected
+   * to the owning BufferLedger of this memory, that memory ownership/accounting
+   * will be transferred to the target allocator. If this DrillBuf does not
+   * currently own the memory underlying it (and is only associated with it),
+   * this does not transfer any ownership to the newly created DrillBuf.
    * <p>
-   * This operation has no impact on the reference count of this DrillBuf. The newly created DrillBuf with either have a
-   * reference count of 1 (in the case that this is the first time this memory is being associated with the new
-   * allocator) or the current value of the reference count for the other AllocationManager/BufferLedger combination in
-   * the case that the provided allocator already had an association to this underlying memory.
+   * This operation has no impact on the reference count of this DrillBuf. The
+   * newly created DrillBuf with either have a reference count of 1 (in the case
+   * that this is the first time this memory is being associated with the new
+   * allocator) or the current value of the reference count for the other
+   * AllocationManager/BufferLedger combination in the case that the provided
+   * allocator already had an association to this underlying memory.
    * <p>
-   * Transfers will always succeed, even if that puts the other allocator into an overlimit situation. This is possible
-   * due to the fact that the original owning allocator may have allocated this memory out of a local reservation
-   * whereas the target allocator may need to allocate new memory from a parent or RootAllocator. This operation is done
-   * in a mostly-lockless but consistent manner. As such, the overlimit==true situation could occur slightly prematurely
-   * to an actual overlimit==true condition. This is simply conservative behavior which means we may return overlimit
-   * slightly sooner than is necessary.
+   * Transfers will always succeed, even if that puts the other allocator into
+   * an overlimit situation. This is possible due to the fact that the original
+   * owning allocator may have allocated this memory out of a local reservation
+   * whereas the target allocator may need to allocate new memory from a parent
+   * or RootAllocator. This operation is done in a mostly-lockless but
+   * consistent manner. As such, the overlimit==true situation could occur
+   * slightly prematurely to an actual overlimit==true condition. This is simply
+   * conservative behavior which means we may return overlimit slightly sooner
+   * than is necessary.
    *
    * @param target
    *          The allocator to transfer ownership to.
-   * @return A new transfer result with the impact of the transfer (whether it was overlimit) as well as the newly
-   *         created DrillBuf.
+   * @return A new transfer result with the impact of the transfer (whether it
+   *         was overlimit) as well as the newly created DrillBuf.
    */
   public TransferResult transferOwnership(BufferAllocator target) {
 
diff --git a/exec/memory/base/src/main/java/org/apache/drill/exec/memory/AllocationManager.java b/exec/memory/base/src/main/java/org/apache/drill/exec/memory/AllocationManager.java
index bfce9a0..36550d0 100644
--- a/exec/memory/base/src/main/java/org/apache/drill/exec/memory/AllocationManager.java
+++ b/exec/memory/base/src/main/java/org/apache/drill/exec/memory/AllocationManager.java
@@ -43,11 +43,11 @@
  * UDLE. Ensures that one allocator owns the memory that multiple allocators may
  * be referencing. Manages a BufferLedger between each of its associated
  * allocators. This class is also responsible for managing when memory is
- * allocated and returned to the Netty-based PooledByteBufAllocatorL.
- *
+ * allocated and returned to the Netty-based {code PooledByteBufAllocatorL}.
+ * <p>
  * The only reason that this isn't package private is we're forced to put
  * DrillBuf in Netty's package which need access to these objects or methods.
- *
+ * <p>
  * Threading: AllocationManager manages thread-safety internally. Operations
  * within the context of a single BufferLedger are lockless in nature and can be
  * leveraged by multiple threads. Operations that cross the context of two
@@ -56,7 +56,6 @@
  * allocation. As such, there will be thousands of these in a typical query. The
  * contention of acquiring a lock on AllocationManager should be very low.
  */
-
 public class AllocationManager {
 
   private static final AtomicLong MANAGER_ID_GENERATOR = new AtomicLong(0);
@@ -151,7 +150,6 @@
    * AllocationManager that it now longer needs to hold a reference to
    * particular piece of memory.
    */
-
   private class ReleaseListener {
 
     private final BufferAllocator allocator;
@@ -163,7 +161,6 @@
     /**
      * Can only be called when you already hold the writeLock.
      */
-
     public void release() {
       allocator.assertOpen();
 
@@ -200,7 +197,6 @@
    * only reason this is public is due to DrillBuf being in io.netty.buffer
    * package.
    */
-
   public class BufferLedger {
 
     private final IdentityHashMap<DrillBuf, Object> buffers =
@@ -223,8 +219,10 @@
     }
 
     /**
-     * Transfer any balance the current ledger has to the target ledger. In the case that the current ledger holds no
-     * memory, no transfer is made to the new ledger.
+     * Transfer any balance the current ledger has to the target ledger. In the
+     * case that the current ledger holds no memory, no transfer is made to the
+     * new ledger.
+     *
      * @param target
      *          The ledger to transfer ownership account to.
      * @return Whether transfer fit within target ledgers limits.
@@ -241,8 +239,8 @@
         return true;
       }
 
-      // since two balance transfers out from the allocator manager could cause incorrect accounting, we need to ensure
-      // that this won't happen by synchronizing on the allocator manager instance.
+      // since two balance transfers out from the allocator manager could cause incorrect accounting,
+      // we need to ensure that this won't happen by synchronizing on the allocator manager instance.
       try (@SuppressWarnings("unused") Closeable write = writeLock.open()) {
         if (owningLedger != this) {
           return true;
@@ -316,7 +314,6 @@
      * zero, this ledger should release its ownership back to the
      * AllocationManager
      */
-
     public int decrement(int decrement) {
       allocator.assertOpen();
 
@@ -346,7 +343,6 @@
      * @param allocator
      * @return The ledger associated with a particular BufferAllocator.
      */
-
     public BufferLedger getLedgerForAllocator(BufferAllocator allocator) {
       return associate((BaseAllocator) allocator);
     }
@@ -362,7 +358,6 @@
      * @return A new DrillBuf that shares references with all DrillBufs
      *         associated with this BufferLedger
      */
-
     public DrillBuf newDrillBuf(int offset, int length) {
       allocator.assertOpen();
       return newDrillBuf(offset, length, null);
@@ -411,7 +406,6 @@
      *
      * @return Size in bytes
      */
-
     public int getSize() {
       return size;
     }
diff --git a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BaseRpcOutcomeListener.java b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BaseRpcOutcomeListener.java
index 8514df0..9498cc8 100644
--- a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BaseRpcOutcomeListener.java
+++ b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BaseRpcOutcomeListener.java
@@ -20,20 +20,16 @@
 import io.netty.buffer.ByteBuf;
 
 public class BaseRpcOutcomeListener<T> implements RpcOutcomeListener<T> {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BaseRpcOutcomeListener.class);
 
   @Override
-  public void failed(RpcException ex) {
-  }
+  public void failed(RpcException ex) { }
 
   @Override
-  public void success(T value, ByteBuf buffer) {
-  }
+  public void success(T value, ByteBuf buffer) { }
 
   /**
    * {@inheritDoc}
    */
   @Override
-  public void interrupted(final InterruptedException ex) {
-  }
+  public void interrupted(final InterruptedException ex) { }
 }
diff --git a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RequestIdMap.java b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RequestIdMap.java
index 83380e2..cc2d9b9 100644
--- a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RequestIdMap.java
+++ b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RequestIdMap.java
@@ -23,6 +23,8 @@
 import org.apache.drill.common.exceptions.UserRemoteException;
 import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.carrotsearch.hppc.IntObjectHashMap;
 import com.carrotsearch.hppc.procedures.IntObjectProcedure;
@@ -38,7 +40,7 @@
  * else works via Atomic variables.
  */
 class RequestIdMap {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RequestIdMap.class);
+  private static final Logger logger = LoggerFactory.getLogger(RequestIdMap.class);
 
   private final AtomicInteger lastCoordinationId = new AtomicInteger();
   private final AtomicBoolean isOpen = new AtomicBoolean(true);
@@ -72,9 +74,9 @@
 
     @Override
     public void apply(int key, RpcOutcome<?> value) {
-      try{
+      try {
         value.setException(exception);
-      }catch (final Exception e){
+      } catch (final Exception e){
         logger.warn("Failure while attempting to fail rpc response.", e);
       }
     }
@@ -158,6 +160,7 @@
     return rpc;
   }
 
+  @SuppressWarnings("unchecked")
   public <V> RpcOutcome<V> getAndRemoveRpcOutcome(int rpcType, int coordinationId, Class<V> clazz) {
 
     final RpcOutcome<?> rpc = removeFromMap(coordinationId);
@@ -172,16 +175,10 @@
           clazz.getCanonicalName(), rpcType, outcomeClass.getCanonicalName()));
     }
 
-    @SuppressWarnings("unchecked")
-    final
-    RpcOutcome<V> crpc = (RpcOutcome<V>) rpc;
-
-    // logger.debug("Returning casted future");
-    return crpc;
+    return (RpcOutcome<V>) rpc;
   }
 
   public void recordRemoteFailure(int coordinationId, DrillPBError failure) {
-    // logger.debug("Updating failed future.");
     try {
       final RpcOutcome<?> rpc = removeFromMap(coordinationId);
       rpc.setException(new UserRemoteException(failure));
diff --git a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcOutcomeListener.java b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcOutcomeListener.java
index a1c125b..4afa159 100644
--- a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcOutcomeListener.java
+++ b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RpcOutcomeListener.java
@@ -27,13 +27,11 @@
    */
   void failed(RpcException ex);
 
-
   void success(V value, ByteBuf buffer);
 
   /**
-   * Called when the sending thread is interrupted. Possible when the fragment is cancelled due to query cancellations or
-   * failures.
+   * Called when the sending thread is interrupted. Possible when the fragment
+   * is cancelled due to query cancellations or failures.
    */
   void interrupted(final InterruptedException e);
-
 }