diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java
index 79b40ee..3e658cb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java
@@ -58,6 +58,8 @@
 import org.apache.drill.exec.vector.ValueVector;
 import org.apache.drill.shaded.guava.com.google.common.annotations.VisibleForTesting;
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import io.netty.buffer.DrillBuf;
 
@@ -65,7 +67,7 @@
  * Record batch used for a particular scan. Operators against one or more
  */
 public class ScanBatch implements CloseableRecordBatch {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ScanBatch.class);
+  private static final Logger logger = LoggerFactory.getLogger(ScanBatch.class);
   private static final ControlsInjector injector = ControlsInjectorFactory.getInjector(ScanBatch.class);
 
   /** Main collection of fields' value vectors. */
@@ -79,7 +81,7 @@
   private BatchSchema schema;
   private final Mutator mutator;
   private boolean done;
-  private Iterator<Map<String, String>> implicitColumns;
+  private final Iterator<Map<String, String>> implicitColumns;
   private Map<String, String> implicitValues;
   private final BufferAllocator allocator;
   private final List<Map<String, String>> implicitColumnList;
@@ -179,18 +181,19 @@
   }
 
   /**
-   * This method is to perform scan specific actions when the scan needs to clean/reset readers and return NONE status
+   * This method is to perform scan specific actions when the scan needs to
+   * clean/reset readers and return NONE status
+   *
    * @return NONE
    */
   private IterOutcome cleanAndReturnNone() {
     if (isRepeatableScan) {
       readers = readerList.iterator();
-      return IterOutcome.NONE;
     } else {
       releaseAssets(); // All data has been read. Release resource.
       done = true;
-      return IterOutcome.NONE;
     }
+    return IterOutcome.NONE;
   }
 
   /**
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashPartition.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashPartition.java
index 0257abc..52fe0a8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashPartition.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashPartition.java
@@ -86,8 +86,8 @@
   private final int maxColumnWidth = VARIABLE_MIN_WIDTH_VALUE_SIZE; // to control memory allocation for varchars
 
   public static final MajorType HVtype = MajorType.newBuilder()
-    .setMinorType(MinorType.INT /* dataType */ )
-    .setMode(DataMode.REQUIRED /* mode */ )
+    .setMinorType(MinorType.INT /* dataType */)
+    .setMode(DataMode.REQUIRED /* mode */)
     .build();
 
   // The vector containers storing all the inner rows
@@ -160,7 +160,7 @@
     }
     this.hjHelper = semiJoin ? null : new HashJoinHelper(context, allocator);
     tmpBatchesList = new ArrayList<>();
-    if ( numPartitions > 1 ) {
+    if (numPartitions > 1) {
       allocateNewCurrentBatchAndHV();
     }
   }
@@ -194,7 +194,7 @@
       while (vci.hasNext()) {
         VectorWrapper<?> vw = vci.next();
         // If processing a spilled container, skip the last column (HV)
-        if ( cycleNum > 0 && ! vci.hasNext() ) { break; }
+        if (cycleNum > 0 && ! vci.hasNext()) { break; }
         ValueVector vv = vw.getValueVector();
         ValueVector newVV = TypeHelper.getNewVector(vv.getField(), allocator);
         newVC.add(newVV); // add first to allow dealloc in case of an OOM
@@ -213,7 +213,7 @@
       newVC.setRecordCount(0);
       success = true;
     } finally {
-      if ( !success ) {
+      if (!success) {
         newVC.clear(); // in case of an OOM
       }
     }
@@ -233,11 +233,12 @@
   /**
    *  Spills if needed
    */
-  public void appendInnerRow(VectorContainer buildContainer, int ind, int hashCode, HashJoinMemoryCalculator.BuildSidePartitioning calc) {
+  public void appendInnerRow(VectorContainer buildContainer, int ind,
+      int hashCode, HashJoinMemoryCalculator.BuildSidePartitioning calc) {
 
-    int pos = currentBatch.appendRow(buildContainer,ind);
+    int pos = currentBatch.appendRow(buildContainer, ind);
     currHVVector.getMutator().set(pos - 1, hashCode);   // store the hash value in the new column
-    if ( pos == recordsPerBatch ) {
+    if (pos == recordsPerBatch) {
       boolean needsSpill = isSpilled || calc.shouldSpill();
       completeAnInnerBatch(true, needsSpill);
     }
@@ -245,12 +246,11 @@
 
   /**
    *  Outer always spills when batch is full
-   *
    */
   public void appendOuterRow(int hashCode, int recordsProcessed) {
     int pos = currentBatch.appendRow(probeBatch.getContainer(),recordsProcessed);
     currHVVector.getMutator().set(pos - 1, hashCode);   // store the hash value in the new column
-    if ( pos == recordsPerBatch ) {
+    if (pos == recordsPerBatch) {
       completeAnOuterBatch(true);
     }
   }
@@ -269,7 +269,7 @@
    * (that is, more rows are coming) - initialize with a new current batch for that partition
    * */
   private void completeABatch(boolean toInitialize, boolean needsSpill) {
-    if ( currentBatch.hasRecordCount() && currentBatch.getRecordCount() > 0) {
+    if (currentBatch.hasRecordCount() && currentBatch.getRecordCount() > 0) {
       currentBatch.add(currHVVector);
       currentBatch.buildSchema(BatchSchema.SelectionVectorMode.NONE);
       tmpBatchesList.add(currentBatch);
@@ -283,10 +283,10 @@
     } else {
       freeCurrentBatchAndHVVector();
     }
-    if ( needsSpill ) { // spill this batch/partition and free its memory
+    if (needsSpill) { // spill this batch/partition and free its memory
       spillThisPartition();
     }
-    if ( toInitialize ) { // allocate a new batch and HV vector
+    if (toInitialize) { // allocate a new batch and HV vector
       allocateNewCurrentBatchAndHV();
     } else {
       currentBatch = null;
@@ -331,11 +331,11 @@
   }
 
   public void spillThisPartition() {
-    if ( tmpBatchesList.size() == 0 ) { return; } // in case empty - nothing to spill
+    if (tmpBatchesList.size() == 0) { return; } // in case empty - nothing to spill
     logger.debug("HashJoin: Spilling partition {}, current cycle {}, part size {} batches", partitionNum, cycleNum, tmpBatchesList.size());
 
     // If this is the first spill for this partition, create an output stream
-    if ( writer == null ) {
+    if (writer == null) {
       final String side = processingOuter ? "outer" : "inner";
       final String suffix = cycleNum > 0 ? side + "_" + Integer.toString(cycleNum) : side;
       spillFile = spillSet.getNextSpillFile(suffix);
@@ -488,10 +488,10 @@
    */
   private void closeWriterInternal(boolean doDeleteFile) {
     try {
-      if ( writer != null ) {
+      if (writer != null) {
         spillSet.close(writer);
       }
-      if ( doDeleteFile && spillFile != null ) {
+      if (doDeleteFile && spillFile != null) {
         spillSet.delete(spillFile);
       }
     } catch (IOException ioe) {
@@ -512,7 +512,7 @@
    * have been consumed.
    */
   public void buildContainersHashTableAndHelper() throws SchemaChangeException {
-    if ( isSpilled ) { return; } // no building for spilled partitions
+    if (isSpilled) { return; } // no building for spilled partitions
     containers = new ArrayList<>();
     hashTable.updateInitialCapacity((int) getNumInMemoryRecords());
     for (int curr = 0; curr < partitionBatchesCount; curr++) {
@@ -520,7 +520,7 @@
       final int currentRecordCount = nextBatch.getRecordCount();
 
       // For every incoming build batch, we create a matching helper batch
-      if ( ! semiJoin ) { hjHelper.addNewBatch(currentRecordCount); }
+      if (! semiJoin) { hjHelper.addNewBatch(currentRecordCount); }
 
       // Holder contains the global index where the key is hashed into using the hash table
       final IndexPointer htIndex = new IndexPointer();
@@ -528,7 +528,7 @@
       assert nextBatch != null;
       assert probeBatch != null;
 
-      hashTable.updateIncoming(nextBatch, probeBatch );
+      hashTable.updateIncoming(nextBatch, probeBatch);
 
       IntVector HV_vector = (IntVector) nextBatch.getLast();
 
@@ -543,7 +543,7 @@
          * the current record index and batch index. This will be used
          * later when we probe and find a match.
          */
-        if ( ! semiJoin ) { hjHelper.setCurrentIndex(htIndex.value, curr /* buildBatchIndex */, recInd); }
+        if (! semiJoin) { hjHelper.setCurrentIndex(htIndex.value, curr /* buildBatchIndex */, recInd); }
       }
 
       containers.add(nextBatch);
@@ -570,11 +570,11 @@
   }
 
   private void freeCurrentBatchAndHVVector() {
-    if ( currentBatch != null ) {
+    if (currentBatch != null) {
       currentBatch.clear();
       currentBatch = null;
     }
-    if ( currHVVector != null ) {
+    if (currHVVector != null) {
       currHVVector.clear();
       currHVVector = null;
     }
@@ -591,13 +591,13 @@
         vc.clear();
       }
     }
-    while ( tmpBatchesList.size() > 0 ) {
+    while (tmpBatchesList.size() > 0) {
       VectorContainer vc = tmpBatchesList.remove(0);
       vc.clear();
     }
     closeWriterInternal(deleteFile);
     clearHashTableAndHelper();
-    if ( containers != null ) {
+    if (containers != null) {
       containers.clear();
     }
   }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
index dc063f9..eab38ec 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
@@ -17,6 +17,10 @@
  */
 package org.apache.drill.exec.physical.impl.join;
 
+import static org.apache.drill.exec.record.RecordBatch.IterOutcome.EMIT;
+import static org.apache.drill.exec.record.RecordBatch.IterOutcome.NONE;
+import static org.apache.drill.exec.record.RecordBatch.IterOutcome.OK_NEW_SCHEMA;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -25,18 +29,14 @@
 import java.util.Map;
 import java.util.Set;
 
-import org.apache.drill.common.expression.ExpressionPosition;
-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.Sets;
-import org.apache.drill.shaded.guava.com.google.common.collect.Iterables;
-
+import org.apache.calcite.rel.core.JoinRelType;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.mutable.MutableBoolean;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.ErrorCollector;
 import org.apache.drill.common.expression.ErrorCollectorImpl;
+import org.apache.drill.common.expression.ExpressionPosition;
 import org.apache.drill.common.expression.FieldReference;
 import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.PathSegment;
@@ -51,7 +51,6 @@
 import org.apache.drill.exec.exception.ClassTransformationException;
 import org.apache.drill.exec.exception.OutOfMemoryException;
 import org.apache.drill.exec.exception.SchemaChangeException;
-import org.apache.drill.exec.expr.CodeGenerator;
 import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
 import org.apache.drill.exec.expr.fn.impl.ValueVectorHashHelper;
 import org.apache.drill.exec.memory.BaseAllocator;
@@ -65,10 +64,10 @@
 import org.apache.drill.exec.physical.impl.common.AbstractSpilledPartitionMetadata;
 import org.apache.drill.exec.physical.impl.common.ChainedHashTable;
 import org.apache.drill.exec.physical.impl.common.Comparator;
+import org.apache.drill.exec.physical.impl.common.HashPartition;
 import org.apache.drill.exec.physical.impl.common.HashTable;
 import org.apache.drill.exec.physical.impl.common.HashTableConfig;
 import org.apache.drill.exec.physical.impl.common.HashTableStats;
-import org.apache.drill.exec.physical.impl.common.HashPartition;
 import org.apache.drill.exec.physical.impl.common.SpilledState;
 import org.apache.drill.exec.physical.impl.spill.SpillSet;
 import org.apache.drill.exec.planner.common.JoinControl;
@@ -85,49 +84,61 @@
 import org.apache.drill.exec.vector.IntVector;
 import org.apache.drill.exec.vector.ValueVector;
 import org.apache.drill.exec.vector.complex.AbstractContainerVector;
-import org.apache.calcite.rel.core.JoinRelType;
 import org.apache.drill.exec.work.filter.BloomFilter;
 import org.apache.drill.exec.work.filter.BloomFilterDef;
 import org.apache.drill.exec.work.filter.RuntimeFilterDef;
 import org.apache.drill.exec.work.filter.RuntimeFilterReporter;
-
-import static org.apache.drill.exec.record.RecordBatch.IterOutcome.EMIT;
-import static org.apache.drill.exec.record.RecordBatch.IterOutcome.NONE;
-import static org.apache.drill.exec.record.RecordBatch.IterOutcome.OK_NEW_SCHEMA;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+import org.apache.drill.shaded.guava.com.google.common.collect.Iterables;
+import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
+import org.apache.drill.shaded.guava.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
- *   This class implements the runtime execution for the Hash-Join operator
- *   supporting INNER, LEFT OUTER, RIGHT OUTER, and FULL OUTER joins
- *
- *   This implementation splits the incoming Build side rows into multiple Partitions, thus allowing spilling of
- *   some of these partitions to disk if memory gets tight. Each partition is implemented as a {@link HashPartition}.
- *   After the build phase is over, in the most general case, some of the partitions were spilled, and the others
- *   are in memory. Each of the partitions in memory would get a {@link HashTable} built.
- *      Next the Probe side is read, and each row is key matched with a Build partition. If that partition is in
- *   memory, then the key is used to probe and perform the join, and the results are added to the outgoing batch.
- *   But if that build side partition was spilled, then the matching Probe size partition is spilled as well.
- *      After all the Probe side was processed, we are left with pairs of spilled partitions. Then each pair is
- *   processed individually (that Build partition should be smaller than the original, hence likely fit whole into
- *   memory to allow probing; if not -- see below).
- *      Processing of each spilled pair is EXACTLY like processing the original Build/Probe incomings. (As a fact,
- *   the {@link #innerNext()} method calls itself recursively !!). Thus the spilled build partition is
- *   read and divided into new partitions, which in turn may spill again (and again...).
- *   The code tracks these spilling "cycles". Normally any such "again" (i.e. cycle of 2 or greater) is a waste,
- *   indicating that the number of partitions chosen was too small.
+ * This class implements the runtime execution for the Hash-Join operator
+ * supporting INNER, LEFT OUTER, RIGHT OUTER, and FULL OUTER joins
+ * <p>
+ * This implementation splits the incoming Build side rows into multiple
+ * Partitions, thus allowing spilling of some of these partitions to disk if
+ * memory gets tight. Each partition is implemented as a {@link HashPartition}.
+ * After the build phase is over, in the most general case, some of the
+ * partitions were spilled, and the others are in memory. Each of the partitions
+ * in memory would get a {@link HashTable} built.
+ * <p>
+ * Next the Probe side is read, and each row is key matched with a Build
+ * partition. If that partition is in memory, then the key is used to probe and
+ * perform the join, and the results are added to the outgoing batch. But if
+ * that build side partition was spilled, then the matching Probe size partition
+ * is spilled as well.
+ * <p>
+ * After all the Probe side was processed, we are left with pairs of spilled
+ * partitions. Then each pair is processed individually (that Build partition
+ * should be smaller than the original, hence likely fit whole into memory to
+ * allow probing; if not -- see below).
+ * <p>
+ * Processing of each spilled pair is EXACTLY like processing the original
+ * Build/Probe incomings. (As a fact, the {@link #innerNext()} method calls
+ * itself recursively !!). Thus the spilled build partition is read and divided
+ * into new partitions, which in turn may spill again (and again...). The code
+ * tracks these spilling "cycles". Normally any such "again" (i.e. cycle of 2 or
+ * greater) is a waste, indicating that the number of partitions chosen was too
+ * small.
  */
+
 public class HashJoinBatch extends AbstractBinaryRecordBatch<HashJoinPOP> implements RowKeyJoin {
-  protected static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(HashJoinBatch.class);
+  private static final Logger logger = LoggerFactory.getLogger(HashJoinBatch.class);
 
   /**
    * The maximum number of records within each internal batch.
    */
-  private int RECORDS_PER_BATCH; // internal batches
+  private final int RECORDS_PER_BATCH; // internal batches
 
   // Join type, INNER, LEFT, RIGHT or OUTER
   private final JoinRelType joinType;
-  private boolean semiJoin;
-  private boolean joinIsLeftOrFull;
-  private boolean joinIsRightOrFull;
+  private final boolean semiJoin;
+  private final boolean joinIsLeftOrFull;
+  private final boolean joinIsRightOrFull;
   private boolean skipHashTableBuild; // when outer side is empty, and the join is inner or left (see DRILL-6755)
 
   // Join conditions
@@ -141,13 +152,15 @@
   private final List<NamedExpression> rightExpr;
 
   /**
-   * Names of the join columns. This names are used in order to help estimate the size of the {@link HashTable}s.
+   * Names of the join columns. This names are used in order to help estimate
+   * the size of the {@link HashTable}s.
    */
   private final Set<String> buildJoinColumns;
 
   // Fields used for partitioning
   /**
-   * The number of {@link HashPartition}s. This is configured via a system option and set in {@link #partitionNumTuning(int, HashJoinMemoryCalculator.BuildSidePartitioning)}.
+   * The number of {@link HashPartition}s. This is configured via a system
+   * option and set in {@link #partitionNumTuning(int, HashJoinMemoryCalculator.BuildSidePartitioning)}.
    */
   private int numPartitions = 1; // must be 2 to the power of bitsInMask
 
@@ -155,8 +168,8 @@
    * The master class used to generate {@link HashTable}s.
    */
   private ChainedHashTable baseHashTable;
-  private MutableBoolean buildSideIsEmpty = new MutableBoolean(false);
-  private MutableBoolean probeSideIsEmpty = new MutableBoolean(false);
+  private final MutableBoolean buildSideIsEmpty = new MutableBoolean(false);
+  private final MutableBoolean probeSideIsEmpty = new MutableBoolean(false);
   private boolean canSpill = true;
   private boolean wasKilled; // a kill was received, may need to clean spilled partns
 
@@ -174,18 +187,18 @@
   private BatchSchema probeSchema;
 
   // Whether this HashJoin is used for a row-key based join
-  private boolean isRowKeyJoin = false;
+  private final boolean isRowKeyJoin;
 
-  private JoinControl joinControl;
+  private final JoinControl joinControl;
 
   // An iterator over the build side hash table (only applicable for row-key joins)
-  private boolean buildComplete = false;
+  private boolean buildComplete;
 
   // indicates if we have previously returned an output batch
   private boolean firstOutputBatch = true;
 
   private int rightHVColPosition;
-  private BufferAllocator allocator;
+  private final BufferAllocator allocator;
   // Local fields for left/right incoming - may be replaced when reading from spilled
   private RecordBatch buildBatch;
   private RecordBatch probeBatch;
@@ -193,27 +206,27 @@
   /**
    * Flag indicating whether or not the first data holding build batch needs to be fetched.
    */
-  private MutableBoolean prefetchedBuild = new MutableBoolean(false);
+  private final MutableBoolean prefetchedBuild = new MutableBoolean(false);
   /**
    * Flag indicating whether or not the first data holding probe batch needs to be fetched.
    */
-  private MutableBoolean prefetchedProbe = new MutableBoolean(false);
+  private final MutableBoolean prefetchedProbe = new MutableBoolean(false);
 
   // For handling spilling
-  private SpillSet spillSet;
+  private final SpillSet spillSet;
   HashJoinPOP popConfig;
 
-  private int originalPartition = -1; // the partition a secondary reads from
+  private final int originalPartition = -1; // the partition a secondary reads from
   IntVector read_right_HV_vector; // HV vector that was read from the spilled batch
-  private int maxBatchesInMemory;
-  private List<String> probeFields = new ArrayList<>(); // keep the same sequence with the bloomFilters
+  private final int maxBatchesInMemory;
+  private final List<String> probeFields = new ArrayList<>(); // keep the same sequence with the bloomFilters
   private boolean enableRuntimeFilter;
   private RuntimeFilterReporter runtimeFilterReporter;
   private ValueVectorHashHelper.Hash64 hash64;
-  private Map<BloomFilter, Integer> bloomFilter2buildId = new HashMap<>();
-  private Map<BloomFilterDef, Integer> bloomFilterDef2buildId = new HashMap<>();
-  private List<BloomFilter> bloomFilters = new ArrayList<>();
-  private boolean bloomFiltersGenerated = false;
+  private final Map<BloomFilter, Integer> bloomFilter2buildId = new HashMap<>();
+  private final Map<BloomFilterDef, Integer> bloomFilterDef2buildId = new HashMap<>();
+  private final List<BloomFilter> bloomFilters = new ArrayList<>();
+  private boolean bloomFiltersGenerated;
 
   /**
    * This holds information about the spilled partitions for the build and probe side.
@@ -223,13 +236,13 @@
     private final String innerSpillFile;
     private int outerSpilledBatches;
     private String outerSpillFile;
-    private boolean updatedOuter = false;
+    private boolean updatedOuter;
 
-    public HashJoinSpilledPartition(final int cycle,
-                                    final int originPartition,
-                                    final int prevOriginPartition,
-                                    final int innerSpilledBatches,
-                                    final String innerSpillFile) {
+    public HashJoinSpilledPartition(int cycle,
+                                    int originPartition,
+                                    int prevOriginPartition,
+                                    int innerSpilledBatches,
+                                    String innerSpillFile) {
       super(cycle, originPartition, prevOriginPartition);
 
       this.innerSpilledBatches = innerSpilledBatches;
@@ -254,7 +267,7 @@
       return outerSpillFile;
     }
 
-    public void updateOuter(final int outerSpilledBatches, final String outerSpillFile) {
+    public void updateOuter(int outerSpilledBatches, String outerSpillFile) {
       Preconditions.checkState(!updatedOuter);
       updatedOuter = true;
 
@@ -294,8 +307,8 @@
   /**
    * Queue of spilled partitions to process.
    */
-  private SpilledState<HashJoinSpilledPartition> spilledState = new SpilledState<>();
-  private HashJoinUpdater spilledStateUpdater = new HashJoinUpdater();
+  private final SpilledState<HashJoinSpilledPartition> spilledState = new SpilledState<>();
+  private final HashJoinUpdater spilledStateUpdater = new HashJoinUpdater();
   private HashJoinSpilledPartition spilledInners[]; // for the outer to find the partition
 
   public enum Metric implements MetricDef {
@@ -349,9 +362,11 @@
 
       if (rightUpstream == OK_NEW_SCHEMA) {
         buildSchema = right.getSchema();
-        // position of the new "column" for keeping the hash values (after the real columns)
+        // position of the new "column" for keeping the hash values
+        // (after the real columns)
         rightHVColPosition = right.getContainer().getNumberOfColumns();
-        // In special cases, when the probe side is empty, and inner/left join - no need for Hash Table
+        // In special cases, when the probe side is empty, and
+        // inner/left join - no need for Hash Table
         skipHashTableBuild = leftUpstream == IterOutcome.NONE && ! joinIsRightOrFull;
         // We only need the hash tables if we have data on the build side.
         setupHashTable();
@@ -364,10 +379,12 @@
       }
     }
 
-    // If we have a valid schema, this will build a valid container. If we were unable to obtain a valid schema,
+    // If we have a valid schema, this will build a valid container.
+    // If we were unable to obtain a valid schema,
     // we still need to build a dummy schema. This code handles both cases for us.
     setupOutputContainerSchema();
     container.buildSchema(BatchSchema.SelectionVectorMode.NONE);
+    container.setEmpty();
   }
 
   /**
@@ -381,7 +398,9 @@
       buildBatch,
       () -> {
         batchMemoryManager.update(RIGHT_INDEX, 0, true);
-        RecordBatchStats.logRecordBatchStats(RecordBatchIOType.INPUT_RIGHT, batchMemoryManager.getRecordBatchSizer(RIGHT_INDEX), getRecordBatchStatsContext());
+        RecordBatchStats.logRecordBatchStats(RecordBatchIOType.INPUT_RIGHT,
+            batchMemoryManager.getRecordBatchSizer(RIGHT_INDEX),
+            getRecordBatchStatsContext());
       });
   }
 
@@ -396,7 +415,9 @@
       probeBatch,
       () -> {
         batchMemoryManager.update(LEFT_INDEX, 0);
-        RecordBatchStats.logRecordBatchStats(RecordBatchIOType.INPUT_LEFT, batchMemoryManager.getRecordBatchSizer(LEFT_INDEX), getRecordBatchStatsContext());
+        RecordBatchStats.logRecordBatchStats(RecordBatchIOType.INPUT_LEFT,
+            batchMemoryManager.getRecordBatchSizer(LEFT_INDEX),
+            getRecordBatchStatsContext());
       });
   }
 
@@ -411,11 +432,11 @@
    * @return The current {@link org.apache.drill.exec.record.RecordBatch.IterOutcome}.
    */
   private IterOutcome prefetchFirstBatch(IterOutcome outcome,
-                                         final MutableBoolean prefetched,
-                                         final MutableBoolean isEmpty,
-                                         final int index,
-                                         final RecordBatch batch,
-                                         final Runnable memoryManagerUpdate) {
+                                         MutableBoolean prefetched,
+                                         MutableBoolean isEmpty,
+                                         int index,
+                                         RecordBatch batch,
+                                         Runnable memoryManagerUpdate) {
     if (prefetched.booleanValue()) {
       // We have already prefetch the first data holding batch
       return outcome;
@@ -429,7 +450,7 @@
       outcome = sniffNonEmptyBatch(outcome, index, batch);
     }
 
-    isEmpty.setValue(outcome == IterOutcome.NONE); // If we recieved NONE there is no data.
+    isEmpty.setValue(outcome == IterOutcome.NONE); // If we received NONE there is no data.
 
     if (outcome == IterOutcome.OUT_OF_MEMORY) {
       // We reached a termination state
@@ -440,7 +461,7 @@
     } else {
       // Got our first batch(es)
       if (spilledState.isFirstCycle()) {
-        // Only collect stats for the first cylce
+        // Only collect stats for the first cycle
         memoryManagerUpdate.run();
       }
       state = BatchState.FIRST;
@@ -488,10 +509,10 @@
    */
   public HashJoinMemoryCalculator getCalculatorImpl() {
     if (maxBatchesInMemory == 0) {
-      final double safetyFactor = context.getOptions().getDouble(ExecConstants.HASHJOIN_SAFETY_FACTOR_KEY);
-      final double fragmentationFactor = context.getOptions().getDouble(ExecConstants.HASHJOIN_FRAGMENTATION_FACTOR_KEY);
-      final double hashTableDoublingFactor = context.getOptions().getDouble(ExecConstants.HASHJOIN_HASH_DOUBLE_FACTOR_KEY);
-      final String hashTableCalculatorType = context.getOptions().getString(ExecConstants.HASHJOIN_HASHTABLE_CALC_TYPE_KEY);
+      double safetyFactor = context.getOptions().getDouble(ExecConstants.HASHJOIN_SAFETY_FACTOR_KEY);
+      double fragmentationFactor = context.getOptions().getDouble(ExecConstants.HASHJOIN_FRAGMENTATION_FACTOR_KEY);
+      double hashTableDoublingFactor = context.getOptions().getDouble(ExecConstants.HASHJOIN_HASH_DOUBLE_FACTOR_KEY);
+      String hashTableCalculatorType = context.getOptions().getString(ExecConstants.HASHJOIN_HASHTABLE_CALC_TYPE_KEY);
 
       return new HashJoinMemoryCalculatorImpl(safetyFactor, fragmentationFactor, hashTableDoublingFactor, hashTableCalculatorType, semiJoin);
     } else {
@@ -502,7 +523,7 @@
   @Override
   public IterOutcome innerNext() {
     if (wasKilled) {
-      // We have recieved a kill signal. We need to stop processing.
+      // We have received a kill signal. We need to stop processing.
       this.cleanup();
       super.close();
       return IterOutcome.NONE;
@@ -522,7 +543,7 @@
        */
       if (state == BatchState.FIRST) {
         // Build the hash table, using the build side record batches.
-        final IterOutcome buildExecuteTermination = executeBuildPhase();
+        IterOutcome buildExecuteTermination = executeBuildPhase();
 
         if (buildExecuteTermination != null) {
           // A termination condition was reached while executing the build phase.
@@ -591,10 +612,7 @@
 
           outputRecords = hashJoinProbe.probeAndProject();
 
-          for (final VectorWrapper<?> v : container) {
-            v.getValueVector().getMutator().setValueCount(outputRecords);
-          }
-          container.setRecordCount(outputRecords);
+          container.setValueCount(outputRecords);
 
           batchMemoryManager.updateOutgoingStats(outputRecords);
           RecordBatchStats.logRecordBatchStats(RecordBatchIOType.OUTPUT, this, getRecordBatchStatsContext());
@@ -685,7 +703,7 @@
   private void killAndDrainUpstream(RecordBatch batch, IterOutcome upstream, boolean isLeft) {
       batch.kill(true);
       while (upstream == IterOutcome.OK_NEW_SCHEMA || upstream == IterOutcome.OK) {
-        for (final VectorWrapper<?> wrapper : batch) {
+        for (VectorWrapper<?> wrapper : batch) {
           wrapper.getValueVector().clear();
         }
         upstream = next( isLeft ? HashJoinHelper.LEFT_INPUT : HashJoinHelper.RIGHT_INPUT, batch);
@@ -695,7 +713,7 @@
   private void killAndDrainRightUpstream() { killAndDrainUpstream(buildBatch, rightUpstream, false); }
 
   private void setupHashTable() throws SchemaChangeException {
-    final List<Comparator> comparators = Lists.newArrayListWithExpectedSize(conditions.size());
+    List<Comparator> comparators = Lists.newArrayListWithExpectedSize(conditions.size());
     conditions.forEach(cond->comparators.add(JoinUtils.checkAndReturnSupportedJoinComparator(cond)));
 
     if ( skipHashTableBuild ) { return; }
@@ -713,13 +731,13 @@
       leftExpr = null;
     } else {
       if (probeBatch.getSchema().getSelectionVectorMode() != BatchSchema.SelectionVectorMode.NONE) {
-        final String errorMsg = new StringBuilder().append("Hash join does not support probe batch with selection vectors. ").append("Probe batch has selection mode = ").append
+        String errorMsg = new StringBuilder().append("Hash join does not support probe batch with selection vectors. ").append("Probe batch has selection mode = ").append
           (probeBatch.getSchema().getSelectionVectorMode()).toString();
         throw new SchemaChangeException(errorMsg);
       }
     }
 
-    final HashTableConfig htConfig = new HashTableConfig((int) context.getOptions().getOption(ExecConstants.MIN_HASH_TABLE_SIZE),
+    HashTableConfig htConfig = new HashTableConfig((int) context.getOptions().getOption(ExecConstants.MIN_HASH_TABLE_SIZE),
       true, HashTable.DEFAULT_LOAD_FACTOR, rightExpr, leftExpr, comparators, joinControl.asInt());
 
     // Create the chained hash table
@@ -735,7 +753,7 @@
     ErrorCollector collector = new ErrorCollectorImpl();
     int i = 0;
     for (NamedExpression ne : htConfig.getKeyExprsBuild()) {
-      final LogicalExpression expr = ExpressionTreeMaterializer.materialize(ne.getExpr(), buildBatch, collector, context.getFunctionRegistry());
+      LogicalExpression expr = ExpressionTreeMaterializer.materialize(ne.getExpr(), buildBatch, collector, context.getFunctionRegistry());
       if (collector.hasErrors()) {
         throw new SchemaChangeException("Failure while materializing expression. " + collector.toErrorString());
       }
@@ -868,7 +886,7 @@
 
       // TODO dirty hack to prevent regressions. Remove this once batch sizing is implemented.
       // We don't have enough memory to do partitioning, we have to do everything in memory
-      final String message = String.format("When using the minimum number of partitions %d we require %s memory but only have %s available. " +
+      String message = String.format("When using the minimum number of partitions %d we require %s memory but only have %s available. " +
           "Forcing legacy behavoir of using unbounded memory in order to prevent regressions.",
         numPartitions,
         FileUtils.byteCountToDisplaySize(buildCalc.getMaxReservedMemory()),
@@ -876,7 +894,7 @@
       logger.warn(message);
 
       // create a Noop memory calculator
-      final HashJoinMemoryCalculator calc = getCalculatorImpl();
+      HashJoinMemoryCalculator calc = getCalculatorImpl();
       calc.initialize(false);
       buildCalc = calc.next();
 
@@ -908,7 +926,7 @@
   private void disableSpilling(String reason) {
     // Fail, or just issue a warning if a reason was given, or a fallback option is enabled
     if ( reason == null ) {
-      final boolean fallbackEnabled = context.getOptions().getOption(ExecConstants.HASHJOIN_FALLBACK_ENABLED_KEY).bool_val;
+      boolean fallbackEnabled = context.getOptions().getOption(ExecConstants.HASHJOIN_FALLBACK_ENABLED_KEY).bool_val;
       if (fallbackEnabled) {
         logger.warn("Spilling is disabled - not enough memory available for internal partitioning. Falling back" +
           " to use unbounded memory");
@@ -986,7 +1004,7 @@
     initializeRuntimeFilter();
 
     // Make the calculator aware of our partitions
-    final HashJoinMemoryCalculator.PartitionStatSet partitionStatSet = new HashJoinMemoryCalculator.PartitionStatSet(partitions);
+    HashJoinMemoryCalculator.PartitionStatSet partitionStatSet = new HashJoinMemoryCalculator.PartitionStatSet(partitions);
     buildCalc.setPartitionStatSet(partitionStatSet);
 
     boolean moreData = true;
@@ -1007,7 +1025,7 @@
         // Fall through
       case OK:
         batchMemoryManager.update(buildBatch, RIGHT_INDEX, 0, true);
-        final int currentRecordCount = buildBatch.getRecordCount();
+        int currentRecordCount = buildBatch.getRecordCount();
         //create runtime filter
         if (spilledState.isFirstCycle() && enableRuntimeFilter) {
           //create runtime filter and send out async
@@ -1040,7 +1058,7 @@
 
           }
           // Append the new inner row to the appropriate partition; spill (that partition) if needed
-          partitions[currPart].appendInnerRow(buildBatch.getContainer(), ind, hashCode, buildCalc); // may spill if needed
+          partitions[currPart].appendInnerRow(buildBatch.getContainer(), ind, hashCode, buildCalc);
         }
 
         if ( read_right_HV_vector != null ) {
@@ -1048,6 +1066,8 @@
           read_right_HV_vector = null;
         }
         break;
+      default:
+        throw new IllegalStateException(rightUpstream.name());
       }
       // Get the next incoming record batch
       rightUpstream = next(HashJoinHelper.RIGHT_INPUT, buildBatch);
@@ -1080,12 +1100,10 @@
     HashJoinMemoryCalculator.PostBuildCalculations postBuildCalc = buildCalc.next();
     postBuildCalc.initialize(probeSideIsEmpty.booleanValue()); // probeEmpty
 
-    //
     //  Traverse all the in-memory partitions' incoming batches, and build their hash tables
-    //
 
     for (int index = 0; index < partitions.length; index++) {
-      final HashPartition partn = partitions[index];
+      HashPartition partn = partitions[index];
 
       if (partn.isSpilled()) {
         // Don't build hash tables for spilled partitions
@@ -1101,7 +1119,7 @@
           partn.buildContainersHashTableAndHelper();
         }
       } catch (OutOfMemoryException e) {
-        final String message = "Failed building hash table on partition " + index + ":\n"
+        String message = "Failed building hash table on partition " + index + ":\n"
           + makeDebugString() + "\n"
           + postBuildCalc.makeDebugString();
         // Include debug info
@@ -1135,9 +1153,9 @@
   private void setupOutputContainerSchema() {
 
     if (buildSchema != null && ! semiJoin ) {
-      for (final MaterializedField field : buildSchema) {
-        final MajorType inputType = field.getType();
-        final MajorType outputType;
+      for (MaterializedField field : buildSchema) {
+        MajorType inputType = field.getType();
+        MajorType outputType;
         // If left or full outer join, then the output type must be nullable. However, map types are
         // not nullable so we must exclude them from the check below (see DRILL-2197).
         if (joinIsLeftOrFull && inputType.getMode() == DataMode.REQUIRED
@@ -1148,16 +1166,16 @@
         }
 
         // make sure to project field with children for children to show up in the schema
-        final MaterializedField projected = field.withType(outputType);
+        MaterializedField projected = field.withType(outputType);
         // Add the vector to our output container
         container.addOrGet(projected);
       }
     }
 
     if (probeSchema != null) { // a probe schema was seen (even though the probe may had no rows)
-      for (final VectorWrapper<?> vv : probeBatch) {
-        final MajorType inputType = vv.getField().getType();
-        final MajorType outputType;
+      for (VectorWrapper<?> vv : probeBatch) {
+        MajorType inputType = vv.getField().getType();
+        MajorType outputType;
 
         // If right or full outer join then the output type should be optional. However, map types are
         // not nullable so we must exclude them from the check below (see DRILL-2771, DRILL-2197).
@@ -1168,7 +1186,7 @@
           outputType = inputType;
         }
 
-        final ValueVector v = container.addOrGet(MaterializedField.create(vv.getField().getName(), outputType));
+        ValueVector v = container.addOrGet(MaterializedField.create(vv.getField().getName(), outputType));
         if (v instanceof AbstractContainerVector) {
           vv.getValueVector().makeTransferPair(v);
           v.clear();
@@ -1213,15 +1231,15 @@
     buildJoinColumns = Sets.newHashSet();
     List<SchemaPath> rightConditionPaths = new ArrayList<>();
     for (int i = 0; i < conditions.size(); i++) {
-      final SchemaPath rightPath = (SchemaPath) conditions.get(i).getRight();
+      SchemaPath rightPath = (SchemaPath) conditions.get(i).getRight();
       rightConditionPaths.add(rightPath);
     }
 
     for (int i = 0; i < conditions.size(); i++) {
-      final SchemaPath rightPath = (SchemaPath) conditions.get(i).getRight();
-      final PathSegment.NameSegment nameSegment = (PathSegment.NameSegment)rightPath.getLastSegment();
+      SchemaPath rightPath = (SchemaPath) conditions.get(i).getRight();
+      PathSegment.NameSegment nameSegment = (PathSegment.NameSegment)rightPath.getLastSegment();
       buildJoinColumns.add(nameSegment.getPath());
-      final String refName = "build_side_" + i;
+      String refName = "build_side_" + i;
       rightExpr.add(new NamedExpression(conditions.get(i).getRight(), new FieldReference(refName)));
     }
 
@@ -1234,7 +1252,7 @@
 
     numPartitions = BaseAllocator.nextPowerOfTwo(numPartitions); // in case not a power of 2
 
-    final long memLimit = context.getOptions().getOption(ExecConstants.HASHJOIN_MAX_MEMORY_VALIDATOR);
+    long memLimit = context.getOptions().getOption(ExecConstants.HASHJOIN_MAX_MEMORY_VALIDATOR);
 
     if (memLimit != 0) {
       allocator.setLimit(memLimit);
@@ -1250,8 +1268,8 @@
     partitions = new HashPartition[0];
 
     // get the output batch size from config.
-    final int configuredBatchSize = (int) context.getOptions().getOption(ExecConstants.OUTPUT_BATCH_SIZE_VALIDATOR);
-    final double avail_mem_factor = (double) context.getOptions().getOption(ExecConstants.OUTPUT_BATCH_SIZE_AVAIL_MEM_FACTOR_VALIDATOR);
+    int configuredBatchSize = (int) context.getOptions().getOption(ExecConstants.OUTPUT_BATCH_SIZE_VALIDATOR);
+    double avail_mem_factor = context.getOptions().getOption(ExecConstants.OUTPUT_BATCH_SIZE_AVAIL_MEM_FACTOR_VALIDATOR);
     int outputBatchSize = Math.min(configuredBatchSize, Integer.highestOneBit((int)(allocator.getLimit() * avail_mem_factor)));
 
     RecordBatchStats.logRecordBatchStats(getRecordBatchStatsContext(),
@@ -1305,11 +1323,11 @@
    * @return A memory dump string.
    */
   public String makeDebugString() {
-    final StringBuilder sb = new StringBuilder();
+    StringBuilder sb = new StringBuilder();
 
     for (int partitionIndex = 0; partitionIndex < partitions.length; partitionIndex++) {
-      final String partitionPrefix = "Partition " + partitionIndex + ": ";
-      final HashPartition hashPartition = partitions[partitionIndex];
+      String partitionPrefix = "Partition " + partitionIndex + ": ";
+      HashPartition hashPartition = partitions[partitionIndex];
       sb.append(partitionPrefix).append(hashPartition.makeDebugString()).append("\n");
     }
 
@@ -1326,7 +1344,7 @@
     if ( buildSideIsEmpty.booleanValue() ) { return; } // no stats when the right side is empty
     if (!spilledState.isFirstCycle()) { return; } // These stats are only for before processing spilled files
 
-    final HashTableStats htStats = new HashTableStats();
+    HashTableStats htStats = new HashTableStats();
     long numSpilled = 0;
     HashTableStats newStats = new HashTableStats();
     // sum the stats from all the partitions
@@ -1446,18 +1464,15 @@
       batchMemoryManager.getNumOutgoingBatches(), batchMemoryManager.getAvgOutputBatchSize(),
       batchMemoryManager.getAvgOutputRowWidth(), batchMemoryManager.getTotalOutputRecords());
 
-    this.cleanup();
+    cleanup();
     super.close();
   }
 
   public HashJoinProbe setupHashJoinProbe() throws ClassTransformationException, IOException {
-    final CodeGenerator<HashJoinProbe> cg = CodeGenerator.get(HashJoinProbe.TEMPLATE_DEFINITION, context.getOptions());
-    cg.plainJavaCapable(true);
-    // cg.saveCodeForDebugging(true);
 
     //  No real code generation !!
 
-    return context.getImplementationClass(cg);
+    return new HashJoinProbeTemplate();
   }
 
   @Override
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbeTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbeTemplate.java
index e862c0e..d40f6a3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbeTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbeTemplate.java
@@ -35,7 +35,7 @@
 
 import static org.apache.drill.exec.record.JoinBatchMemoryManager.LEFT_INDEX;
 
-public abstract class HashJoinProbeTemplate implements HashJoinProbe {
+public class HashJoinProbeTemplate implements HashJoinProbe {
 
   VectorContainer container; // the outgoing container
 
@@ -50,13 +50,13 @@
   // joinControl determines how to handle INTERSECT_DISTINCT vs. INTERSECT_ALL
   private JoinControl joinControl;
 
-  private HashJoinBatch outgoingJoinBatch = null;
+  private HashJoinBatch outgoingJoinBatch;
 
   // Number of records to process on the probe side
-  private int recordsToProcess = 0;
+  private int recordsToProcess;
 
   // Number of records processed on the probe side
-  private int recordsProcessed = 0;
+  private int recordsProcessed;
 
   // Number of records in the output container
   private int outputRecords;
@@ -71,7 +71,7 @@
   private ProbeState probeState = ProbeState.PROBE_PROJECT;
 
   // For outer or right joins, this is a list of unmatched records that needs to be projected
-  private IntArrayList unmatchedBuildIndexes = null;
+  private IntArrayList unmatchedBuildIndexes;
 
   private  HashPartition partitions[];
 
@@ -79,14 +79,14 @@
   // probing later on the same chain of duplicates
   private HashPartition currPartition;
 
-  private int currRightPartition = 0; // for returning RIGHT/FULL
+  private int currRightPartition; // for returning RIGHT/FULL
   IntVector read_left_HV_vector; // HV vector that was read from the spilled batch
-  private int cycleNum = 0; // 1-primary, 2-secondary, 3-tertiary, etc.
+  private int cycleNum; // 1-primary, 2-secondary, 3-tertiary, etc.
   private HashJoinBatch.HashJoinSpilledPartition spilledInners[]; // for the outer to find the partition
   private boolean buildSideIsEmpty = true;
   private int numPartitions = 1; // must be 2 to the power of bitsInMask
-  private int partitionMask = 0; // numPartitions - 1
-  private int bitsInMask = 0; // number of bits in the MASK
+  private int partitionMask; // numPartitions - 1
+  private int bitsInMask; // number of bits in the MASK
   private int numberOfBuildSideColumns;
   private int targetOutputRecords;
   private boolean semiJoin;
@@ -96,6 +96,7 @@
     this.targetOutputRecords = targetOutputRecords;
   }
 
+  @Override
   public int getOutputCount() {
     return outputRecords;
   }
@@ -143,7 +144,7 @@
     this.recordsProcessed = 0;
 
     // A special case - if the left was an empty file
-    if ( leftStartState == IterOutcome.NONE ){
+    if (leftStartState == IterOutcome.NONE){
       changeToFinalProbeState();
     } else {
       this.recordsToProcess = probeBatch.getRecordCount();
@@ -151,16 +152,16 @@
 
     // for those outer partitions that need spilling (cause their matching inners spilled)
     // initialize those partitions' current batches and hash-value vectors
-    for ( HashPartition partn : this.partitions) {
+    for (HashPartition partn : this.partitions) {
       partn.allocateNewCurrentBatchAndHV();
     }
 
     currRightPartition = 0; // In case it's a Right/Full outer join
 
     // Initialize the HV vector for the first (already read) left batch
-    if ( this.cycleNum > 0 ) {
-      if ( read_left_HV_vector != null ) { read_left_HV_vector.clear();}
-      if ( leftStartState != IterOutcome.NONE ) { // Skip when outer spill was empty
+    if (this.cycleNum > 0) {
+      if (read_left_HV_vector != null) { read_left_HV_vector.clear();}
+      if (leftStartState != IterOutcome.NONE) { // Skip when outer spill was empty
         read_left_HV_vector = (IntVector) probeBatch.getContainer().getLast();
       }
     }
@@ -178,6 +179,7 @@
       destVector.copyEntry(container.getRecordCount(), srcVector, buildSrcIndex);
     }
   }
+
   /**
    * Append the given probe side row into the outgoing container, following the build side part
    * @param probeSrcContainer The container for the left/outer side
@@ -190,6 +192,7 @@
       destVector.copyEntry(container.getRecordCount(), srcVector, probeSrcIndex);
     }
   }
+
   /**
    *  A special version of the VectorContainer's appendRow for the HashJoin; (following a probe) it
    *  copies the build and probe sides into the outgoing container. (It uses a composite
@@ -204,12 +207,14 @@
   private int outputRow(ArrayList<VectorContainer> buildSrcContainers, int compositeBuildSrcIndex,
                         VectorContainer probeSrcContainer, int probeSrcIndex) {
 
-    if ( buildSrcContainers != null ) {
+    if (buildSrcContainers != null) {
       int buildBatchIndex = compositeBuildSrcIndex >>> 16;
       int buildOffset = compositeBuildSrcIndex & 65535;
       appendBuild(buildSrcContainers.get(buildBatchIndex), buildOffset);
     }
-    if ( probeSrcContainer != null ) { appendProbe(probeSrcContainer, probeSrcIndex); }
+    if (probeSrcContainer != null) {
+      appendProbe(probeSrcContainer, probeSrcIndex);
+    }
     return container.incRecordCount();
   }
 
@@ -221,7 +226,7 @@
     while (outputRecords < targetOutputRecords && recordsProcessed < recordsToProcess) {
       outputRecords =
         outputRow(partitions[currBuildPart].getContainers(), unmatchedBuildIndexes.get(recordsProcessed),
-          null /* no probeBatch */, 0 /* no probe index */ );
+          null /* no probeBatch */, 0 /* no probe index */);
       recordsProcessed++;
     }
   }
@@ -248,8 +253,8 @@
             recordsToProcess = 0;
             changeToFinalProbeState();
             // in case some outer partitions were spilled, need to spill their last batches
-            for ( HashPartition partn : partitions ) {
-              if ( ! partn.isSpilled() ) { continue; } // skip non-spilled
+            for (HashPartition partn : partitions) {
+              if (! partn.isSpilled()) { continue; } // skip non-spilled
               partn.completeAnOuterBatch(false);
               // update the partition's spill record with the outer side
               HashJoinBatch.HashJoinSpilledPartition sp = spilledInners[partn.getPartitionNum()];
@@ -262,7 +267,7 @@
 
           case OK_NEW_SCHEMA:
             if (probeBatch.getSchema().equals(probeSchema)) {
-              for ( HashPartition partn : partitions ) { partn.updateBatches(); }
+              for (HashPartition partn : partitions) { partn.updateBatches(); }
 
             } else {
               throw SchemaChangeException.schemaChanged("Hash join does not support schema changes in probe side.",
@@ -278,7 +283,7 @@
             if (recordsToProcess == 0) {
               continue;
             }
-            if ( cycleNum > 0 ) {
+            if (cycleNum > 0) {
               read_left_HV_vector = (IntVector) probeBatch.getContainer().getLast(); // Needed ?
             }
         }
@@ -287,8 +292,8 @@
       int probeIndex = -1;
       // Check if we need to drain the next row in the probe side
       if (getNextRecord) {
-        if ( !buildSideIsEmpty ) {
-          int hashCode = ( cycleNum == 0 ) ?
+        if (!buildSideIsEmpty) {
+          int hashCode = (cycleNum == 0) ?
             partitions[0].getProbeHashCode(recordsProcessed)
             : read_left_HV_vector.getAccessor().get(recordsProcessed);
           int currBuildPart = hashCode & partitionMask;
@@ -299,7 +304,7 @@
           currPartition = partitions[currBuildPart]; // inner if not spilled, else outer
 
           // If the matching inner partition was spilled
-          if ( outgoingJoinBatch.isSpilledInner(currBuildPart) ) {
+          if (outgoingJoinBatch.isSpilledInner(currBuildPart)) {
             // add this row to its outer partition (may cause a spill, when the batch is full)
 
             currPartition.appendOuterRow(hashCode, recordsProcessed);
@@ -312,8 +317,8 @@
 
         }
 
-        if ( semiJoin ) {
-          if ( probeIndex != -1 ) {
+        if (semiJoin) {
+          if (probeIndex != -1) {
             // output the probe side only
             outputRecords =
               outputRow(null, 0, probeBatch.getContainer(), recordsProcessed);
@@ -392,7 +397,6 @@
         }
       }
     }
-
   }
 
   /**
@@ -408,7 +412,7 @@
     outputRecords = 0;
 
     // When handling spilled partitions, the state becomes DONE at the end of each partition
-    if ( probeState == ProbeState.DONE ) {
+    if (probeState == ProbeState.DONE) {
       return outputRecords; // that is zero
     }
 
@@ -422,7 +426,7 @@
       do {
 
         if (unmatchedBuildIndexes == null) { // first time for this partition ?
-          if ( buildSideIsEmpty ) { return outputRecords; } // in case of an empty right
+          if (buildSideIsEmpty) { return outputRecords; } // in case of an empty right
           // Get this partition's list of build indexes that didn't match any record on the probe side
           unmatchedBuildIndexes = partitions[currRightPartition].getNextUnmatchedIndex();
           recordsProcessed = 0;
@@ -432,14 +436,14 @@
         // Project the list of unmatched records on the build side
         executeProjectRightPhase(currRightPartition);
 
-        if ( recordsProcessed < recordsToProcess ) { // more records in this partition?
+        if (recordsProcessed < recordsToProcess) { // more records in this partition?
           return outputRecords;  // outgoing is full; report and come back later
         } else {
           currRightPartition++; // on to the next right partition
           unmatchedBuildIndexes = null;
         }
 
-      }   while ( currRightPartition < numPartitions );
+      }   while (currRightPartition < numPartitions);
 
       probeState = ProbeState.DONE; // last right partition was handled; we are done now
     }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/PartitionLimitRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/PartitionLimitRecordBatch.java
index ae04579..3f90a3b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/PartitionLimitRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/PartitionLimitRecordBatch.java
@@ -147,10 +147,7 @@
       incoming.getContainer().zeroVectors();
       outgoingSv.setRecordCount(0);
       outgoingSv.setBatchActualRecordCount(0);
-      // Must allocate vectors to allow for offset vectors which
-      // require a zero in the 0th position.
-      container.allocateNew();
-      container.setRecordCount(0);
+      container.setEmpty();
       // Release buffer for sv2 (if any)
       if (incomingSv != null) {
         incomingSv.clear();
@@ -162,16 +159,9 @@
       tp.transfer();
     }
 
-    // Must report the actual value count as the record
-    // count. This is NOT the input record count when the input
-    // is an SV2.
-    int inputValueCount = incoming.getContainer().getRecordCount();
-    container.setRecordCount(inputValueCount);
-
     // Allocate SV2 vectors for the record count size since we transfer all the vectors buffer from input record
     // batch to output record batch and later an SV2Remover copies the needed records.
     outgoingSv.allocateNew(inputRecordCount);
-    outgoingSv.setBatchActualRecordCount(inputValueCount);
     limit(inputRecordCount);
 
     // Release memory for incoming sv (if any)
@@ -224,7 +214,7 @@
       }
     }
 
-    outgoingSv.setRecordCount(svIndex);
+    setOutgoingRecordCount(inputRecordCount, svIndex);
   }
 
   private void updateOutputSV2(int svIndex, int incomingIndex) {
@@ -243,6 +233,16 @@
     }
   }
 
+  private void setOutgoingRecordCount(int inputRecordCount, int outputCount) {
+    outgoingSv.setRecordCount(outputCount);
+    // Must report the actual value count as the record
+    // count. This is NOT the input record count when the input
+    // is an SV2.
+    int inputValueCount = incoming.getContainer().getRecordCount();
+    container.setRecordCount(inputValueCount);
+    outgoingSv.setBatchActualRecordCount(inputValueCount);
+  }
+
   /**
    * Reset the states for recordStartOffset, numberOfRecords and based on the
    * {@link PartitionLimit} passed to the operator. It also resets the
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
index 8f16b51..420b61a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
@@ -18,19 +18,17 @@
 package org.apache.drill.exec.physical.impl.mergereceiver;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
-import java.util.Iterator;
 import java.util.List;
 import java.util.PriorityQueue;
-import java.util.ArrayList;
 
 import org.apache.calcite.rel.RelFieldCollation.Direction;
 import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.common.expression.ErrorCollector;
 import org.apache.drill.common.expression.ErrorCollectorImpl;
 import org.apache.drill.common.expression.LogicalExpression;
-import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.logical.data.Order.Ordering;
 import org.apache.drill.exec.compile.sig.GeneratorMapping;
 import org.apache.drill.exec.compile.sig.MappingSet;
@@ -63,13 +61,8 @@
 import org.apache.drill.exec.record.RecordBatch;
 import org.apache.drill.exec.record.RecordBatchLoader;
 import org.apache.drill.exec.record.SchemaBuilder;
-import org.apache.drill.exec.record.TypedFieldId;
 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.record.WritableBatch;
-import org.apache.drill.exec.record.selection.SelectionVector2;
-import org.apache.drill.exec.record.selection.SelectionVector4;
 import org.apache.drill.exec.rpc.RpcException;
 import org.apache.drill.exec.rpc.RpcOutcomeListener;
 import org.apache.drill.exec.testing.ControlsInjector;
@@ -78,20 +71,21 @@
 import org.apache.drill.exec.vector.CopyUtil;
 import org.apache.drill.exec.vector.FixedWidthVector;
 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;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.sun.codemodel.JConditional;
 import com.sun.codemodel.JExpr;
 
-
 import io.netty.buffer.ByteBuf;
 
 /**
  * The MergingRecordBatch merges pre-sorted record batches from remote senders.
  */
 public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP> implements RecordBatch {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MergingRecordBatch.class);
+  private static final Logger logger = LoggerFactory.getLogger(MergingRecordBatch.class);
   private static final ControlsInjector injector = ControlsInjectorFactory.getInjector(MergingRecordBatch.class);
 
   private static final int OUTGOING_BATCH_SIZE = 32 * 1024;
@@ -99,21 +93,20 @@
   private RecordBatchLoader[] batchLoaders;
   private final RawFragmentBatchProvider[] fragProviders;
   private final ExchangeFragmentContext context;
-  private VectorContainer outgoingContainer;
   private MergingReceiverGeneratorBase merger;
   private final MergingReceiverPOP config;
-  private boolean hasRun = false;
+  private boolean hasRun;
   private boolean outgoingBatchHasSpace = true;
   private boolean hasMoreIncoming = true;
 
-  private int outgoingPosition = 0;
-  private int senderCount = 0;
+  private int outgoingPosition;
+  private int senderCount;
   private RawFragmentBatch[] incomingBatches;
   private int[] batchOffsets;
   private PriorityQueue <Node> pqueue;
   private RawFragmentBatch[] tempBatchHolder;
-  private long[] inputCounts;
-  private long[] outputCounts;
+  private final long[] inputCounts;
+  private final long[] outputCounts;
 
   public enum Metric implements MetricDef {
     BYTES_RECEIVED,
@@ -132,7 +125,6 @@
     super(config, context, true, context.newOperatorContext(config));
     this.fragProviders = fragProviders;
     this.context = context;
-    this.outgoingContainer = new VectorContainer(oContext);
     this.stats.setLongStat(Metric.NUM_SENDERS, config.getNumSenders());
     this.config = config;
     this.inputCounts = new long[config.getNumSenders()];
@@ -343,11 +335,11 @@
         bldr.addField(v.getField());
 
         // allocate a new value vector
-        outgoingContainer.addOrGet(v.getField());
+        container.addOrGet(v.getField());
       }
       allocateOutgoing();
 
-      outgoingContainer.buildSchema(BatchSchema.SelectionVectorMode.NONE);
+      container.buildSchema(BatchSchema.SelectionVectorMode.NONE);
 
       // generate code for merge operations (copy and compare)
       try {
@@ -479,9 +471,7 @@
     }
 
     // set the value counts in the outgoing vectors
-    for (final VectorWrapper<?> vw : outgoingContainer) {
-      vw.getValueVector().getMutator().setValueCount(outgoingPosition);
-    }
+    container.setValueCount(outgoingPosition);
 
     if (pqueue.isEmpty()) {
       state = BatchState.DONE;
@@ -535,8 +525,8 @@
 
   @Override
   public BatchSchema getSchema() {
-    if (outgoingContainer.hasSchema()) {
-      return outgoingContainer.getSchema();
+    if (container.hasSchema()) {
+      return container.getSchema();
     }
     return null;
   }
@@ -567,7 +557,7 @@
         }
         tempBatchHolder[i] = batch;
         for (final SerializedField field : batch.getHeader().getDef().getFieldList()) {
-          final ValueVector v = outgoingContainer.addOrGet(MaterializedField.create(field));
+          final ValueVector v = container.addOrGet(MaterializedField.create(field));
           v.allocateNew();
         }
         break;
@@ -575,7 +565,8 @@
     } catch (final IOException e) {
       throw new DrillRuntimeException(e);
     }
-    outgoingContainer.buildSchema(SelectionVectorMode.NONE);
+    container.buildSchema(SelectionVectorMode.NONE);
+    container.setEmpty();
   }
 
   @Override
@@ -644,36 +635,6 @@
     //No op
   }
 
-  @Override
-  public Iterator<VectorWrapper<?>> iterator() {
-    return outgoingContainer.iterator();
-  }
-
-  @Override
-  public SelectionVector2 getSelectionVector2() {
-    throw new UnsupportedOperationException();
-  }
-
-  @Override
-  public SelectionVector4 getSelectionVector4() {
-    throw new UnsupportedOperationException();
-  }
-
-  @Override
-  public TypedFieldId getValueVectorId(final SchemaPath path) {
-    return outgoingContainer.getValueVectorId(path);
-  }
-
-  @Override
-  public VectorWrapper<?> getValueAccessorById(final Class<?> clazz, final int... ids) {
-    return outgoingContainer.getValueAccessorById(clazz, ids);
-  }
-
-  @Override
-  public WritableBatch getWritableBatch() {
-    return WritableBatch.get(this);
-  }
-
   private boolean isSameSchemaAmongBatches(final RecordBatchLoader[] batchLoaders) {
     Preconditions.checkArgument(batchLoaders.length > 0, "0 batch is not allowed!");
 
@@ -689,7 +650,7 @@
   }
 
   private void allocateOutgoing() {
-    for (final VectorWrapper<?> w : outgoingContainer) {
+    for (final VectorWrapper<?> w : container) {
       final ValueVector v = w.getValueVector();
       if (v instanceof FixedWidthVector) {
         AllocationHelper.allocate(v, OUTGOING_BATCH_SIZE, 1);
@@ -732,7 +693,7 @@
       g.setMappingSet(MAIN_MAPPING);
       final MergingReceiverGeneratorBase merger = context.getImplementationClass(cg);
 
-      merger.doSetup(context, batch, outgoingContainer);
+      merger.doSetup(context, batch, container);
       return merger;
     } catch (ClassTransformationException | IOException e) {
       throw new SchemaChangeException(e);
@@ -742,7 +703,7 @@
   private final MappingSet MAIN_MAPPING = new MappingSet( (String) null, null, ClassGenerator.DEFAULT_SCALAR_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
   private final MappingSet LEFT_MAPPING = new MappingSet("leftIndex", null, ClassGenerator.DEFAULT_SCALAR_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
   private final MappingSet RIGHT_MAPPING = new MappingSet("rightIndex", null, ClassGenerator.DEFAULT_SCALAR_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
-  private GeneratorMapping COPIER_MAPPING = new GeneratorMapping("doSetup", "doCopy", null, null);
+  private final GeneratorMapping COPIER_MAPPING = new GeneratorMapping("doSetup", "doCopy", null, null);
   private final MappingSet COPIER_MAPPING_SET = new MappingSet(COPIER_MAPPING, COPIER_MAPPING);
 
   private void generateComparisons(final ClassGenerator<?> g, final VectorAccessible batch) throws SchemaChangeException {
@@ -817,7 +778,7 @@
 
   @Override
   public void close() {
-    outgoingContainer.clear();
+    container.clear();
     if (batchLoaders != null) {
       for (final RecordBatchLoader rbl : batchLoaders) {
         if (rbl != null) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
index e8b522c..c309e8c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
@@ -23,6 +23,7 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.calcite.rel.RelFieldCollation.Direction;
 import org.apache.drill.common.expression.ErrorCollector;
 import org.apache.drill.common.expression.ErrorCollectorImpl;
 import org.apache.drill.common.expression.FieldReference;
@@ -69,43 +70,45 @@
 import org.apache.drill.exec.vector.AllocationHelper;
 import org.apache.drill.exec.vector.IntVector;
 import org.apache.drill.exec.vector.ValueVector;
-import org.apache.calcite.rel.RelFieldCollation.Direction;
-
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.sun.codemodel.JConditional;
 import com.sun.codemodel.JExpr;
 
 /**
- * The purpose of this operator is to generate an ordered partition, rather than a random hash partition. This could be
- * used to do a total order sort, for example. This operator reads in a few incoming record batches, samples these
- * batches, and stores them in the distributed cache. The samples from all the parallel-running fragments are merged,
- * and a partition-table is built and stored in the distributed cache for use by all fragments. A new column is added to
- * the outgoing batch, whose value is determined by where each record falls in the partition table. This column is used
- * by PartitionSenderRootExec to determine which bucket to assign each record to.
+ * Generates an ordered partition, rather than a random hash partition. This
+ * could be used to do a total order sort, for example. This operator reads in a
+ * few incoming record batches, samples these batches, and stores them in the
+ * distributed cache. The samples from all the parallel-running fragments are
+ * merged, and a partition-table is built and stored in the distributed cache
+ * for use by all fragments. A new column is added to the outgoing batch, whose
+ * value is determined by where each record falls in the partition table. This
+ * column is used by PartitionSenderRootExec to determine which bucket to assign
+ * each record to.
  */
 public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPartitionSender> {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(OrderedPartitionRecordBatch.class);
+  static final Logger logger = LoggerFactory.getLogger(OrderedPartitionRecordBatch.class);
 
-//  private static final long ALLOCATOR_INITIAL_RESERVATION = 1*1024*1024;
-//  private static final long ALLOCATOR_MAX_RESERVATION = 20L*1000*1000*1000;
-
-  public static final CacheConfig<String, CachedVectorContainer> SINGLE_CACHE_CONFIG = CacheConfig //
-      .newBuilder(CachedVectorContainer.class) //
-      .name("SINGLE-" + CachedVectorContainer.class.getSimpleName()) //
-      .mode(SerializationMode.DRILL_SERIALIZIABLE) //
+  public static final CacheConfig<String, CachedVectorContainer> SINGLE_CACHE_CONFIG = CacheConfig
+      .newBuilder(CachedVectorContainer.class)
+      .name("SINGLE-" + CachedVectorContainer.class.getSimpleName())
+      .mode(SerializationMode.DRILL_SERIALIZIABLE)
       .build();
-  public static final CacheConfig<String, CachedVectorContainer> MULTI_CACHE_CONFIG = CacheConfig //
-      .newBuilder(CachedVectorContainer.class) //
-      .name("MULTI-" + CachedVectorContainer.class.getSimpleName()) //
-      .mode(SerializationMode.DRILL_SERIALIZIABLE) //
+  public static final CacheConfig<String, CachedVectorContainer> MULTI_CACHE_CONFIG = CacheConfig
+      .newBuilder(CachedVectorContainer.class)
+      .name("MULTI-" + CachedVectorContainer.class.getSimpleName())
+      .mode(SerializationMode.DRILL_SERIALIZIABLE)
       .build();
 
-  private final MappingSet mainMapping = new MappingSet( (String) null, null, ClassGenerator.DEFAULT_CONSTANT_MAP,
-      ClassGenerator.DEFAULT_SCALAR_MAP);
+  private final MappingSet mainMapping = new MappingSet( (String) null, null,
+      ClassGenerator.DEFAULT_CONSTANT_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
   private final MappingSet incomingMapping = new MappingSet("inIndex", null, "incoming", null,
       ClassGenerator.DEFAULT_CONSTANT_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
-  private final MappingSet partitionMapping = new MappingSet("partitionIndex", null, "partitionVectors", null,
+  private final MappingSet partitionMapping = new MappingSet("partitionIndex",
+      null, "partitionVectors", null,
       ClassGenerator.DEFAULT_CONSTANT_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
 
   private final int recordsToSample; // How many records must be received before analyzing
@@ -115,13 +118,13 @@
   protected final RecordBatch incoming;
   private boolean first = true;
   private OrderedPartitionProjector projector;
-  private VectorContainer partitionVectors = new VectorContainer();
-  private int partitions;
+  private final VectorContainer partitionVectors = new VectorContainer();
+  private final int partitions;
   private Queue<VectorContainer> batchQueue;
   private int recordsSampled;
-  private int sendingMajorFragmentWidth;
-  private boolean startedUnsampledBatches = false;
-  private boolean upstreamNone = false;
+  private final int sendingMajorFragmentWidth;
+  private boolean startedUnsampledBatches;
+  private boolean upstreamNone;
   private int recordCount;
 
   private final IntVector partitionKeyVector;
@@ -142,6 +145,7 @@
 
     DistributedCache cache = null;
     // Clearly, this code is not used!
+    // cache can only be null here!
     this.mmap = cache.getMultiMap(MULTI_CACHE_CONFIG);
     this.tableMap = cache.getMap(SINGLE_CACHE_CONFIG);
     Preconditions.checkNotNull(tableMap);
@@ -161,7 +165,6 @@
     partitionKeyVector.clear();
   }
 
-
   private boolean saveSamples() throws SchemaChangeException, ClassTransformationException, IOException {
     recordsSampled = 0;
     IterOutcome upstream;
@@ -223,10 +226,7 @@
           allocationSize *= 2;
         }
       }
-      for (VectorWrapper<?> vw : containerToCache) {
-        vw.getValueVector().getMutator().setValueCount(copier.getOutputRecords());
-      }
-      containerToCache.setRecordCount(copier.getOutputRecords());
+      containerToCache.setValueCount(copier.getOutputRecords());
 
       // Get a distributed multimap handle from the distributed cache, and put the vectors from the new vector container
       // into a serializable wrapper object, and then add to distributed map
@@ -251,9 +251,13 @@
   }
 
   /**
-   * Wait until the at least the given timeout is expired or interrupted and the fragment status is not runnable.
-   * @param timeout Timeout in milliseconds.
-   * @return True if the given timeout is expired. False when interrupted and the fragment status is not runnable.
+   * Wait until the at least the given timeout is expired or interrupted and the
+   * fragment status is not runnable.
+   *
+   * @param timeout
+   *          Timeout in milliseconds.
+   * @return True if the given timeout is expired. False when interrupted and
+   *         the fragment status is not runnable.
    */
   private boolean waitUntilTimeOut(final long timeout) {
     while(true) {
@@ -269,11 +273,14 @@
   }
 
   /**
-   * This method is called when the first batch comes in. Incoming batches are collected until a threshold is met. At
-   * that point, the records in the batches are sorted and sampled, and the sampled records are stored in the
-   * distributed cache. Once a sufficient fraction of the fragments have shared their samples, each fragment grabs all
-   * the samples, sorts all the records, builds a partition table, and attempts to push the partition table to the
-   * distributed cache. Whichever table gets pushed first becomes the table used by all fragments for partitioning.
+   * Called when the first batch comes in. Incoming batches are collected until
+   * a threshold is met. At that point, the records in the batches are sorted
+   * and sampled, and the sampled records are stored in the distributed cache.
+   * Once a sufficient fraction of the fragments have shared their samples, each
+   * fragment grabs all the samples, sorts all the records, builds a partition
+   * table, and attempts to push the partition table to the distributed cache.
+   * Whichever table gets pushed first becomes the table used by all fragments
+   * for partitioning.
    *
    * @return True is successful. False if failed.
    */
@@ -288,8 +295,9 @@
       long val = minorFragmentSampleCount.incrementAndGet();
       logger.debug("Incremented mfsc, got {}", val);
 
-      final long fragmentsBeforeProceed = (long) Math.ceil(sendingMajorFragmentWidth * completionFactor);
-      final String finalTableKey = mapKey + "final";
+      long fragmentsBeforeProceed =
+          (long) Math.ceil(sendingMajorFragmentWidth * completionFactor);
+      String finalTableKey = mapKey + "final";
 
       if (val == fragmentsBeforeProceed) { // we crossed the barrier, build table and get data.
         buildTable();
@@ -326,7 +334,7 @@
         partitionVectors.add(w.getValueVector());
       }
 
-    } catch (final ClassTransformationException | IOException | SchemaChangeException ex) {
+    } catch (ClassTransformationException | IOException | SchemaChangeException ex) {
       kill(false);
       context.getExecutorState().fail(ex);
       return false;
@@ -340,8 +348,8 @@
     // Get all samples from distributed map
 
     SortRecordBatchBuilder containerBuilder = new SortRecordBatchBuilder(context.getAllocator());
-    final VectorContainer allSamplesContainer = new VectorContainer();
-    final VectorContainer candidatePartitionTable = new VectorContainer();
+    VectorContainer allSamplesContainer = new VectorContainer();
+    VectorContainer candidatePartitionTable = new VectorContainer();
     CachedVectorContainer wrap = null;
     try {
       for (CachedVectorContainer w : mmap.get(mapKey)) {
@@ -375,9 +383,7 @@
         int skipRecords = containerBuilder.getSv4().getTotalCount() / partitions;
         if (copier.copyRecords(skipRecords, skipRecords, partitions - 1)) {
           assert copier.getOutputRecords() == partitions - 1 : String.format("output records: %d partitions: %d", copier.getOutputRecords(), partitions);
-          for (VectorWrapper<?> vw : candidatePartitionTable) {
-            vw.getValueVector().getMutator().setValueCount(copier.getOutputRecords());
-          }
+          candidatePartitionTable.setValueCount(copier.getOutputRecords());
           break;
         } else {
           candidatePartitionTable.zeroVectors();
@@ -397,7 +403,6 @@
         wrap.clear();
       }
     }
-
   }
 
   /**
@@ -415,8 +420,8 @@
    */
   private SampleCopier getCopier(SelectionVector4 sv4, VectorContainer incoming, VectorContainer outgoing,
       List<Ordering> orderings, List<ValueVector> localAllocationVectors) throws SchemaChangeException {
-    final ErrorCollector collector = new ErrorCollectorImpl();
-    final ClassGenerator<SampleCopier> cg = CodeGenerator.getRoot(SampleCopier.TEMPLATE_DEFINITION, context.getOptions());
+    ErrorCollector collector = new ErrorCollectorImpl();
+    ClassGenerator<SampleCopier> cg = CodeGenerator.getRoot(SampleCopier.TEMPLATE_DEFINITION, context.getOptions());
     // Note: disabled for now. This may require some debugging:
     // no tests are available for this operator.
     // cg.getCodeGenerator().plainOldJavaCapable(true);
@@ -425,7 +430,7 @@
 
     int i = 0;
     for (Ordering od : orderings) {
-      final LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), incoming, collector, context.getFunctionRegistry());
+      LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), incoming, collector, context.getFunctionRegistry());
       TypeProtos.MajorType.Builder builder = TypeProtos.MajorType.newBuilder().mergeFrom(expr.getMajorType())
           .clearMode().setMode(TypeProtos.DataMode.REQUIRED);
       TypeProtos.MajorType newType = builder.build();
@@ -567,31 +572,29 @@
     int recordCount = batch.getRecordCount();
     AllocationHelper.allocate(partitionKeyVector, recordCount, 50);
     projector.projectRecords(recordCount, 0);
-    for (VectorWrapper<?> v : container) {
-      ValueVector.Mutator m = v.getValueVector().getMutator();
-      m.setValueCount(recordCount);
-    }
+    container.setValueCount(recordCount);
   }
 
   /**
-   * Sets up projection that will transfer all of the columns in batch, and also populate the partition column based on
-   * which partition a record falls into in the partition table
+   * Sets up projection that will transfer all of the columns in batch, and also
+   * populate the partition column based on which partition a record falls into
+   * in the partition table
    *
    * @param batch
    * @throws SchemaChangeException
    */
   protected void setupNewSchema(VectorAccessible batch) throws SchemaChangeException {
     container.clear();
-    final ErrorCollector collector = new ErrorCollectorImpl();
-    final List<TransferPair> transfers = Lists.newArrayList();
+    ErrorCollector collector = new ErrorCollectorImpl();
+    List<TransferPair> transfers = Lists.newArrayList();
 
-    final ClassGenerator<OrderedPartitionProjector> cg = CodeGenerator.getRoot(
+    ClassGenerator<OrderedPartitionProjector> cg = CodeGenerator.getRoot(
         OrderedPartitionProjector.TEMPLATE_DEFINITION, context.getOptions());
     // Note: disabled for now. This may require some debugging:
     // no tests are available for this operator.
-//    cg.getCodeGenerator().plainOldJavaCapable(true);
+    // cg.getCodeGenerator().plainOldJavaCapable(true);
     // Uncomment out this line to debug the generated code.
-//    cg.getCodeGenerator().saveCodeForDebugging(true);
+    // cg.getCodeGenerator().saveCodeForDebugging(true);
 
     for (VectorWrapper<?> vw : batch) {
       TransferPair tp = vw.getValueVector().getTransferPair(oContext.getAllocator());
@@ -603,7 +606,7 @@
 
     int count = 0;
     for (Ordering od : popConfig.getOrderings()) {
-      final LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), batch, collector, context.getFunctionRegistry());
+      LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), batch, collector, context.getFunctionRegistry());
       if (collector.hasErrors()) {
         throw new SchemaChangeException("Failure while materializing expression. " + collector.toErrorString());
       }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java
index edbbfe2..a3149b8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java
@@ -51,13 +51,15 @@
 import org.apache.drill.exec.record.selection.SelectionVector2;
 import org.apache.drill.exec.record.selection.SelectionVector4;
 import org.apache.drill.exec.vector.ValueVector;
-
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public abstract class PartitionerTemplate implements Partitioner {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PartitionerTemplate.class);
+  static final Logger logger = LoggerFactory.getLogger(PartitionerTemplate.class);
 
-  // Always keep the recordCount as (2^x) - 1 to better utilize the memory allocation in ValueVectors
+  // Always keep the recordCount as (2^x) - 1 to better utilize the memory
+  // allocation in ValueVectors
   private static final int DEFAULT_RECORD_BATCH_SIZE = (1 << 10) - 1;
 
   private SelectionVector2 sv2;
@@ -68,7 +70,7 @@
   protected FragmentContext context;
   private int start;
   private int end;
-  private List<OutgoingRecordBatch> outgoingBatches = Lists.newArrayList();
+  private final List<OutgoingRecordBatch> outgoingBatches = Lists.newArrayList();
 
   private int outgoingRecordBatchSize = DEFAULT_RECORD_BATCH_SIZE;
 
@@ -306,10 +308,14 @@
 
     public void flush(boolean schemaChanged) throws IOException {
       if (dropAll) {
-        // If we are in dropAll mode, we still want to copy the data, because we can't stop copying a single outgoing
-        // batch with out stopping all outgoing batches. Other option is check for status of dropAll before copying
-        // every single record in copy method which has the overhead for every record all the time. Resetting the output
-        // count, reusing the same buffers and copying has overhead only for outgoing batches whose receiver has
+        // If we are in dropAll mode, we still want to copy the data, because we
+        // can't stop copying a single outgoing
+        // batch with out stopping all outgoing batches. Other option is check
+        // for status of dropAll before copying
+        // every single record in copy method which has the overhead for every
+        // record all the time. Resetting the output
+        // count, reusing the same buffers and copying has overhead only for
+        // outgoing batches whose receiver has
         // terminated.
 
         // Reset the count to 0 and use existing buffers for exhausting input where receiver of this batch is terminated
@@ -332,11 +338,7 @@
         return;
       }
 
-      if (recordCount != 0) {
-        for (VectorWrapper<?> w : vectorContainer) {
-          w.getValueVector().getMutator().setValueCount(recordCount);
-        }
-      }
+      vectorContainer.setValueCount(recordCount);
 
       FragmentWritableBatch writableBatch = new FragmentWritableBatch(isLastBatch,
           handle.getQueryId(),
@@ -354,8 +356,10 @@
         stats.stopWait();
       }
 
-      // If the current batch is the last batch, then set a flag to ignore any requests to flush the data
-      // This is possible when the receiver is terminated, but we still get data from input operator
+      // If the current batch is the last batch, then set a flag to ignore any
+      // requests to flush the data
+      // This is possible when the receiver is terminated, but we still get data
+      // from input operator
       if (isLastBatch) {
         dropAll = true;
       }
@@ -421,7 +425,6 @@
       return recordCount;
     }
 
-
     @Override
     public long getTotalRecords() {
       return totalRecords;
@@ -459,6 +462,5 @@
     public void clear(){
       vectorContainer.clear();
     }
-
   }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
index 8215fde..674b3ab 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
@@ -886,7 +886,7 @@
 
     doAlloc(0);
     container.buildSchema(SelectionVectorMode.NONE);
-    container.setValueCount(0);
+    container.setEmpty();
     wasNone = true;
     return IterOutcome.OK_NEW_SCHEMA;
   }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java
index 0c11893..dd30c44 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java
@@ -34,6 +34,7 @@
 
   private ImmutableList<TransferPair> transfers;
   private SelectionVector2 vector2;
+  @SuppressWarnings("unused")
   private SelectionVector4 vector4;
   private SelectionVectorMode svMode;
 
@@ -97,8 +98,10 @@
     case TWO_BYTE:
       vector2 = incoming.getSelectionVector2();
       break;
-    default:
+    case NONE:
       break;
+    default:
+      throw new UnsupportedOperationException();
     }
     this.transfers = ImmutableList.copyOf(transfers);
     doSetup(context, incoming, outgoing);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/rangepartitioner/RangePartitionRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/rangepartitioner/RangePartitionRecordBatch.java
index 6a54828..11d307b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/rangepartitioner/RangePartitionRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/rangepartitioner/RangePartitionRecordBatch.java
@@ -35,34 +35,33 @@
 import org.apache.drill.exec.record.VectorAccessible;
 import org.apache.drill.exec.record.VectorWrapper;
 import org.apache.drill.exec.vector.IntVector;
-import org.apache.drill.exec.vector.ValueVector;
-
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
- * A RangePartitionRecordBatch is a run-time operator that provides the ability to divide up the input rows
- * into a fixed number of separate ranges or 'buckets' based on the values of a set of columns (the range
- * partitioning columns).
+ * Provides the ability to divide up the input rows into a fixed number of
+ * separate ranges or 'buckets' based on the values of a set of columns (the
+ * range partitioning columns).
  */
 public class RangePartitionRecordBatch extends AbstractSingleRecordBatch<RangePartitionSender> {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RangePartitionRecordBatch.class);
+  static final Logger logger = LoggerFactory.getLogger(RangePartitionRecordBatch.class);
 
-  private int numPartitions;
+  private final int numPartitions;
   private int recordCount;
   private final IntVector partitionIdVector;
   private final List<TransferPair> transfers;
 
   public RangePartitionRecordBatch(RangePartitionSender popConfig, RecordBatch incoming, FragmentContext context) throws OutOfMemoryException {
     super(popConfig, context, incoming);
-    this.numPartitions = popConfig.getDestinations().size();
+    numPartitions = popConfig.getDestinations().size();
 
     SchemaPath outputPath = popConfig.getPartitionFunction().getPartitionFieldRef();
     MaterializedField outputField = MaterializedField.create(outputPath.getAsNamePart().getName(), Types.required(TypeProtos.MinorType.INT));
-    this.partitionIdVector = (IntVector) TypeHelper.getNewVector(outputField, oContext.getAllocator());
-    this.transfers = Lists.newArrayList();
+    partitionIdVector = (IntVector) TypeHelper.getNewVector(outputField, oContext.getAllocator());
+    transfers = Lists.newArrayList();
   }
 
-
   @Override
   public void close() {
     super.close();
@@ -91,10 +90,7 @@
       partitionIdVector.allocateNew(num);
 
       recordCount = projectRecords(num, 0);
-      for (VectorWrapper<?> v : container) {
-        ValueVector.Mutator m = v.getValueVector().getMutator();
-        m.setValueCount(recordCount);
-      }
+      container.setValueCount(recordCount);
     }
     // returning OK here is fine because the base class AbstractSingleRecordBatch
     // is handling the actual return status; thus if there was a new schema, the
@@ -169,8 +165,8 @@
    * @param firstOutputIndex
    * @return the number of records projected
    */
-  private final int projectRecords(final int recordCount, int firstOutputIndex) {
-    final int countN = recordCount;
+  private final int projectRecords(int recordCount, int firstOutputIndex) {
+    int countN = recordCount;
     int counter = 0;
     for (int i = 0; i < countN; i++, firstOutputIndex++) {
       int partition = getPartition(i);
@@ -183,7 +179,6 @@
     return counter;
   }
 
-
   @Override
   public void dump() {
     logger.error("RangePartitionRecordBatch[container={}, numPartitions={}, recordCount={}, partitionIdVector={}]",
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/AbstractSV4Copier.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/AbstractSV4Copier.java
index 970f970..1394938 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/AbstractSV4Copier.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/AbstractSV4Copier.java
@@ -23,8 +23,10 @@
 import org.apache.drill.exec.record.selection.SelectionVector4;
 
 public abstract class AbstractSV4Copier extends AbstractCopier {
-  // Storing VectorWrapper reference instead of ValueVector[]. With EMIT outcome support underlying operator
-  // operator can generate multiple output batches with no schema changes which will change the ValueVector[]
+  // Storing VectorWrapper reference instead of ValueVector[]. With EMIT outcome
+  // support underlying operator
+  // operator can generate multiple output batches with no schema changes which
+  // will change the ValueVector[]
   // reference but not VectorWrapper reference.
   protected VectorWrapper<?>[] vvIn;
   private SelectionVector4 sv4;
@@ -32,21 +34,18 @@
   @Override
   public void setup(VectorAccessible incoming, VectorContainer outgoing) {
     super.setup(incoming, outgoing);
-    this.sv4 = incoming.getSelectionVector4();
+    sv4 = incoming.getSelectionVector4();
 
-    final int count = outgoing.getNumberOfColumns();
+    int count = outgoing.getNumberOfColumns();
     vvIn = new VectorWrapper[count];
 
-    {
-      int index = 0;
-
-      for (VectorWrapper vectorWrapper: incoming) {
-        vvIn[index] = vectorWrapper;
-        index++;
-      }
+    int index = 0;
+    for (VectorWrapper<?> vectorWrapper: incoming) {
+      vvIn[index++] = vectorWrapper;
     }
   }
 
+  @Override
   public void copyEntryIndirect(int inIndex, int outIndex) {
     copyEntry(sv4.get(inIndex), outIndex);
   }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/GenericCopier.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/GenericCopier.java
index f64a11e..0bb7186 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/GenericCopier.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/GenericCopier.java
@@ -42,7 +42,7 @@
     {
       int index = 0;
 
-      for (VectorWrapper vectorWrapper: incoming) {
+      for (VectorWrapper<?> vectorWrapper: incoming) {
         vvIn[index] = vectorWrapper.getValueVector();
         index++;
       }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/GenericSV4Copier.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/GenericSV4Copier.java
index e81eb43..1117ab3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/GenericSV4Copier.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/GenericSV4Copier.java
@@ -27,7 +27,7 @@
 
   public GenericSV4Copier(RecordBatch incomingBatch, VectorContainer outputContainer,
                           SchemaChangeCallBack callBack) {
-    for(VectorWrapper<?> vv : incomingBatch){
+    for (VectorWrapper<?> vv : incomingBatch) {
       ValueVector v = vv.getValueVectors()[0];
       v.makeTransferPair(outputContainer.addOrGet(v.getField(), callBack));
     }
@@ -37,7 +37,7 @@
   public void copyEntry(int inIndex, int outIndex) {
     int inOffset = inIndex & 0xFFFF;
     int inVector = inIndex >>> 16;
-    for ( int i = 0;  i < vvIn.length;  i++ ) {
+    for (int i = 0;  i < vvIn.length;  i++) {
       ValueVector[] vectorsFromIncoming = vvIn[i].getValueVectors();
       vvOut[i].copyEntry(outIndex, vectorsFromIncoming[inVector], inOffset);
     }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java
index d45157a..4471248 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java
@@ -43,14 +43,18 @@
 
   @Override
   protected boolean setupNewSchema() throws SchemaChangeException {
-    // Don't clear off container just because an OK_NEW_SCHEMA was received from upstream. For cases when there is just
-    // change in container type but no actual schema change, RemovingRecordBatch should consume OK_NEW_SCHEMA and
-    // send OK to downstream instead. Since the output of RemovingRecordBatch is always going to be a regular container
+    // Don't clear off container just because an OK_NEW_SCHEMA was received from
+    // upstream. For cases when there is just
+    // change in container type but no actual schema change, RemovingRecordBatch
+    // should consume OK_NEW_SCHEMA and
+    // send OK to downstream instead. Since the output of RemovingRecordBatch is
+    // always going to be a regular container
     // change in incoming container type is not actual schema change.
     container.zeroVectors();
     copier = GenericCopierFactory.createAndSetupCopier(incoming, container, callBack);
 
-    // If there is an actual schema change then below condition will be true and it will send OK_NEW_SCHEMA
+    // If there is an actual schema change then below condition will be true and
+    // it will send OK_NEW_SCHEMA
     // downstream too
     if (container.isSchemaChanged()) {
       container.buildSchema(SelectionVectorMode.NONE);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java
index d56f848..dca23cf 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java
@@ -23,7 +23,6 @@
 import org.apache.drill.common.exceptions.ExecutionSetupException;
 import org.apache.drill.exec.ExecConstants;
 import org.apache.drill.exec.cache.VectorAccessibleSerializable;
-import org.apache.drill.exec.exception.SchemaChangeException;
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.ops.FragmentContext;
 import org.apache.drill.exec.physical.config.Trace;
@@ -38,24 +37,27 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-/** TraceRecordBatch contains value vectors which are exactly the same
+/**
+ * Contains value vectors which are exactly the same
  * as the incoming record batch's value vectors. If the incoming
  * record batch has a selection vector (type 2) then TraceRecordBatch
  * will also contain a selection vector.
- *
+ * <p>
  * Purpose of this record batch is to dump the data associated with all
  * the value vectors and selection vector to disk.
- *
+ * <p>
  * This record batch does not modify any data or schema, it simply
  * consumes the incoming record batch's data, dump to disk and pass the
  * same set of value vectors (and selection vectors) to its parent record
- * batch
+ * batch.
  */
 public class TraceRecordBatch extends AbstractSingleRecordBatch<Trace> {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TraceRecordBatch.class);
+  static final Logger logger = LoggerFactory.getLogger(TraceRecordBatch.class);
 
-  private SelectionVector2 sv = null;
+  private SelectionVector2 sv;
 
   private final BufferAllocator localAllocator;
 
@@ -75,13 +77,14 @@
     localAllocator = context.getNewChildAllocator("trace", 200, 0, Long.MAX_VALUE);
     String fileName = getFileName();
 
-    /* Create the log file we will dump to and initialize the file descriptors */
+    // Create the log file we will dump to and initialize the file descriptors
     try {
       Configuration conf = new Configuration();
-      conf.set(FileSystem.FS_DEFAULT_NAME_KEY, context.getConfig().getString(ExecConstants.TRACE_DUMP_FILESYSTEM));
+      conf.set(FileSystem.FS_DEFAULT_NAME_KEY,
+          context.getConfig().getString(ExecConstants.TRACE_DUMP_FILESYSTEM));
       FileSystem fs = FileSystem.get(conf);
 
-      /* create the file */
+      // create the file
       fos = fs.create(new Path(fileName));
     } catch (IOException e) {
         throw new ExecutionSetupException("Unable to create file: " + fileName + " check permissions or if directory exists", e);
@@ -91,15 +94,15 @@
   @Override
   public int getRecordCount() {
     if (sv == null) {
-      return incoming.getRecordCount();
+      return container.getRecordCount();
     } else {
       return sv.getCount();
     }
   }
 
   /**
-   * Function is invoked for every record batch and it simply dumps the buffers associated with all the value vectors in
-   * this record batch to a log file.
+   * Invoked for every record batch and it simply dumps the buffers
+   * associated with all the value vectors in this record batch to a log file.
    */
   @Override
   protected IterOutcome doWork() {
@@ -110,7 +113,7 @@
     } else {
       sv = null;
     }
-    WritableBatch batch = WritableBatch.getBatchNoHVWrap(incoming.getRecordCount(), incoming, incomingHasSv2);
+    WritableBatch batch = WritableBatch.getBatchNoHVWrap(incoming.getContainer().getRecordCount(), incoming, incomingHasSv2);
     VectorAccessibleSerializable wrap = new VectorAccessibleSerializable(batch, sv, oContext.getAllocator());
 
     try {
@@ -121,28 +124,31 @@
     batch.reconstructContainer(localAllocator, container);
     if (incomingHasSv2) {
       sv = wrap.getSv2();
+      container.setRecordCount(sv.getBatchActualRecordCount());
+    } else {
+      container.setRecordCount(batch.getDef().getRecordCount());
     }
     return getFinalOutcome(false);
   }
 
   @Override
-  protected boolean setupNewSchema() throws SchemaChangeException {
-    /* Trace operator does not deal with hyper vectors yet */
+  protected boolean setupNewSchema() {
+    // Trace operator does not deal with hyper vectors yet
     if (incoming.getSchema().getSelectionVectorMode() == SelectionVectorMode.FOUR_BYTE) {
-      throw new SchemaChangeException("Trace operator does not work with hyper vectors");
+      throw new UnsupportedOperationException("Trace operator does not work with hyper vectors");
     }
 
-    /*
-     * we have a new schema, clear our existing container to load the new value vectors
-     */
+    // we have a new schema, clear our existing container to load the new value vectors
+
     container.clear();
 
-    /* Add all the value vectors in the container */
+    // Add all the value vectors in the container
     for (VectorWrapper<?> vv : incoming) {
       TransferPair tp = vv.getValueVector().getTransferPair(oContext.getAllocator());
       container.add(tp.getTo());
     }
     container.buildSchema(incoming.getSchema().getSelectionVectorMode());
+    container.setRecordCount(incoming.getRecordCount());
     return true;
   }
 
@@ -157,12 +163,12 @@
 
   @Override
   public void close() {
-    /* Release the selection vector */
+    // Release the selection vector
     if (sv != null) {
       sv.clear();
     }
 
-    /* Close the file descriptors */
+    // Close the file descriptors
     try {
       fos.close();
     } catch (IOException e) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java
index 8c9bb47..25dae80 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java
@@ -338,11 +338,10 @@
     public int getRemainingRecords() {
       return (totalRecordsToProcess - recordsProcessed);
     }
-
   }
 
   private class UnionInputIterator implements Iterator<Pair<IterOutcome, BatchStatusWrappper>> {
-    private final Stack<BatchStatusWrappper> batchStatusStack = new Stack<>();
+    private Stack<BatchStatusWrappper> batchStatusStack = new Stack<>();
 
     UnionInputIterator(IterOutcome leftOutCome, RecordBatch left, IterOutcome rightOutCome, RecordBatch right) {
       if (rightOutCome == IterOutcome.OK_NEW_SCHEMA) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
index fa8c954..1715c99 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
@@ -374,8 +374,7 @@
     TransferPair tp = resetUnnestTransferPair();
     container.add(TypeHelper.getNewVector(tp.getTo().getField(), oContext.getAllocator()));
     container.buildSchema(SelectionVectorMode.NONE);
-    container.allocateNew();
-    container.setValueCount(0);
+    container.setEmpty();
     return true;
   }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
index a6ebef0..d40bd6d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
@@ -51,9 +51,11 @@
 import org.apache.drill.exec.rpc.RpcOutcomeListener;
 import org.apache.drill.exec.testing.ControlsInjector;
 import org.apache.drill.exec.testing.ControlsInjectorFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class UnorderedReceiverBatch implements CloseableRecordBatch {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnorderedReceiverBatch.class);
+  private static final Logger logger = LoggerFactory.getLogger(UnorderedReceiverBatch.class);
   private static final ControlsInjector injector = ControlsInjectorFactory.getInjector(UnorderedReceiverBatch.class);
 
   private final RecordBatchLoader batchLoader;
@@ -78,11 +80,15 @@
     }
   }
 
-  public UnorderedReceiverBatch(final ExchangeFragmentContext context, final RawFragmentBatchProvider fragProvider, final UnorderedReceiver config) throws OutOfMemoryException {
+  public UnorderedReceiverBatch(ExchangeFragmentContext context,
+      RawFragmentBatchProvider fragProvider, UnorderedReceiver config)
+          throws OutOfMemoryException {
     this.fragProvider = fragProvider;
     this.context = context;
-    // In normal case, batchLoader does not require an allocator. However, in case of splitAndTransfer of a value vector,
-    // we may need an allocator for the new offset vector. Therefore, here we pass the context's allocator to batchLoader.
+    // In normal case, batchLoader does not require an allocator. However, in
+    // case of splitAndTransfer of a value vector,
+    // we may need an allocator for the new offset vector. Therefore, here we
+    // pass the context's allocator to batchLoader.
     oContext = context.newOperatorContext(config);
     this.batchLoader = new RecordBatchLoader(oContext.getAllocator());
 
@@ -90,8 +96,10 @@
     this.stats.setLongStat(Metric.NUM_SENDERS, config.getNumSenders());
     this.config = config;
 
-    // Register this operator's buffer allocator so that incoming buffers are owned by this allocator
-    context.getBuffers().getCollector(config.getOppositeMajorFragmentId()).setAllocator(oContext.getAllocator());
+    // Register this operator's buffer allocator so that incoming buffers are
+    // owned by this allocator
+    context.getBuffers().getCollector(config.getOppositeMajorFragmentId())
+      .setAllocator(oContext.getAllocator());
   }
 
   @Override
@@ -110,7 +118,7 @@
   }
 
   @Override
-  public void kill(final boolean sendUpstream) {
+  public void kill(boolean sendUpstream) {
     if (sendUpstream) {
       informSenders();
     }
@@ -133,12 +141,12 @@
   }
 
   @Override
-  public TypedFieldId getValueVectorId(final SchemaPath path) {
+  public TypedFieldId getValueVectorId(SchemaPath path) {
     return batchLoader.getValueVectorId(path);
   }
 
   @Override
-  public VectorWrapper<?> getValueAccessorById(final Class<?> clazz, final int... ids) {
+  public VectorWrapper<?> getValueAccessorById(Class<?> clazz, int... ids) {
     return batchLoader.getValueAccessorById(clazz, ids);
   }
 
@@ -146,8 +154,9 @@
     try {
       injector.injectInterruptiblePause(context.getExecutionControls(), "waiting-for-data", logger);
       return fragProvider.getNext();
-    } catch(final InterruptedException e) {
-      // Preserve evidence that the interruption occurred so that code higher up on the call stack can learn of the
+    } catch(InterruptedException e) {
+      // Preserve evidence that the interruption occurred so that code higher up
+      // on the call stack can learn of the
       // interruption and respond to it if it wants to.
       Thread.currentThread().interrupt();
 
@@ -190,8 +199,8 @@
         return lastOutcome;
       }
 
-      final RecordBatchDef rbd = batch.getHeader().getDef();
-      final boolean schemaChanged = batchLoader.load(rbd, batch.getBody());
+      RecordBatchDef rbd = batch.getHeader().getDef();
+      boolean schemaChanged = batchLoader.load(rbd, batch.getBody());
       // TODO:  Clean:  DRILL-2933:  That load(...) no longer throws
       // SchemaChangeException, so check/clean catch clause below.
       stats.addLongStat(Metric.BYTES_RECEIVED, batch.getByteCount());
@@ -230,7 +239,9 @@
 
   @Override
   public VectorContainer getOutgoingContainer() {
-    throw new UnsupportedOperationException(String.format(" You should not call getOutgoingContainer() for class %s", this.getClass().getCanonicalName()));
+    throw new UnsupportedOperationException(
+        String.format("You should not call getOutgoingContainer() for class %s",
+            getClass().getCanonicalName()));
   }
 
   @Override
@@ -240,15 +251,15 @@
 
   private void informSenders() {
     logger.info("Informing senders of request to terminate sending.");
-    final FragmentHandle handlePrototype = FragmentHandle.newBuilder()
+    FragmentHandle handlePrototype = FragmentHandle.newBuilder()
             .setMajorFragmentId(config.getOppositeMajorFragmentId())
             .setQueryId(context.getHandle().getQueryId())
             .build();
-    for (final MinorFragmentEndpoint providingEndpoint : config.getProvidingEndpoints()) {
-      final FragmentHandle sender = FragmentHandle.newBuilder(handlePrototype)
+    for (MinorFragmentEndpoint providingEndpoint : config.getProvidingEndpoints()) {
+      FragmentHandle sender = FragmentHandle.newBuilder(handlePrototype)
               .setMinorFragmentId(providingEndpoint.getId())
               .build();
-      final FinishedReceiver finishedReceiver = FinishedReceiver.newBuilder()
+      FinishedReceiver finishedReceiver = FinishedReceiver.newBuilder()
               .setReceiver(context.getHandle())
               .setSender(sender)
               .build();
@@ -262,19 +273,19 @@
   private class OutcomeListener implements RpcOutcomeListener<Ack> {
 
     @Override
-    public void failed(final RpcException ex) {
+    public void failed(RpcException ex) {
       logger.warn("Failed to inform upstream that receiver is finished");
     }
 
     @Override
-    public void success(final Ack value, final ByteBuf buffer) {
+    public void success(Ack value, ByteBuf buffer) {
       // Do nothing
     }
 
     @Override
-    public void interrupted(final InterruptedException e) {
+    public void interrupted(InterruptedException e) {
       if (context.getExecutorState().shouldContinue()) {
-        final String errMsg = "Received an interrupt RPC outcome while sending ReceiverFinished message";
+        String errMsg = "Received an interrupt RPC outcome while sending ReceiverFinished message";
         logger.error(errMsg, e);
         context.getExecutorState().fail(new RpcException(errMsg, e));
       }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unpivot/UnpivotMapsRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unpivot/UnpivotMapsRecordBatch.java
index 0f3517f..99bd6d1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unpivot/UnpivotMapsRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unpivot/UnpivotMapsRecordBatch.java
@@ -31,12 +31,15 @@
 import org.apache.drill.exec.record.MaterializedField;
 import org.apache.drill.exec.record.RecordBatch;
 import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.record.VectorAccessibleUtilities;
 import org.apache.drill.exec.record.VectorContainer;
 import org.apache.drill.exec.record.VectorWrapper;
 import org.apache.drill.exec.vector.ValueVector;
 import org.apache.drill.exec.vector.complex.MapVector;
 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;
 
 /**
  * Unpivot maps. Assumptions are:
@@ -44,7 +47,7 @@
  *  2) Each map contains the same number of fields and field names are also same (types could be different).
  *
  * Example input and output:
- * Schema of input:
+ * Schema of input: <pre>
  *    "schema"        : BIGINT - Schema number. For each schema change this number is incremented.
  *    "computed"      : BIGINT - What time is it computed?
  *    "columns" : MAP - Column names
@@ -69,22 +72,23 @@
  *  "statscount"       : BIGINT
  *  "nonnullstatcount" : BIGINT
  *  .... one column for each map type ...
+ *  </pre>
  */
 public class UnpivotMapsRecordBatch extends AbstractSingleRecordBatch<UnpivotMaps> {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnpivotMapsRecordBatch.class);
+  private static final Logger logger = LoggerFactory.getLogger(UnpivotMapsRecordBatch.class);
 
   private final List<String> mapFieldsNames;
   private boolean first = true;
-  private int keyIndex = 0;
-  private List<String> keyList = null;
+  private int keyIndex;
+  private List<String> keyList;
 
-  private Map<MaterializedField, Map<String, ValueVector>> dataSrcVecMap = null;
+  private Map<MaterializedField, Map<String, ValueVector>> dataSrcVecMap;
 
   // Map of non-map fields to VV in the incoming schema
-  private Map<MaterializedField, ValueVector> copySrcVecMap = null;
+  private Map<MaterializedField, ValueVector> copySrcVecMap;
 
   private List<TransferPair> transferList;
-  private int recordCount = 0;
+  private int recordCount;
 
   public UnpivotMapsRecordBatch(UnpivotMaps pop, RecordBatch incoming, FragmentContext context)
       throws OutOfMemoryException {
@@ -150,6 +154,7 @@
     }
   }
 
+  @Override
   public VectorContainer getOutgoingContainer() {
     return this.container;
   }
@@ -170,11 +175,10 @@
 
     keyIndex = (keyIndex + 1) % keyList.size();
     recordCount = outRecordCount;
+    container.setRecordCount(recordCount);
 
     if (keyIndex == 0) {
-      for (VectorWrapper w : incoming) {
-        w.clear();
-      }
+      VectorAccessibleUtilities.clear(incoming.getContainer());
     }
     return IterOutcome.OK;
   }
@@ -270,6 +274,7 @@
     container.clear();
     buildKeyList();
     buildOutputContainer();
+    container.setEmpty();
     return true;
   }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/BatchValidator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/BatchValidator.java
index 5c4f8b4..39189b1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/BatchValidator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/BatchValidator.java
@@ -22,19 +22,31 @@
 
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.impl.ScanBatch;
+import org.apache.drill.exec.physical.impl.WriterRecordBatch;
+import org.apache.drill.exec.physical.impl.TopN.TopNBatch;
 import org.apache.drill.exec.physical.impl.aggregate.HashAggBatch;
 import org.apache.drill.exec.physical.impl.aggregate.StreamingAggBatch;
 import org.apache.drill.exec.physical.impl.filter.FilterRecordBatch;
 import org.apache.drill.exec.physical.impl.filter.RuntimeFilterRecordBatch;
 import org.apache.drill.exec.physical.impl.flatten.FlattenRecordBatch;
+import org.apache.drill.exec.physical.impl.join.HashJoinBatch;
 import org.apache.drill.exec.physical.impl.join.MergeJoinBatch;
 import org.apache.drill.exec.physical.impl.join.NestedLoopJoinBatch;
 import org.apache.drill.exec.physical.impl.limit.LimitRecordBatch;
 import org.apache.drill.exec.physical.impl.limit.PartitionLimitRecordBatch;
+import org.apache.drill.exec.physical.impl.mergereceiver.MergingRecordBatch;
+import org.apache.drill.exec.physical.impl.orderedpartitioner.OrderedPartitionRecordBatch;
 import org.apache.drill.exec.physical.impl.project.ProjectRecordBatch;
 import org.apache.drill.exec.physical.impl.protocol.OperatorRecordBatch;
+import org.apache.drill.exec.physical.impl.rangepartitioner.RangePartitionRecordBatch;
 import org.apache.drill.exec.physical.impl.svremover.RemovingRecordBatch;
+import org.apache.drill.exec.physical.impl.trace.TraceRecordBatch;
+import org.apache.drill.exec.physical.impl.union.UnionAllRecordBatch;
 import org.apache.drill.exec.physical.impl.unnest.UnnestRecordBatch;
+import org.apache.drill.exec.physical.impl.unorderedreceiver.UnorderedReceiverBatch;
+import org.apache.drill.exec.physical.impl.unpivot.UnpivotMapsRecordBatch;
+import org.apache.drill.exec.physical.impl.window.WindowFrameRecordBatch;
+import org.apache.drill.exec.physical.impl.xsort.managed.ExternalSortBatch;
 import org.apache.drill.exec.record.CloseableRecordBatch;
 import org.apache.drill.exec.record.RecordBatch;
 import org.apache.drill.exec.record.SimpleVectorWrapper;
@@ -223,6 +235,18 @@
     rules.put(MergeJoinBatch.class, CheckMode.VECTORS);
     rules.put(NestedLoopJoinBatch.class, CheckMode.VECTORS);
     rules.put(LimitRecordBatch.class, CheckMode.VECTORS);
+    rules.put(MergingRecordBatch.class, CheckMode.VECTORS);
+    rules.put(OrderedPartitionRecordBatch.class, CheckMode.VECTORS);
+    rules.put(RangePartitionRecordBatch.class, CheckMode.VECTORS);
+    rules.put(TraceRecordBatch.class, CheckMode.VECTORS);
+    rules.put(UnionAllRecordBatch.class, CheckMode.VECTORS);
+    rules.put(UnorderedReceiverBatch.class, CheckMode.VECTORS);
+    rules.put(UnpivotMapsRecordBatch.class, CheckMode.VECTORS);
+    rules.put(WindowFrameRecordBatch.class, CheckMode.VECTORS);
+    rules.put(TopNBatch.class, CheckMode.VECTORS);
+    rules.put(HashJoinBatch.class, CheckMode.VECTORS);
+    rules.put(ExternalSortBatch.class, CheckMode.VECTORS);
+    rules.put(WriterRecordBatch.class, CheckMode.VECTORS);
     return rules;
   }
 
@@ -518,9 +542,13 @@
       // empty map vector that causes validation to fail.
       ValueVector child = internalMap.getChild(type.name());
       if (child == null) {
-        error(name, vector, String.format(
-            "Union vector includes type %s, but the internal map has no matching member",
-            type.name()));
+        // Disabling this check for now. TopNBatch, SortBatch
+        // and perhaps others will create vectors with a set of
+        // types, but won't actually populate some of the types.
+        //
+        // error(name, vector, String.format(
+        //     "Union vector includes type %s, but the internal map has no matching member",
+        //     type.name()));
       } else {
         validateVector(name + "-type-" + type.name(),
             valueCount, child);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/NoFrameSupportTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/NoFrameSupportTemplate.java
index cc7a04d..a759399 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/NoFrameSupportTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/NoFrameSupportTemplate.java
@@ -26,18 +26,22 @@
 import org.apache.drill.exec.record.VectorWrapper;
 import org.apache.drill.exec.vector.BaseDataValueVector;
 import org.apache.drill.exec.vector.ValueVector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.inject.Named;
 import java.util.List;
 
 
 /**
- * WindowFramer implementation that doesn't support the FRAME clause (will assume the default frame).
- * <br>According to the SQL standard, LEAD, LAG, ROW_NUMBER, NTILE and all ranking functions don't support the FRAME clause.
- * This class will handle such functions.
+ * WindowFramer implementation that doesn't support the FRAME clause (will
+ * assume the default frame). <p>
+ * According to the SQL standard, LEAD, LAG, ROW_NUMBER, NTILE and all ranking
+ * functions don't support the FRAME clause. This class will handle such
+ * functions.
  */
 public abstract class NoFrameSupportTemplate implements WindowFramer {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(NoFrameSupportTemplate.class);
+  private static final Logger logger = LoggerFactory.getLogger(NoFrameSupportTemplate.class);
 
   private VectorContainer container;
   private VectorContainer internal;
@@ -53,8 +57,8 @@
   private Partition partition; // current partition being processed
 
   @Override
-  public void setup(final List<WindowDataBatch> batches, final VectorContainer container, final OperatorContext oContext,
-                    final boolean requireFullPartition, final WindowPOP popConfig) throws SchemaChangeException {
+  public void setup(List<WindowDataBatch> batches, VectorContainer container, OperatorContext oContext,
+                    boolean requireFullPartition, WindowPOP popConfig) throws SchemaChangeException {
     this.container = container;
     this.batches = batches;
 
@@ -81,9 +85,7 @@
   @Override
   public void doWork() throws DrillException {
     int currentRow = 0;
-
-    this.current = batches.get(0);
-
+    current = batches.get(0);
     outputCount = current.getRecordCount();
 
     while (currentRow < outputCount) {
@@ -108,10 +110,9 @@
     }
   }
 
-  private void newPartition(final WindowDataBatch current, final int currentRow) throws SchemaChangeException {
+  private void newPartition(WindowDataBatch current, int currentRow) throws SchemaChangeException {
     partition = new Partition();
     updatePartitionSize(partition, currentRow);
-
     setupPartition(current, container);
   }
 
@@ -131,16 +132,19 @@
   }
 
   /**
-   * process all rows (computes and writes function values) of current batch that are part of current partition.
-   * @param currentRow first unprocessed row
+   * Process all rows (computes and writes function values) of current batch
+   * that are part of current partition.
+   *
+   * @param currentRow
+   *          first unprocessed row
    * @return index of next unprocessed row
-   * @throws DrillException if it can't write into the container
+   * @throws DrillException
+   *           if it can't write into the container
    */
-  private int processPartition(final int currentRow) throws DrillException {
+  private int processPartition(int currentRow) throws DrillException {
     logger.trace("process partition {}, currentRow: {}, outputCount: {}", partition, currentRow, outputCount);
 
     setupCopyNext(current, container);
-
     copyPrevFromInternal();
 
     // copy remaining from current
@@ -198,15 +202,14 @@
     }
   }
 
-  private void processRow(final int row) throws DrillException {
+  private void processRow(int row) throws DrillException {
     if (partition.isFrameDone()) {
       // because all peer rows share the same frame, we only need to compute and aggregate the frame once
-      final long peers = countPeers(row);
+      long peers = countPeers(row);
       partition.newFrame(peers);
     }
 
     outputRow(row, partition);
-
     partition.rowAggregated();
   }
 
@@ -214,7 +217,7 @@
    * updates partition's length after computing the number of rows for the current the partition starting at the specified
    * row of the first batch. If !requiresFullPartition, this method will only count the rows in the current batch
    */
-  private void updatePartitionSize(final Partition partition, final int start) {
+  private void updatePartitionSize(Partition partition, int start) {
     logger.trace("compute partition size starting from {} on {} batches", start, batches.size());
 
     long length = 0;
@@ -226,7 +229,7 @@
 
     outer:
     for (WindowDataBatch batch : batches) {
-      final int recordCount = batch.getRecordCount();
+      int recordCount = batch.getRecordCount();
 
       // check first container from start row, and subsequent containers from first row
       for (; row < recordCount; row++, length++) {
@@ -262,18 +265,18 @@
   }
 
   /**
-   * count number of peer rows for current row
+   * Count number of peer rows for current row
    * @param start starting row of the current frame
    * @return num peer rows for current row
    * @throws SchemaChangeException
    */
-  private long countPeers(final int start) throws SchemaChangeException {
+  private long countPeers(int start) throws SchemaChangeException {
     long length = 0;
 
     // a single frame can include rows from multiple batches
     // start processing first batch and, if necessary, move to next batches
     for (WindowDataBatch batch : batches) {
-      final int recordCount = batch.getRecordCount();
+      int recordCount = batch.getRecordCount();
 
       // for every remaining row in the partition, count it if it's a peer row
       for (int row = (batch == current) ? start : 0; row < recordCount; row++, length++) {
@@ -308,12 +311,14 @@
         + "]";
   }
 
-
   /**
-   * called once for each row after we evaluate all peer rows. Used to write a value in the row
+   * Called once for each row after we evaluate all peer rows. Used to write a
+   * value in the row
    *
-   * @param outIndex index of row
-   * @param partition object used by "computed" window functions
+   * @param outIndex
+   *          index of row
+   * @param partition
+   *          object used by "computed" window functions
    */
   public abstract void outputRow(@Named("outIndex") int outIndex,
                                  @Named("partition") Partition partition)
@@ -331,10 +336,13 @@
                        throws SchemaChangeException;
 
   /**
-   * copies value(s) from inIndex row to outIndex row. Mostly used by LEAD. inIndex always points to the row next to
-   * outIndex
-   * @param inIndex source row of the copy
-   * @param outIndex destination row of the copy.
+   * Copies value(s) from inIndex row to outIndex row. Mostly used by LEAD.
+   * inIndex always points to the row next to outIndex
+   *
+   * @param inIndex
+   *          source row of the copy
+   * @param outIndex
+   *          destination row of the copy.
    */
   public abstract void copyNext(@Named("inIndex") int inIndex,
                                 @Named("outIndex") int outIndex)
@@ -344,10 +352,13 @@
                        throws SchemaChangeException;
 
   /**
-   * copies value(s) from inIndex row to outIndex row. Mostly used by LAG. inIndex always points to the previous row
+   * Copies value(s) from inIndex row to outIndex row. Mostly used by LAG.
+   * inIndex always points to the previous row
    *
-   * @param inIndex source row of the copy
-   * @param outIndex destination row of the copy.
+   * @param inIndex
+   *          source row of the copy
+   * @param outIndex
+   *          destination row of the copy.
    */
   public abstract void copyPrev(@Named("inIndex") int inIndex,
                                 @Named("outIndex") int outIndex)
@@ -364,12 +375,12 @@
                        throws SchemaChangeException;
 
   /**
-   * reset all window functions
+   * Reset all window functions
    */
   public abstract boolean resetValues() throws SchemaChangeException;
 
   /**
-   * compares two rows from different batches (can be the same), if they have the same value for the partition by
+   * Compares two rows from different batches (can be the same), if they have the same value for the partition by
    * expression
    * @param b1Index index of first row
    * @param b1 batch for first row
@@ -385,7 +396,7 @@
                           throws SchemaChangeException;
 
   /**
-   * compares two rows from different batches (can be the same), if they have the same value for the order by
+   * Compares two rows from different batches (can be the same), if they have the same value for the order by
    * expression
    * @param b1Index index of first row
    * @param b1 batch for first row
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/Partition.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/Partition.java
index 55fa647..a7d9a8d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/Partition.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/Partition.java
@@ -17,13 +17,17 @@
  */
 package org.apache.drill.exec.physical.impl.window;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
- * Used internally to keep track of partitions and frames.<br>
- * A partition can be partial, which means we don't know "yet" the total number of records that are part of this partition.
- * Even for partial partitions, we know the number of rows that are part of current frame
+ * Used internally to keep track of partitions and frames.<p>
+ * A partition can be partial, which means we don't know "yet" the total number
+ * of records that are part of this partition. Even for partial partitions, we
+ * know the number of rows that are part of current frame
  */
 public class Partition {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Partition.class);
+  private static final Logger logger = LoggerFactory.getLogger(Partition.class);
 
   private boolean partial; // true if we don't know yet the full length of this partition
   private long length; // size of this partition (if partial is true, then this is a partial length of the partition)
@@ -48,10 +52,13 @@
   public long getLength() {
     return length;
   }
+
   /**
-   * @param length number of rows in this partition
-   * @param partial if true, then length is not the full length of the partition but just the number of rows in the
-   *                current batch
+   * @param length
+   *          number of rows in this partition
+   * @param partial
+   *          if true, then length is not the full length of the partition but
+   *          just the number of rows in the current batch
    */
   public void updateLength(long length, boolean partial) {
     this.length += length;
@@ -62,7 +69,6 @@
   public void rowAggregated() {
     remaining--;
     peers--;
-
     row_number++;
   }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameRecordBatch.java
index 59e84ef..6ed004f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameRecordBatch.java
@@ -21,8 +21,6 @@
 import java.util.Arrays;
 import java.util.List;
 
-import org.apache.drill.shaded.guava.com.google.common.collect.Iterables;
-
 import org.apache.drill.common.exceptions.DrillException;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.FunctionCall;
@@ -47,23 +45,25 @@
 import org.apache.drill.exec.record.TransferPair;
 import org.apache.drill.exec.record.VectorAccessible;
 import org.apache.drill.exec.record.VectorWrapper;
-
-import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
-import com.sun.codemodel.JExpr;
 import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.shaded.guava.com.google.common.collect.Iterables;
+import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.sun.codemodel.JExpr;
 
 /**
  * support for OVER(PARTITION BY expression1,expression2,... [ORDER BY expressionA, expressionB,...])
  *
  */
 public class WindowFrameRecordBatch extends AbstractRecordBatch<WindowPOP> {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(WindowFrameRecordBatch.class);
+  static final Logger logger = LoggerFactory.getLogger(WindowFrameRecordBatch.class);
 
   private final RecordBatch incoming;
   private List<WindowDataBatch> batches;
 
   private WindowFramer[] framers;
-  private boolean hasOrderBy; // true if window definition contains an order-by clause
   private final List<WindowFunction> functions = Lists.newArrayList();
 
   private boolean noMoreBatches; // true when downstream returns NONE
@@ -71,14 +71,16 @@
 
   private boolean shouldStop; // true if we received an early termination request
 
-  public WindowFrameRecordBatch(WindowPOP popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
+  public WindowFrameRecordBatch(WindowPOP popConfig, FragmentContext context,
+      RecordBatch incoming) throws OutOfMemoryException {
     super(popConfig, context);
     this.incoming = incoming;
     batches = Lists.newArrayList();
   }
 
   /**
-   * Hold incoming batches in memory until all window functions are ready to process the batch on top of the queue
+   * Hold incoming batches in memory until all window functions are ready to
+   * process the batch on top of the queue
    */
   @Override
   public IterOutcome innerNext() {
@@ -104,7 +106,8 @@
       return IterOutcome.NONE;
     }
 
-    // keep saving incoming batches until the first unprocessed batch can be processed, or upstream == NONE
+    // keep saving incoming batches until the first unprocessed batch can be
+    // processed, or upstream == NONE
     while (!noMoreBatches && !canDoWork()) {
       IterOutcome upstream = next(incoming);
       logger.trace("next(incoming) returned {}", upstream);
@@ -124,7 +127,7 @@
             if (schema != null) {
               throw new UnsupportedOperationException("OVER clause doesn't currently support changing schemas.");
             }
-            this.schema = incoming.getSchema();
+            schema = incoming.getSchema();
           }
         case OK:
           if (incoming.getRecordCount() > 0) {
@@ -160,15 +163,13 @@
 
   private void doWork() throws DrillException {
 
-    final WindowDataBatch current = batches.get(0);
-    final int recordCount = current.getRecordCount();
+    WindowDataBatch current = batches.get(0);
+    int recordCount = current.getRecordCount();
 
     logger.trace("WindowFramer.doWork() START, num batches {}, current batch has {} rows", batches.size(), recordCount);
 
     // allocate outgoing vectors
-    for (VectorWrapper<?> w : container) {
-      w.getValueVector().allocateNew();
-    }
+    container.allocateNew();
 
     for (WindowFramer framer : framers) {
       framer.doWork();
@@ -181,10 +182,7 @@
       tp.transfer();
     }
 
-    container.setRecordCount(recordCount);
-    for (VectorWrapper<?> v : container) {
-      v.getValueVector().getMutator().setValueCount(recordCount);
-    }
+    container.setValueCount(recordCount);
 
     // we can safely free the current batch
     current.clear();
@@ -194,8 +192,8 @@
   }
 
   /**
-   * @return true when all window functions are ready to process the current batch (it's the first batch currently
-   * held in memory)
+   * @return true when all window functions are ready to process the current
+   *         batch (it's the first batch currently held in memory)
    */
   private boolean canDoWork() {
     if (batches.size() < 2) {
@@ -204,10 +202,10 @@
       return false;
     }
 
-    final VectorAccessible current = batches.get(0);
-    final int currentSize = current.getRecordCount();
-    final VectorAccessible last = batches.get(batches.size() - 1);
-    final int lastSize = last.getRecordCount();
+    VectorAccessible current = batches.get(0);
+    int currentSize = current.getRecordCount();
+    VectorAccessible last = batches.get(batches.size() - 1);
+    int lastSize = last.getRecordCount();
 
     boolean partitionEndReached;
     boolean frameEndReached;
@@ -215,7 +213,7 @@
       partitionEndReached = !framers[0].isSamePartition(currentSize - 1, current, lastSize - 1, last);
       frameEndReached = partitionEndReached || !framers[0].isPeer(currentSize - 1, current, lastSize - 1, last);
 
-      for (final WindowFunction function : functions) {
+      for (WindowFunction function : functions) {
         if (!function.canDoWork(batches.size(), popConfig, frameEndReached, partitionEndReached)) {
           return false;
         }
@@ -230,18 +228,20 @@
   @Override
   protected void buildSchema() throws SchemaChangeException {
     logger.trace("buildSchema()");
-    final IterOutcome outcome = next(incoming);
+    IterOutcome outcome = next(incoming);
     switch (outcome) {
-      case NONE:
-        state = BatchState.DONE;
-        container.buildSchema(BatchSchema.SelectionVectorMode.NONE);
-        return;
-      case STOP:
-        state = BatchState.STOP;
-        return;
-      case OUT_OF_MEMORY:
-        state = BatchState.OUT_OF_MEMORY;
-        return;
+    case NONE:
+      state = BatchState.DONE;
+      container.buildSchema(BatchSchema.SelectionVectorMode.NONE);
+      return;
+    case STOP:
+      state = BatchState.STOP;
+      return;
+    case OUT_OF_MEMORY:
+      state = BatchState.OUT_OF_MEMORY;
+      return;
+    default:
+      break;
     }
 
     try {
@@ -260,30 +260,28 @@
 
     logger.trace("creating framer(s)");
 
-    final List<LogicalExpression> keyExprs = Lists.newArrayList();
-    final List<LogicalExpression> orderExprs = Lists.newArrayList();
+    List<LogicalExpression> keyExprs = Lists.newArrayList();
+    List<LogicalExpression> orderExprs = Lists.newArrayList();
     boolean requireFullPartition = false;
 
     boolean useDefaultFrame = false; // at least one window function uses the DefaultFrameTemplate
     boolean useCustomFrame = false; // at least one window function uses the CustomFrameTemplate
 
-    hasOrderBy = popConfig.getOrderings().size() > 0;
-
     // all existing vectors will be transferred to the outgoing container in framer.doWork()
-    for (final VectorWrapper<?> wrapper : batch) {
+    for (VectorWrapper<?> wrapper : batch) {
       container.addOrGet(wrapper.getField());
     }
 
     // add aggregation vectors to the container, and materialize corresponding expressions
-    for (final NamedExpression ne : popConfig.getAggregations()) {
+    for (NamedExpression ne : popConfig.getAggregations()) {
       if (!(ne.getExpr() instanceof FunctionCall)) {
         throw UserException.functionError()
           .message("Unsupported window function '%s'", ne.getExpr())
           .build(logger);
       }
 
-      final FunctionCall call = (FunctionCall) ne.getExpr();
-      final WindowFunction winfun = WindowFunction.fromExpression(call);
+      FunctionCall call = (FunctionCall) ne.getExpr();
+      WindowFunction winfun = WindowFunction.fromExpression(call);
       if (winfun.materialize(ne, container, context.getFunctionRegistry())) {
         functions.add(winfun);
         requireFullPartition |= winfun.requiresFullPartition(popConfig);
@@ -300,12 +298,12 @@
     container.setRecordCount(0);
 
     // materialize partition by expressions
-    for (final NamedExpression ne : popConfig.getWithins()) {
+    for (NamedExpression ne : popConfig.getWithins()) {
       keyExprs.add(ExpressionTreeMaterializer.materializeAndCheckErrors(ne.getExpr(), batch, context.getFunctionRegistry()));
     }
 
     // materialize order by expressions
-    for (final Order.Ordering oe : popConfig.getOrderings()) {
+    for (Order.Ordering oe : popConfig.getOrderings()) {
       orderExprs.add(ExpressionTreeMaterializer.materializeAndCheckErrors(oe.getExpr(), batch, context.getFunctionRegistry()));
     }
 
@@ -328,31 +326,31 @@
     }
   }
 
-  private WindowFramer generateFramer(final List<LogicalExpression> keyExprs, final List<LogicalExpression> orderExprs,
-      final List<WindowFunction> functions, boolean useCustomFrame) throws IOException, ClassTransformationException {
+  private WindowFramer generateFramer(List<LogicalExpression> keyExprs, List<LogicalExpression> orderExprs,
+      List<WindowFunction> functions, boolean useCustomFrame) throws IOException, ClassTransformationException {
 
     TemplateClassDefinition<WindowFramer> definition = useCustomFrame ?
       WindowFramer.FRAME_TEMPLATE_DEFINITION : WindowFramer.NOFRAME_TEMPLATE_DEFINITION;
-    final ClassGenerator<WindowFramer> cg = CodeGenerator.getRoot(definition, context.getOptions());
+    ClassGenerator<WindowFramer> cg = CodeGenerator.getRoot(definition, context.getOptions());
 
     {
       // generating framer.isSamePartition()
-      final GeneratorMapping IS_SAME_PARTITION_READ = GeneratorMapping.create("isSamePartition", "isSamePartition", null, null);
-      final MappingSet isaB1 = new MappingSet("b1Index", null, "b1", null, IS_SAME_PARTITION_READ, IS_SAME_PARTITION_READ);
-      final MappingSet isaB2 = new MappingSet("b2Index", null, "b2", null, IS_SAME_PARTITION_READ, IS_SAME_PARTITION_READ);
+      GeneratorMapping IS_SAME_PARTITION_READ = GeneratorMapping.create("isSamePartition", "isSamePartition", null, null);
+      MappingSet isaB1 = new MappingSet("b1Index", null, "b1", null, IS_SAME_PARTITION_READ, IS_SAME_PARTITION_READ);
+      MappingSet isaB2 = new MappingSet("b2Index", null, "b2", null, IS_SAME_PARTITION_READ, IS_SAME_PARTITION_READ);
       setupIsFunction(cg, keyExprs, isaB1, isaB2);
     }
 
     {
       // generating framer.isPeer()
-      final GeneratorMapping IS_SAME_PEER_READ = GeneratorMapping.create("isPeer", "isPeer", null, null);
-      final MappingSet isaP1 = new MappingSet("b1Index", null, "b1", null, IS_SAME_PEER_READ, IS_SAME_PEER_READ);
-      final MappingSet isaP2 = new MappingSet("b2Index", null, "b2", null, IS_SAME_PEER_READ, IS_SAME_PEER_READ);
+      GeneratorMapping IS_SAME_PEER_READ = GeneratorMapping.create("isPeer", "isPeer", null, null);
+      MappingSet isaP1 = new MappingSet("b1Index", null, "b1", null, IS_SAME_PEER_READ, IS_SAME_PEER_READ);
+      MappingSet isaP2 = new MappingSet("b2Index", null, "b2", null, IS_SAME_PEER_READ, IS_SAME_PEER_READ);
       // isPeer also checks if it's the same partition
       setupIsFunction(cg, Iterables.concat(keyExprs, orderExprs), isaP1, isaP2);
     }
 
-    for (final WindowFunction function : functions) {
+    for (WindowFunction function : functions) {
       // only generate code for the proper window functions
       if (function.supportsCustomFrames() == useCustomFrame) {
         function.generateCode(cg);
@@ -363,7 +361,7 @@
     CodeGenerator<WindowFramer> codeGen = cg.getCodeGenerator();
     codeGen.plainJavaCapable(true);
     // Uncomment out this line to debug the generated code.
-//    codeGen.saveCodeForDebugging(true);
+    codeGen.saveCodeForDebugging(true);
 
     return context.getImplementationClass(codeGen);
   }
@@ -371,8 +369,8 @@
   /**
    * setup comparison functions isSamePartition and isPeer
    */
-  private void setupIsFunction(final ClassGenerator<WindowFramer> cg, final Iterable<LogicalExpression> exprs,
-                               final MappingSet leftMapping, final MappingSet rightMapping) {
+  private void setupIsFunction(ClassGenerator<WindowFramer> cg, Iterable<LogicalExpression> exprs,
+                               MappingSet leftMapping, MappingSet rightMapping) {
     cg.setMappingSet(leftMapping);
     for (LogicalExpression expr : exprs) {
       if (expr == null) {
@@ -384,10 +382,10 @@
       cg.setMappingSet(rightMapping);
       ClassGenerator.HoldingContainer second = cg.addExpr(expr, ClassGenerator.BlkCreateMode.FALSE);
 
-      final LogicalExpression fh =
+      LogicalExpression fh =
         FunctionGenerationHelper
           .getOrderingComparatorNullsHigh(first, second, context.getFunctionRegistry());
-      final ClassGenerator.HoldingContainer out = cg.addExpr(fh, ClassGenerator.BlkCreateMode.FALSE);
+      ClassGenerator.HoldingContainer out = cg.addExpr(fh, ClassGenerator.BlkCreateMode.FALSE);
       cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)))._then()._return(JExpr.FALSE);
     }
     cg.getEvalBlock()._return(JExpr.TRUE);
@@ -404,7 +402,7 @@
     }
 
     if (batches != null) {
-      for (final WindowDataBatch bd : batches) {
+      for (WindowDataBatch bd : batches) {
         bd.clear();
       }
       batches = null;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFramer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFramer.java
index a7964d6..4bb3d38 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFramer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFramer.java
@@ -29,11 +29,14 @@
 import java.util.List;
 
 public interface WindowFramer {
-  TemplateClassDefinition<WindowFramer> NOFRAME_TEMPLATE_DEFINITION = new TemplateClassDefinition<>(WindowFramer.class, NoFrameSupportTemplate.class);
-  TemplateClassDefinition<WindowFramer> FRAME_TEMPLATE_DEFINITION = new TemplateClassDefinition<>(WindowFramer.class, FrameSupportTemplate.class);
+  TemplateClassDefinition<WindowFramer> NOFRAME_TEMPLATE_DEFINITION =
+      new TemplateClassDefinition<>(WindowFramer.class, NoFrameSupportTemplate.class);
+  TemplateClassDefinition<WindowFramer> FRAME_TEMPLATE_DEFINITION =
+      new TemplateClassDefinition<>(WindowFramer.class, FrameSupportTemplate.class);
 
-  void setup(final List<WindowDataBatch> batches, final VectorContainer container, final OperatorContext operatorContext,
-             final boolean requireFullPartition, final WindowPOP popConfig) throws SchemaChangeException;
+  void setup(final List<WindowDataBatch> batches, final VectorContainer container,
+      final OperatorContext operatorContext, final boolean requireFullPartition,
+      final WindowPOP popConfig) throws SchemaChangeException;
 
   /**
    * process the inner batch and write the aggregated values in the container
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java
index 9587325..3006529 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java
@@ -28,23 +28,23 @@
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.ops.FragmentContext;
 import org.apache.drill.exec.record.VectorAccessible;
-import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.record.VectorAccessibleUtilities;
 import org.apache.drill.exec.record.selection.SelectionVector4;
-import org.apache.drill.exec.vector.AllocationHelper;
+
 
 public abstract class PriorityQueueCopierTemplate implements PriorityQueueCopier {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PriorityQueueCopierTemplate.class);
 
   private SelectionVector4 vector4;
   private List<BatchGroup> batchGroups;
   private VectorAccessible hyperBatch;
   private VectorAccessible outgoing;
   private int size;
-  private int queueSize = 0;
+  private int queueSize;
 
   @Override
-  public void setup(FragmentContext context, BufferAllocator allocator, VectorAccessible hyperBatch, List<BatchGroup> batchGroups,
-                    VectorAccessible outgoing) throws SchemaChangeException {
+  public void setup(FragmentContext context, BufferAllocator allocator,
+      VectorAccessible hyperBatch, List<BatchGroup> batchGroups,
+      VectorAccessible outgoing) throws SchemaChangeException {
     this.hyperBatch = hyperBatch;
     this.batchGroups = batchGroups;
     this.outgoing = outgoing;
@@ -90,20 +90,14 @@
   }
 
   private void setValueCount(int count) {
-    for (VectorWrapper w: outgoing) {
-      w.getValueVector().getMutator().setValueCount(count);
-    }
+    VectorAccessibleUtilities.setValueCount(outgoing, count);
   }
 
   @Override
   public void close() throws IOException {
     vector4.clear();
-    for (final VectorWrapper<?> w: outgoing) {
-      w.getValueVector().clear();
-    }
-    for (final VectorWrapper<?> w : hyperBatch) {
-      w.clear();
-    }
+    VectorAccessibleUtilities.clear(outgoing);
+    VectorAccessibleUtilities.clear(hyperBatch);
 
     for (BatchGroup batchGroup : batchGroups) {
       batchGroup.close();
@@ -123,9 +117,7 @@
   }
 
   private void allocateVectors(int targetRecordCount) {
-    for (VectorWrapper w: outgoing) {
-      AllocationHelper.allocateNew(w.getValueVector(), targetRecordCount);
-    }
+    VectorAccessibleUtilities.allocateVectors(outgoing, targetRecordCount);
   }
 
   private void siftDown() {
@@ -162,8 +154,8 @@
     return doEval(sv1, sv2);
   }
 
-  public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorAccessible incoming, @Named("outgoing") VectorAccessible outgoing);
+  public abstract void doSetup(@Named("context") FragmentContext context,
+      @Named("incoming") VectorAccessible incoming, @Named("outgoing") VectorAccessible outgoing);
   public abstract int doEval(@Named("leftIndex") int leftIndex, @Named("rightIndex") int rightIndex);
   public abstract void doCopy(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
-
 }
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 96b9cee..cb835e7 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
@@ -22,14 +22,17 @@
 import org.apache.drill.exec.physical.base.PhysicalOperator;
 
 /**
- * Implements an AbstractUnaryRecordBatch where the incoming record batch is known at the time of creation
+ * Implements an AbstractUnaryRecordBatch where the incoming record batch is
+ * known at the time of creation
+ *
  * @param <T>
  */
 public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> extends AbstractUnaryRecordBatch<T> {
 
   protected final RecordBatch incoming;
 
-  public AbstractSingleRecordBatch(T popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
+  public AbstractSingleRecordBatch(T popConfig, FragmentContext context,
+      RecordBatch incoming) throws OutOfMemoryException {
     super(popConfig, context);
     this.incoming = incoming;
   }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractUnaryRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractUnaryRecordBatch.java
index a7cb01f..808f6ed 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractUnaryRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractUnaryRecordBatch.java
@@ -23,6 +23,8 @@
 import org.apache.drill.exec.physical.base.PhysicalOperator;
 import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
 import org.apache.drill.exec.vector.SchemaChangeCallBack;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * The base class for operators that have a single input. The concrete implementations provide the
@@ -34,9 +36,9 @@
  * @param <T>
  */
 public abstract class AbstractUnaryRecordBatch<T extends PhysicalOperator> extends AbstractRecordBatch<T> {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(new Object() {}.getClass().getEnclosingClass());
+  private static final Logger logger = LoggerFactory.getLogger(new Object() {}.getClass().getEnclosingClass());
 
-  protected boolean outOfMemory = false;
+  protected boolean outOfMemory;
   protected SchemaChangeCallBack callBack = new SchemaChangeCallBack();
   private IterOutcome lastKnownOutcome;
 
@@ -140,18 +142,17 @@
   protected abstract IterOutcome doWork();
 
   /**
-   * Default behavior to handle NULL input (aka FAST NONE): incoming return NONE before return a OK_NEW_SCHEMA:
-   * This could happen when the underneath Scan operators do not produce any batch with schema.
-   *
+   * Default behavior to handle NULL input (aka FAST NONE): incoming return NONE
+   * before return a OK_NEW_SCHEMA: This could happen when the underneath Scan
+   * operators do not produce any batch with schema.
    * <p>
-   * Notice that NULL input is different from input with an empty batch. In the later case, input provides
-   * at least a batch, thought it's empty.
-   *</p>
-   *
+   * Notice that NULL input is different from input with an empty batch. In the
+   * later case, input provides at least a batch, thought it's empty.
+   * </p>
    * <p>
-   * This behavior could be override in each individual operator, if the operator's semantics is to
-   * inject a batch with schema.
-   *</p>
+   * This behavior could be override in each individual operator, if the
+   * operator's semantics is to inject a batch with schema.
+   * </p>
    *
    * @return IterOutcome.NONE.
    */
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 8361dbe..1f06710 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
@@ -89,18 +89,18 @@
     // the schema has changed since the previous call.
 
     // Set up to recognize previous fields that no longer exist.
-    final Map<String, ValueVector> oldFields = CaseInsensitiveMap.newHashMap();
-    for (final VectorWrapper<?> wrapper : container) {
-      final ValueVector vector = wrapper.getValueVector();
+    Map<String, ValueVector> oldFields = CaseInsensitiveMap.newHashMap();
+    for (VectorWrapper<?> wrapper : container) {
+      ValueVector vector = wrapper.getValueVector();
       oldFields.put(vector.getField().getName(), vector);
     }
 
-    final VectorContainer newVectors = new VectorContainer();
+    VectorContainer newVectors = new VectorContainer();
     try {
-      final List<SerializedField> fields = def.getFieldList();
+      List<SerializedField> fields = def.getFieldList();
       int bufOffset = 0;
-      for (final SerializedField field : fields) {
-        final MaterializedField fieldDef = MaterializedField.create(field);
+      for (SerializedField field : fields) {
+        MaterializedField fieldDef = MaterializedField.create(field);
         ValueVector vector = oldFields.remove(fieldDef.getName());
 
         if (vector == null) {
@@ -144,8 +144,8 @@
       }
 
       // rebuild the schema.
-      final SchemaBuilder builder = BatchSchema.newBuilder();
-      for (final VectorWrapper<?> v : newVectors) {
+      SchemaBuilder builder = BatchSchema.newBuilder();
+      for (VectorWrapper<?> v : newVectors) {
         builder.addField(v.getField());
       }
       builder.setSelectionVectorMode(BatchSchema.SelectionVectorMode.NONE);
@@ -156,14 +156,12 @@
     } 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.
-      for (final VectorWrapper<?> wrapper:newVectors) {
-        wrapper.getValueVector().clear();
-      }
+      VectorAccessibleUtilities.clear(newVectors);
       throw cause;
     } finally {
       if (! oldFields.isEmpty()) {
         schemaChanged = true;
-        for (final ValueVector vector : oldFields.values()) {
+        for (ValueVector vector : oldFields.values()) {
           vector.clear();
         }
       }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java
index 04cf32e..e03b17c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordIterator.java
@@ -36,7 +36,6 @@
  * RecordIterator will hold onto multiple record batches in order to support resetting beyond record batch boundary.
  */
 public class RecordIterator implements VectorAccessible {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RecordIterator.class);
 
   private final RecordBatch incoming;
   private final AbstractRecordBatch<?> outgoing;
@@ -48,10 +47,10 @@
   private int markedInnerPosition;
   private long markedOuterPosition;
   private IterOutcome lastOutcome;
-  private int inputIndex;           // For two way merge join 0:left, 1:right
+  private final int inputIndex;           // For two way merge join 0:left, 1:right
   private boolean lastBatchRead;    // True if all batches are consumed.
   private boolean initialized;
-  private OperatorContext oContext;
+  private final OperatorContext oContext;
   private final boolean enableMarkAndReset;
 
   private final VectorContainer container; // Holds VectorContainer of current record batch
@@ -217,6 +216,7 @@
               container.addOrGet(w.getField());
             }
             container.buildSchema(rbd.getContainer().getSchema().getSelectionVectorMode());
+            container.setEmpty();
             initialized = true;
           }
           if (innerRecordCount > 0) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java
index 6d4e057..1cfc61d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorContainer.java
@@ -32,6 +32,7 @@
 import org.apache.drill.exec.record.selection.SelectionVector4;
 import org.apache.drill.exec.vector.SchemaChangeCallBack;
 import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.shaded.guava.com.google.common.annotations.VisibleForTesting;
 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.Sets;
@@ -124,9 +125,15 @@
    * Transfer vectors from containerIn to this.
    */
   public void transferIn(VectorContainer containerIn) {
-    Preconditions.checkArgument(this.wrappers.size() == containerIn.wrappers.size());
-    for (int i = 0; i < this.wrappers.size(); ++i) {
-      containerIn.wrappers.get(i).transfer(this.wrappers.get(i));
+    rawTransferIn(containerIn);
+    setRecordCount(containerIn.getRecordCount());
+  }
+
+  @VisibleForTesting
+  public void rawTransferIn(VectorContainer containerIn) {
+    Preconditions.checkArgument(wrappers.size() == containerIn.wrappers.size());
+    for (int i = 0; i < wrappers.size(); ++i) {
+      containerIn.wrappers.get(i).transfer(wrappers.get(i));
     }
   }
 
@@ -237,14 +244,14 @@
     * @param srcIndex The index of the row to copy from the source {@link VectorContainer}.
     * @return Position one above where the row was appended
     */
-    public int appendRow(VectorContainer srcContainer, int srcIndex) {
-      for (int vectorIndex = 0; vectorIndex < wrappers.size(); vectorIndex++) {
-        ValueVector destVector = wrappers.get(vectorIndex).getValueVector();
-        ValueVector srcVector = srcContainer.wrappers.get(vectorIndex).getValueVector();
-        destVector.copyEntry(recordCount, srcVector, srcIndex);
-      }
-      return incRecordCount();
+  public int appendRow(VectorContainer srcContainer, int srcIndex) {
+    for (int vectorIndex = 0; vectorIndex < wrappers.size(); vectorIndex++) {
+      ValueVector destVector = wrappers.get(vectorIndex).getValueVector();
+      ValueVector srcVector = srcContainer.wrappers.get(vectorIndex).getValueVector();
+      destVector.copyEntry(recordCount, srcVector, srcIndex);
     }
+    return incRecordCount();
+  }
 
   public TypedFieldId add(ValueVector vv) {
     schemaChanged();
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 9e31ff8..577517d 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
@@ -31,10 +31,10 @@
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 
 /**
- * A specialized version of record batch that can moves out buffers and preps them for writing.
+ * A specialized version of record batch that can moves out buffers and preps
+ * them for writing.
  */
 public class WritableBatch implements AutoCloseable {
-  //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(WritableBatch.class);
 
   private final RecordBatchDef def;
   private final DrillBuf[] buffers;
@@ -74,7 +74,8 @@
   public void reconstructContainer(BufferAllocator allocator, VectorContainer container) {
     Preconditions.checkState(!cleared,
         "Attempted to reconstruct a container from a WritableBatch after it had been cleared");
-    if (buffers.length > 0) { /* If we have DrillBuf's associated with value vectors */
+    // If we have DrillBuf's associated with value vectors
+    if (buffers.length > 0) {
       int len = 0;
       for (DrillBuf b : buffers) {
         len += b.capacity();
@@ -82,7 +83,7 @@
 
       DrillBuf newBuf = allocator.buffer(len);
       try {
-        /* Copy data from each buffer into the compound buffer */
+        // Copy data from each buffer into the compound buffer
         int offset = 0;
         for (DrillBuf buf : buffers) {
           newBuf.setBytes(offset, buf);
@@ -94,9 +95,9 @@
 
         int bufferOffset = 0;
 
-        /*
-         * For each value vector slice up the appropriate size from the compound buffer and load it into the value vector
-         */
+        // For each value vector slice up the appropriate size from the
+        // compound buffer and load it into the value vector
+
         int vectorIndex = 0;
 
         for (VectorWrapper<?> vv : container) {
@@ -120,16 +121,11 @@
       svMode = SelectionVectorMode.NONE;
     }
     container.buildSchema(svMode);
-
-    /* Set the record count in the value vector */
-    for (VectorWrapper<?> v : container) {
-      ValueVector.Mutator m = v.getValueVector().getMutator();
-      m.setValueCount(def.getRecordCount());
-    }
+    container.setValueCount(def.getRecordCount());
   }
 
   public void clear() {
-    if(cleared) {
+    if (cleared) {
       return;
     }
     for (DrillBuf buf : buffers) {
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 69500c0..99b0723 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
@@ -43,6 +43,8 @@
 
 import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
 import org.apache.drill.shaded.guava.com.google.common.collect.Queues;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Encapsulates the future management of query submissions.  This entails a
@@ -56,8 +58,8 @@
  * handle this case and then do a switcheroo.
  */
 public class QueryResultHandler {
-  private static final org.slf4j.Logger logger =
-      org.slf4j.LoggerFactory.getLogger(QueryResultHandler.class);
+  private static final Logger logger =
+      LoggerFactory.getLogger(QueryResultHandler.class);
 
   /**
    * Current listener for results, for each active query.
@@ -74,7 +76,7 @@
   }
 
   public RpcConnectionHandler<UserToBitConnection> getWrappedConnectionHandler(
-      final RpcConnectionHandler<UserToBitConnection> handler) {
+      RpcConnectionHandler<UserToBitConnection> handler) {
     return new ChannelClosedHandler(handler);
   }
 
@@ -82,11 +84,11 @@
    * Maps internal low-level API protocol to {@link UserResultsListener}-level API protocol.
    * handles data result messages
    */
-  public void resultArrived( ByteBuf pBody ) throws RpcException {
-    final QueryResult queryResult = RpcBus.get( pBody, QueryResult.PARSER );
+  public void resultArrived(ByteBuf pBody) throws RpcException {
+    QueryResult queryResult = RpcBus.get(pBody, QueryResult.PARSER);
 
-    final QueryId queryId = queryResult.getQueryId();
-    final QueryState queryState = queryResult.getQueryState();
+    QueryId queryId = queryResult.getQueryId();
+    QueryState queryState = queryResult.getQueryState();
 
     if (logger.isDebugEnabled()) {
       logger.debug("resultArrived: queryState: {}, queryId = {}", queryState,
@@ -95,18 +97,18 @@
 
     assert queryResult.hasQueryState() : "received query result without QueryState";
 
-    final boolean isFailureResult = QueryState.FAILED == queryState;
+    boolean isFailureResult = QueryState.FAILED == queryState;
     // CANCELED queries are handled the same way as COMPLETED
-    final boolean isTerminalResult;
-    switch ( queryState ) {
+    boolean isTerminalResult;
+    switch (queryState) {
       case FAILED:
       case CANCELED:
       case COMPLETED:
         isTerminalResult = true;
         break;
       default:
-        logger.error( "Unexpected/unhandled QueryState " + queryState
-          + " (for query " + queryId +  ")" );
+        logger.error("Unexpected/unhandled QueryState " + queryState
+          + " (for query " + queryId +  ")");
         isTerminalResult = false;
         break;
     }
@@ -127,19 +129,19 @@
 
         try {
           resultsListener.queryCompleted(queryState);
-        } catch ( Exception e ) {
+        } catch (Exception e) {
           resultsListener.submissionFailed(UserException.systemError(e).build(logger));
         }
       } else {
         logger.warn("queryState {} was ignored", queryState);
       }
     } finally {
-      if ( isTerminalResult ) {
+      if (isTerminalResult) {
         // TODO:  What exactly are we checking for?  How should we really check
         // for it?
-        if ( (! ( resultsListener instanceof BufferingResultsListener )
-          || ((BufferingResultsListener) resultsListener).output != null ) ) {
-          queryIdToResultsListenersMap.remove( queryId, resultsListener );
+        if ((! (resultsListener instanceof BufferingResultsListener)
+          || ((BufferingResultsListener) resultsListener).output != null)) {
+          queryIdToResultsListenersMap.remove(queryId, resultsListener);
         }
       }
     }
@@ -149,50 +151,52 @@
    * Maps internal low-level API protocol to {@link UserResultsListener}-level API protocol.
    * handles query data messages
    */
-  public void batchArrived( ConnectionThrottle throttle,
-                            ByteBuf pBody, ByteBuf dBody ) throws RpcException {
-    final QueryData queryData = RpcBus.get( pBody, QueryData.PARSER );
+  public void batchArrived(ConnectionThrottle throttle,
+                            ByteBuf pBody, ByteBuf dBody) throws RpcException {
+    QueryData queryData = RpcBus.get(pBody, QueryData.PARSER);
     // Current batch coming in.
-    final DrillBuf drillBuf = (DrillBuf) dBody;
-    final QueryDataBatch batch = new QueryDataBatch( queryData, drillBuf );
+    DrillBuf drillBuf = (DrillBuf) dBody;
+    QueryDataBatch batch = new QueryDataBatch(queryData, drillBuf);
 
-    final QueryId queryId = queryData.getQueryId();
+    QueryId queryId = queryData.getQueryId();
 
     if (logger.isDebugEnabled()) {
       logger.debug("batchArrived: queryId = {}", QueryIdHelper.getQueryId(queryId));
     }
-    logger.trace( "batchArrived: batch = {}", batch );
+    logger.trace("batchArrived: batch = {}", batch);
 
-    final UserResultsListener resultsListener = newUserResultsListener(queryId);
+    UserResultsListener resultsListener = newUserResultsListener(queryId);
 
     // A data case--pass on via dataArrived
     try {
       resultsListener.dataArrived(batch, throttle);
       // That releases batch if successful.
-    } catch ( Exception e ) {
+    } catch (Exception e) {
       batch.release();
       resultsListener.submissionFailed(UserException.systemError(e).build(logger));
     }
   }
 
   /**
-   * Return {@link UserResultsListener} associated with queryId. Will create a new {@link BufferingResultsListener}
-   * if no listener found.
-   * @param queryId queryId we are getting the listener for
+   * Return {@link UserResultsListener} associated with queryId. Will create a
+   * new {@link BufferingResultsListener} if no listener found.
+   *
+   * @param queryId
+   *          queryId we are getting the listener for
    * @return {@link UserResultsListener} associated with queryId
    */
   private UserResultsListener newUserResultsListener(QueryId queryId) {
-    UserResultsListener resultsListener = queryIdToResultsListenersMap.get( queryId );
-    logger.trace( "For QueryId [{}], retrieved results listener {}", queryId, resultsListener );
-    if ( null == resultsListener ) {
+    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 );
+      resultsListener = queryIdToResultsListenersMap.putIfAbsent(queryId, bl);
       // If we had a successful insertion, use that reference.  Otherwise, just
       // throw away the new buffering listener.
-      if ( null == resultsListener ) {
+      if (null == resultsListener) {
         resultsListener = bl;
       }
     }
@@ -201,7 +205,7 @@
 
   private static class BufferingResultsListener implements UserResultsListener {
 
-    private ConcurrentLinkedQueue<QueryDataBatch> results = Queues.newConcurrentLinkedQueue();
+    private final ConcurrentLinkedQueue<QueryDataBatch> results = Queues.newConcurrentLinkedQueue();
     private volatile UserException ex;
     private volatile QueryState queryState;
     private volatile UserResultsListener output;
@@ -252,8 +256,10 @@
     @Override
     public void submissionFailed(UserException ex) {
       assert queryState == null;
-      // there is one case when submissionFailed() is called even though the query didn't fail on the server side
-      // it happens when UserResultsListener.batchArrived() throws an exception that will be passed to
+      // there is one case when submissionFailed() is called even though the
+      // query didn't fail on the server side
+      // it happens when UserResultsListener.batchArrived() throws an exception
+      // that will be passed to
       // submissionFailed() by QueryResultHandler.dataArrived()
       queryState = QueryState.FAILED;
       synchronized (this) {
@@ -335,7 +341,7 @@
     }
 
     @Override
-    public void interrupted(final InterruptedException ex) {
+    public void interrupted(InterruptedException ex) {
       if (!isTerminal.compareAndSet(false, true)) {
         logger.warn("Received multiple responses to run query request.");
         return;
@@ -350,26 +356,27 @@
   }
 
   /**
-   * When a {@link UserToBitConnection connection} to a server is successfully created, this handler adds a
-   * listener to that connection that listens to connection closure. If the connection is closed, all active
+   * When a {@link UserToBitConnection connection} to a server is successfully
+   * created, this handler adds a listener to that connection that listens to
+   * connection closure. If the connection is closed, all active
    * {@link UserResultsListener result listeners} are failed.
    */
   private class ChannelClosedHandler implements RpcConnectionHandler<UserToBitConnection> {
 
     private final RpcConnectionHandler<UserToBitConnection> parentHandler;
 
-    public ChannelClosedHandler(final RpcConnectionHandler<UserToBitConnection> parentHandler) {
+    public ChannelClosedHandler(RpcConnectionHandler<UserToBitConnection> parentHandler) {
       this.parentHandler = parentHandler;
     }
 
     @Override
-    public void connectionSucceeded(final UserToBitConnection connection) {
+    public void connectionSucceeded(UserToBitConnection connection) {
       connection.getChannel().closeFuture().addListener(
           new GenericFutureListener<Future<? super Void>>() {
             @Override
             public void operationComplete(Future<? super Void> future)
                 throws Exception {
-              for (final UserResultsListener listener : queryIdToResultsListenersMap.values()) {
+              for (UserResultsListener listener : queryIdToResultsListenersMap.values()) {
                 listener.submissionFailed(UserException.connectionError()
                     .message("Connection %s closed unexpectedly. Drillbit down?",
                         connection.getName())
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestAggNullable.java b/exec/java-exec/src/test/java/org/apache/drill/TestAggNullable.java
index 865aae4..223fecf 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestAggNullable.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestAggNullable.java
@@ -26,7 +26,6 @@
 
 @Category(OperatorTest.class)
 public class TestAggNullable extends BaseTestQuery {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestAggNullable.class);
 
   private static void enableAggr(boolean ha, boolean sa) throws Exception {
 
@@ -72,5 +71,4 @@
     assertEquals(String.format("Received unexpected number of rows in output: expected=%d, received=%s",
         expectedRecordCount, actualRecordCount), expectedRecordCount, actualRecordCount);
   }
-
 }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java b/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
index 3a02544..ddb1e5b 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
@@ -37,6 +37,7 @@
 
 @Category({SqlFunctionTest.class, OperatorTest.class, PlannerTest.class, UnlikelyTest.class})
 public class TestExampleQueries extends BaseTestQuery {
+
   @BeforeClass
   public static void setupTestFiles() {
     dirTestWatcher.copyResourceToRoot(Paths.get("tpchmulti"));
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestStarQueries.java b/exec/java-exec/src/test/java/org/apache/drill/TestStarQueries.java
index d3cf0d8..f79eb60 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestStarQueries.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestStarQueries.java
@@ -36,7 +36,6 @@
 
 @Category({SqlTest.class, PlannerTest.class})
 public class TestStarQueries extends BaseTestQuery {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestStarQueries.class);
 
   @BeforeClass
   public static void setupTestFiles() {
@@ -298,7 +297,7 @@
     try {
       test("select x.n_nationkey, x.n_name, x.n_regionkey, x.r_name from (select * from cp.`tpch/nation.parquet` n, cp.`tpch/region.parquet` r where n.n_regionkey = r.r_regionkey) x " );
     } catch (UserException e) {
-      logger.info("***** Test resulted in expected failure: " + e.getMessage());
+      // Expected
       throw e;
     }
   }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestUnionAll.java b/exec/java-exec/src/test/java/org/apache/drill/TestUnionAll.java
index 3523bdf..292602f 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestUnionAll.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestUnionAll.java
@@ -347,7 +347,6 @@
         .baselineColumns("n_nationkey", "n_regionkey", "n_name")
         .build().run();
 
-
     testBuilder()
         .sqlQuery(query2)
         .unOrdered()
@@ -355,7 +354,7 @@
         .baselineTypes(TypeProtos.MinorType.INT, TypeProtos.MinorType.INT, TypeProtos.MinorType.VARCHAR)
         .baselineColumns("n_nationkey", "n_regionkey", "n_name")
         .build().run();
-    }
+  }
 
   @Test // see DRILL-1977, DRILL-2376, DRILL-2377, DRILL-2378, DRILL-2379
   @Category(UnlikelyTest.class)
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/DrillSeparatePlanningTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/DrillSeparatePlanningTest.java
index 0ffd04c..a4d10b6 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/DrillSeparatePlanningTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/DrillSeparatePlanningTest.java
@@ -51,6 +51,7 @@
  */
 @Category({SlowTest.class, PlannerTest.class})
 public class DrillSeparatePlanningTest extends ClusterTest {
+
   @BeforeClass
   public static void setupFiles() {
     dirTestWatcher.copyResourceToRoot(Paths.get("multilevel", "json"));
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/nested/TestFastComplexSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/nested/TestFastComplexSchema.java
index de8a77e..fd43fa1 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/nested/TestFastComplexSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/nested/TestFastComplexSchema.java
@@ -39,54 +39,67 @@
   public void test2() throws Exception {
     test("alter session set `planner.enable_hashjoin` = false");
     test("alter session set `planner.slice_target` = 1");
-    test("SELECT r.r_name, \n" +
-            "       t1.f \n" +
-            "FROM   cp.`tpch/region.parquet` r \n" +
-            "       JOIN (SELECT Flatten(x) AS f \n" +
-            "             FROM   (SELECT Convert_from('[0, 1]', 'json') AS x \n" +
-            "                     FROM   cp.`tpch/region.parquet`)) t1 \n" +
-            "         ON t1.f = cast(r.r_regionkey as bigint) \n" +
-            "ORDER  BY r.r_name");
-    test("alter session set `planner.enable_hashjoin` = true");
-    test("alter session set `planner.slice_target` = 1000000");
+    try {
+      test("SELECT r.r_name, \n" +
+              "       t1.f \n" +
+              "FROM   cp.`tpch/region.parquet` r \n" +
+              "       JOIN (SELECT Flatten(x) AS f \n" +
+              "             FROM   (SELECT Convert_from('[0, 1]', 'json') AS x \n" +
+              "                     FROM   cp.`tpch/region.parquet`)) t1 \n" +
+              "         ON t1.f = cast(r.r_regionkey as bigint) \n" +
+              "ORDER  BY r.r_name");
+    } finally {
+      test("alter session reset `planner.enable_hashjoin`");
+      test("alter session reset `planner.slice_target`");
+    }
   }
 
   @Test
   public void test3() throws Exception {
     test("alter session set `planner.enable_hashjoin` = false");
     test("alter session set `planner.slice_target` = 1");
-    test("select f from\n" +
-            "(select convert_from(nation, 'json') as f from\n" +
-            "(select concat('{\"name\": \"', n.n_name, '\", ', '\"regionkey\": ', r.r_regionkey, '}') as nation\n" +
-            "       from cp.`tpch/nation.parquet` n,\n" +
-            "            cp.`tpch/region.parquet` r\n" +
-            "        where \n" +
-            "        n.n_regionkey = r.r_regionkey)) t\n" +
-            "order by t.f.name");
+    try {
+      test("select f from\n" +
+              "(select convert_from(nation, 'json') as f from\n" +
+              "(select concat('{\"name\": \"', n.n_name, '\", ', '\"regionkey\": ', r.r_regionkey, '}') as nation\n" +
+              "       from cp.`tpch/nation.parquet` n,\n" +
+              "            cp.`tpch/region.parquet` r\n" +
+              "        where \n" +
+              "        n.n_regionkey = r.r_regionkey)) t\n" +
+              "order by t.f.name");
+    } finally {
+      test("alter session reset `planner.enable_hashjoin`");
+      test("alter session reset `planner.slice_target`");
+    }
   }
 
   @Test
   public void test4() throws Exception {
     test("alter session set `planner.enable_hashjoin` = false");
     test("alter session set `planner.slice_target` = 1");
-    test("SELECT f \n" +
-            "FROM   (SELECT Convert_from(nation, 'json') AS f \n" +
-            "        FROM   (SELECT Concat('{\"name\": \"', n.n_name, '\", ', '\"regionkey\": ', \n" +
-            "                       r.r_regionkey, \n" +
-            "                               '}') AS \n" +
-            "                       nation \n" +
-            "                FROM   cp.`tpch/nation.parquet` n, \n" +
-            "                       cp.`tpch/region.parquet` r \n" +
-            "                WHERE  n.n_regionkey = r.r_regionkey \n" +
-            "                       AND r.r_regionkey = 4)) t \n" +
-            "ORDER  BY t.f.name");
+    try {
+      test("SELECT f \n" +
+              "FROM   (SELECT Convert_from(nation, 'json') AS f \n" +
+              "        FROM   (SELECT Concat('{\"name\": \"', n.n_name, '\", ', '\"regionkey\": ', \n" +
+              "                       r.r_regionkey, \n" +
+              "                               '}') AS \n" +
+              "                       nation \n" +
+              "                FROM   cp.`tpch/nation.parquet` n, \n" +
+              "                       cp.`tpch/region.parquet` r \n" +
+              "                WHERE  n.n_regionkey = r.r_regionkey \n" +
+              "                       AND r.r_regionkey = 4)) t \n" +
+              "ORDER  BY t.f.name");
+    } finally {
+      test("alter session reset `planner.enable_hashjoin`");
+      test("alter session reset `planner.slice_target`");
+    }
   }
 
   @Test //DRILL-4783 when resultset is empty, don't throw exception.
   @Category(UnlikelyTest.class)
   public void test5() throws Exception {
 
-    //when there is no incoming record, flatten won't throw exception
+    // when there is no incoming record, flatten won't throw exception
     testBuilder().sqlQuery("select flatten(j) from \n" +
            " (select convert_from(names, 'json') j \n" +
            " from (select concat('[\"', first_name, '\", ', '\"', last_name, '\"]') names \n" +
@@ -94,7 +107,7 @@
         .expectsEmptyResultSet()
         .build().run();
 
-    //result is not empty and is list type,
+    // result is not empty and is list type,
     testBuilder().sqlQuery("select flatten(j) n from \n" +
         " (select convert_from(names, 'json') j \n" +
         " from (select concat('[\"', first_name, '\", ', '\"', last_name, '\"]') names \n" +
@@ -105,7 +118,7 @@
         .baselineValues("Nowmer")
         .build().run();
 
-    //result is not empty, and flatten got incompatible (non-list) incoming records. got exception thrown
+    // result is not empty, and flatten got incompatible (non-list) incoming records. got exception thrown
     errorMsgTestHelper("select flatten(first_name) from \n" +
         "(select first_name from cp.`employee.json` where first_name='Sheri')",
         "Flatten does not support inputs of non-list values");
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
index 51e47ba..2d1a1ba 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
@@ -77,9 +77,6 @@
     this.allocator = context.getAllocator();
     this.container = new VectorContainer(allocator, schema);
     this.allOutcomes = iterOutcomes;
-    this.currentContainerIndex = 0;
-    this.currentOutcomeIndex = 0;
-    this.isDone = false;
   }
 
   @Deprecated
@@ -193,7 +190,7 @@
   @Override
   public IterOutcome next() {
 
-    if(isDone) {
+    if (isDone) {
       return IterOutcome.NONE;
     }
 
@@ -213,16 +210,18 @@
       switch (rowSet.indirectionType()) {
         case NONE:
         case TWO_BYTE:
-          container.transferIn(input);
-          if ( input.hasRecordCount() ) { // in case special test of uninitialized input container
-            container.setRecordCount(input.getRecordCount());
+          if (input.hasRecordCount()) { // in case special test of uninitialized input container
+            container.transferIn(input);
+          } else {
+            // Not normally a valid condition, supported here just for testing
+            container.rawTransferIn(input);
           }
           SelectionVector2 inputSv2 = ((RowSet.SingleRowSet) rowSet).getSv2();
 
           if (sv2 != null) {
             // Operators assume that new values for an Sv2 are transferred in.
             sv2.allocateNewSafe(inputSv2.getCount());
-            for (int i=0; i<inputSv2.getCount(); ++i) {
+            for (int i=0; i < inputSv2.getCount(); ++i) {
               sv2.setIndex(i, inputSv2.getIndex(i));
             }
             sv2.setRecordCount(inputSv2.getCount());
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggrSpill.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggrSpill.java
index 376ac03..87a324f 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggrSpill.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggrSpill.java
@@ -57,8 +57,6 @@
 
   /**
    *  A template for Hash Aggr spilling tests
-   *
-   * @throws Exception
    */
   private void testSpill(long maxMem, long numPartitions, long minBatches, int maxParallel, boolean fallback, boolean predict,
                          String sql, long expectedRows, int cycle, int fromPart, int toPart) throws Exception {
@@ -84,8 +82,6 @@
   /**
    * Test "normal" spilling: Only 2 (or 3) partitions (out of 4) would require spilling
    * ("normal spill" means spill-cycle = 1 )
-   *
-   * @throws Exception
    */
   @Test
   public void testSimpleHashAggrSpill() throws Exception {
@@ -96,8 +92,6 @@
   /**
    * Test with "needed memory" prediction turned off
    * (i.e., exercise code paths that catch OOMs from the Hash Table and recover)
-   *
-   * @throws Exception
    */
   @Test
   @Ignore("DRILL-7301")
@@ -126,9 +120,8 @@
   }
 
   /**
-   * Test Secondary and Tertiary spill cycles - Happens when some of the spilled partitions cause more spilling as they are read back
-   *
-   * @throws Exception
+   * Test Secondary and Tertiary spill cycles - Happens when some of the spilled
+   * partitions cause more spilling as they are read back
    */
   @Test
   public void testHashAggrSecondaryTertiarySpill() throws Exception {
@@ -139,9 +132,8 @@
   }
 
   /**
-   * Test with the "fallback" option disabled: When not enough memory available to allow spilling, then fail (Resource error) !!
-   *
-   * @throws Exception
+   * Test with the "fallback" option disabled: When not enough memory available
+   * to allow spilling, then fail (Resource error) !!
    */
   @Test
   public void testHashAggrFailWithFallbackDisabed() throws Exception {
@@ -158,10 +150,10 @@
   }
 
   /**
-   * Test with the "fallback" option ON: When not enough memory is available to allow spilling (internally need enough memory to
-   * create multiple partitions), then behave like the pre-1.11 Hash Aggregate: Allocate unlimited memory, no spill.
-   *
-   * @throws Exception
+   * Test with the "fallback" option ON: When not enough memory is available to
+   * allow spilling (internally need enough memory to create multiple
+   * partitions), then behave like the pre-1.11 Hash Aggregate: Allocate
+   * unlimited memory, no spill.
    */
   @Test
   public void testHashAggrSuccessWithFallbackEnabled() throws Exception {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/flatten/TestFlatten.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/flatten/TestFlatten.java
index 7610e71..d2485f1 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/flatten/TestFlatten.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/flatten/TestFlatten.java
@@ -276,7 +276,7 @@
   @Test
   @Category(UnlikelyTest.class)
   public void drill1652() throws Exception {
-    if(RUN_ADVANCED_TESTS){
+    if (RUN_ADVANCED_TESTS) {
       test("select uid, flatten(transactions) from dfs.`tmp/bigfile.json`");
     }
   }
@@ -330,7 +330,7 @@
   public void testKVGenFlatten2() throws Exception {
     // currently runs
     // TODO - re-verify results by hand
-    if(RUN_ADVANCED_TESTS){
+    if (RUN_ADVANCED_TESTS) {
       test("select flatten(kvgen(visited_cellid_counts)) as mytb from dfs.`tmp/mapkv.json`");
     }
   }
@@ -351,7 +351,7 @@
 
     // FIXED BY RETURNING PROPER SCHEMA DURING FAST SCHEMA STEP
     // these types of problems are being solved more generally as we develp better support for chaning schema
-    if(RUN_ADVANCED_TESTS){
+    if (RUN_ADVANCED_TESTS) {
       test("select celltbl.catl from (\n" +
           "        select flatten(categories) catl from dfs.`tmp/yelp_academic_dataset_business.json` b limit 100\n" +
           "    )  celltbl where celltbl.catl = 'Doctors'");
@@ -360,7 +360,7 @@
 
   @Test
   public void countAggFlattened() throws Exception {
-    if(RUN_ADVANCED_TESTS){
+    if (RUN_ADVANCED_TESTS) {
       test("select celltbl.catl, count(celltbl.catl) from ( " +
           "select business_id, flatten(categories) catl from dfs.`tmp/yelp_academic_dataset_business.json` b limit 100 " +
           ")  celltbl group by celltbl.catl limit 10 ");
@@ -369,35 +369,33 @@
 
   @Test
   public void flattenAndAdditionalColumn() throws Exception {
-    if(RUN_ADVANCED_TESTS){
+    if (RUN_ADVANCED_TESTS) {
       test("select business_id, flatten(categories) from dfs.`tmp/yelp_academic_dataset_business.json` b");
     }
   }
 
   @Test
   public void testFailingFlattenAlone() throws Exception {
-    if(RUN_ADVANCED_TESTS){
+    if (RUN_ADVANCED_TESTS) {
       test("select flatten(categories) from dfs.`tmp/yelp_academic_dataset_business.json` b  ");
     }
   }
 
   @Test
   public void testDistinctAggrFlattened() throws Exception {
-    if(RUN_ADVANCED_TESTS){
+    if (RUN_ADVANCED_TESTS) {
       test(" select distinct(celltbl.catl) from (\n" +
           "        select flatten(categories) catl from dfs.`tmp/yelp_academic_dataset_business.json` b\n" +
           "    )  celltbl");
     }
-
   }
 
   @Test
   @Category(UnlikelyTest.class)
   public void testDrill1665() throws Exception {
-    if(RUN_ADVANCED_TESTS){
+    if (RUN_ADVANCED_TESTS) {
       test("select id, flatten(evnts) as rpt from dfs.`tmp/drill1665.json`");
     }
-
   }
 
   @Test
@@ -479,7 +477,6 @@
         .unOrdered()
         .jsonBaselineFile("flatten/drill-2106-result.json")
         .go();
-
   }
 
   @Test // see DRILL-2146
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/window/TestWindowFrame.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/window/TestWindowFrame.java
index a015ff7..5530176 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/window/TestWindowFrame.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/window/TestWindowFrame.java
@@ -36,6 +36,7 @@
 import org.junit.experimental.categories.Category;
 
 public class TestWindowFrame extends BaseTestQuery {
+
   @BeforeClass
   public static void setupMSortBatchSize() throws IOException {
     // make sure memory sorter outputs 20 rows per batch
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/sql/TestAnalyze.java b/exec/java-exec/src/test/java/org/apache/drill/exec/sql/TestAnalyze.java
index f828cf7..01110c4 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/sql/TestAnalyze.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/sql/TestAnalyze.java
@@ -21,7 +21,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.drill.PlanTestBase;
-import org.apache.drill.exec.ExecConstants;
 import org.apache.drill.exec.proto.UserBitShared.QueryType;
 import org.apache.drill.exec.record.RecordBatchLoader;
 import org.apache.drill.exec.record.VectorWrapper;
@@ -69,7 +68,8 @@
           .baselineValues("`sales_district_id`", 110.0, 110.0, 23L, 8.0)
           .go();
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
     }
   }
 
@@ -96,7 +96,8 @@
           .baselineValues("`birth_date`", 1155.0, 1155.0, 52L, 10.0)
           .go();
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
     }
   }
 
@@ -124,8 +125,9 @@
               .baselineValues("`birth_date`", 1138.0, 1138.0, 38L, 10.001597699312988)
               .go();
     } finally {
-      test("ALTER SESSION SET `exec.statistics.deterministic_sampling` = false");
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("exec.statistics.deterministic_sampling");
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
     }
   }
 
@@ -143,7 +145,9 @@
       test("ALTER SESSION SET `planner.statistics.use` = true");
       test("SELECT * FROM dfs.tmp.`lineitem` l JOIN dfs.tmp.`orders` o ON l.l_orderkey = o.o_orderkey");
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
+      resetSessionOption("planner.statistics.use");
     }
   }
 
@@ -166,7 +170,8 @@
       verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.employee_basic4 COMPUTE STATISTICS",
           "16");
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
     }
   }
 
@@ -204,7 +209,8 @@
           .baselineValues("`dir1`", 120.0, 120.0, 4L, 2.0)
           .go();
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
     }
   }
 
@@ -214,21 +220,26 @@
     final String tmpLocation = "/multilevel/parquet";
     test("ALTER SESSION SET `planner.slice_target` = 1");
     test("ALTER SESSION SET `store.format` = 'parquet'");
-    test("CREATE TABLE dfs.tmp.parquetStale AS SELECT o_orderkey, o_custkey, o_orderstatus, " +
-         "o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment from dfs.`%s`", tmpLocation);
-    verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
-    verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS",
-        "Table parquetStale has not changed since last ANALYZE!");
-    // Verify we recompute statistics once a new file/directory is added. Update the directory some
-    // time after ANALYZE so that the timestamps are different.
-    Thread.sleep(1000);
-    final String Q4 = "/multilevel/parquet/1996/Q4";
-    test("CREATE TABLE dfs.tmp.`parquetStale/1996/Q5` AS SELECT o_orderkey, o_custkey, o_orderstatus, " +
-         "o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment from dfs.`%s`", Q4);
-    verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
-    Thread.sleep(1000);
-    test("DROP TABLE dfs.tmp.`parquetStale/1996/Q5`");
-    verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
+    try {
+      test("CREATE TABLE dfs.tmp.parquetStale AS SELECT o_orderkey, o_custkey, o_orderstatus, " +
+           "o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment from dfs.`%s`", tmpLocation);
+      verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
+      verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS",
+          "Table parquetStale has not changed since last ANALYZE!");
+      // Verify we recompute statistics once a new file/directory is added. Update the directory some
+      // time after ANALYZE so that the timestamps are different.
+      Thread.sleep(1000);
+      final String Q4 = "/multilevel/parquet/1996/Q4";
+      test("CREATE TABLE dfs.tmp.`parquetStale/1996/Q5` AS SELECT o_orderkey, o_custkey, o_orderstatus, " +
+           "o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment from dfs.`%s`", Q4);
+      verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
+      Thread.sleep(1000);
+      test("DROP TABLE dfs.tmp.`parquetStale/1996/Q5`");
+      verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
+    } finally {
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
+    }
   }
 
   @Test
@@ -236,99 +247,104 @@
     //Test ndv/rowcount for scan
     test("ALTER SESSION SET `planner.slice_target` = 1");
     test("ALTER SESSION SET `store.format` = 'parquet'");
-    test("CREATE TABLE dfs.tmp.employeeUseStat AS SELECT * from cp.`employee.json`");
-    test("CREATE TABLE dfs.tmp.departmentUseStat AS SELECT * from cp.`department.json`");
-    test("ANALYZE TABLE dfs.tmp.employeeUseStat COMPUTE STATISTICS");
-    test("ANALYZE TABLE dfs.tmp.departmentUseStat COMPUTE STATISTICS");
-    test("ALTER SESSION SET `planner.statistics.use` = true");
-    String query = " select employee_id from dfs.tmp.employeeUseStat where department_id = 2";
-    String[] expectedPlan1 = {"Filter\\(condition.*\\).*rowcount = 96.25,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan1, new String[]{});
+    try {
+      test("CREATE TABLE dfs.tmp.employeeUseStat AS SELECT * from cp.`employee.json`");
+      test("CREATE TABLE dfs.tmp.departmentUseStat AS SELECT * from cp.`department.json`");
+      test("ANALYZE TABLE dfs.tmp.employeeUseStat COMPUTE STATISTICS");
+      test("ANALYZE TABLE dfs.tmp.departmentUseStat COMPUTE STATISTICS");
+      test("ALTER SESSION SET `planner.statistics.use` = true");
+      String query = " select employee_id from dfs.tmp.employeeUseStat where department_id = 2";
+      String[] expectedPlan1 = {"Filter\\(condition.*\\).*rowcount = 96.25,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan1, new String[]{});
 
-    query = " select employee_id from dfs.tmp.employeeUseStat where department_id IN (2, 5)";
-    String[] expectedPlan2 = {"Filter\\(condition.*\\).*rowcount = 192.5,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan2, new String[]{});
+      query = " select employee_id from dfs.tmp.employeeUseStat where department_id IN (2, 5)";
+      String[] expectedPlan2 = {"Filter\\(condition.*\\).*rowcount = 192.5,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan2, new String[]{});
 
-    query = "select employee_id from dfs.tmp.employeeUseStat where department_id IN (2, 5) and employee_id = 5";
-    String[] expectedPlan3 = {"Filter\\(condition.*\\).*rowcount = 1.0,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan3, new String[]{});
+      query = "select employee_id from dfs.tmp.employeeUseStat where department_id IN (2, 5) and employee_id = 5";
+      String[] expectedPlan3 = {"Filter\\(condition.*\\).*rowcount = 1.0,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan3, new String[]{});
 
-    query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
-        + " on emp.department_id = dept.department_id";
-    String[] expectedPlan4 = {"HashJoin\\(condition.*\\).*rowcount = 1155.0,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
-            "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan4, new String[]{});
+      query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
+          + " on emp.department_id = dept.department_id";
+      String[] expectedPlan4 = {"HashJoin\\(condition.*\\).*rowcount = 1155.0,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
+              "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan4, new String[]{});
 
-    query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
-            + " on emp.department_id = dept.department_id where dept.department_id = 5";
-    String[] expectedPlan5 = {"HashJoin\\(condition.*\\).*rowcount = 96.25,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
-            "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan5, new String[]{});
+      query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
+              + " on emp.department_id = dept.department_id where dept.department_id = 5";
+      String[] expectedPlan5 = {"HashJoin\\(condition.*\\).*rowcount = 96.25,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
+              "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan5, new String[]{});
 
-    query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
-            + " on emp.department_id = dept.department_id"
-            + " where dept.department_id = 5 and emp.employee_id = 10";
-    String[] expectedPlan6 = {"MergeJoin\\(condition.*\\).*rowcount = 1.0,.*",
-            "Filter\\(condition=\\[AND\\(=\\(\\$1, 10\\), =\\(\\$0, 5\\)\\)\\]\\).*rowcount = 1.0,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
-            "Filter\\(condition=\\[=\\(\\$0, 5\\)\\]\\).*rowcount = 1.0,.*",
-            "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan6, new String[]{});
+      query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
+              + " on emp.department_id = dept.department_id"
+              + " where dept.department_id = 5 and emp.employee_id = 10";
+      String[] expectedPlan6 = {"MergeJoin\\(condition.*\\).*rowcount = 1.0,.*",
+              "Filter\\(condition=\\[AND\\(=\\(\\$1, 10\\), =\\(\\$0, 5\\)\\)\\]\\).*rowcount = 1.0,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
+              "Filter\\(condition=\\[=\\(\\$0, 5\\)\\]\\).*rowcount = 1.0,.*",
+              "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan6, new String[]{});
 
-    query = " select emp.employee_id, count(*)"
-            + " from dfs.tmp.employeeUseStat emp"
-            + " group by emp.employee_id";
-    String[] expectedPlan7 = {"HashAgg\\(group=\\[\\{0\\}\\], EXPR\\$1=\\[COUNT\\(\\)\\]\\).*rowcount = 1155.0,.*",
-            "Scan.*columns=\\[`employee_id`\\].*rowcount = 1155.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan7, new String[]{});
+      query = " select emp.employee_id, count(*)"
+              + " from dfs.tmp.employeeUseStat emp"
+              + " group by emp.employee_id";
+      String[] expectedPlan7 = {"HashAgg\\(group=\\[\\{0\\}\\], EXPR\\$1=\\[COUNT\\(\\)\\]\\).*rowcount = 1155.0,.*",
+              "Scan.*columns=\\[`employee_id`\\].*rowcount = 1155.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan7, new String[]{});
 
-    query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
-            + " on emp.department_id = dept.department_id "
-            + " group by emp.employee_id";
-    String[] expectedPlan8 = {"HashAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 730.0992454469841,.*",
-            "HashJoin\\(condition.*\\).*rowcount = 1155.0,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
-            "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan8, new String[]{});
+      query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
+              + " on emp.department_id = dept.department_id "
+              + " group by emp.employee_id";
+      String[] expectedPlan8 = {"HashAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 730.0992454469841,.*",
+              "HashJoin\\(condition.*\\).*rowcount = 1155.0,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
+              "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan8, new String[]{});
 
-    query = "select emp.employee_id, dept.department_description"
-            + " from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
-            + " on emp.department_id = dept.department_id "
-            + " group by emp.employee_id, emp.store_id, dept.department_description "
-            + " having dept.department_description = 'FINANCE'";
-    String[] expectedPlan9 = {"HashAgg\\(group=\\[\\{0, 1, 2\\}\\]\\).*rowcount = 60.84160378724867.*",
-            "HashJoin\\(condition.*\\).*rowcount = 96.25,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`, `store_id`\\].*rowcount = 1155.0.*",
-            "Filter\\(condition=\\[=\\(\\$1, 'FINANCE'\\)\\]\\).*rowcount = 1.0,.*",
-            "Scan.*columns=\\[`department_id`, `department_description`\\].*rowcount = 12.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan9, new String[]{});
+      query = "select emp.employee_id, dept.department_description"
+              + " from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
+              + " on emp.department_id = dept.department_id "
+              + " group by emp.employee_id, emp.store_id, dept.department_description "
+              + " having dept.department_description = 'FINANCE'";
+      String[] expectedPlan9 = {"HashAgg\\(group=\\[\\{0, 1, 2\\}\\]\\).*rowcount = 60.84160378724867.*",
+              "HashJoin\\(condition.*\\).*rowcount = 96.25,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`, `store_id`\\].*rowcount = 1155.0.*",
+              "Filter\\(condition=\\[=\\(\\$1, 'FINANCE'\\)\\]\\).*rowcount = 1.0,.*",
+              "Scan.*columns=\\[`department_id`, `department_description`\\].*rowcount = 12.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan9, new String[]{});
 
-    query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept\n"
-            + " on emp.department_id = dept.department_id "
-            + " group by emp.employee_id, emp.store_id "
-            + " having emp.store_id = 7";
-    String[] expectedPlan10 = {"HashAgg\\(group=\\[\\{0, 1\\}\\]\\).*rowcount = 29.203969817879365.*",
-            "HashJoin\\(condition.*\\).*rowcount = 46.2,.*",
-            "Filter\\(condition=\\[=\\(\\$2, 7\\)\\]\\).*rowcount = 46.2,.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`, `store_id`\\].*rowcount = 1155.0.*",
-            "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan10, new String[]{});
+      query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept\n"
+              + " on emp.department_id = dept.department_id "
+              + " group by emp.employee_id, emp.store_id "
+              + " having emp.store_id = 7";
+      String[] expectedPlan10 = {"HashAgg\\(group=\\[\\{0, 1\\}\\]\\).*rowcount = 29.203969817879365.*",
+              "HashJoin\\(condition.*\\).*rowcount = 46.2,.*",
+              "Filter\\(condition=\\[=\\(\\$2, 7\\)\\]\\).*rowcount = 46.2,.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`, `store_id`\\].*rowcount = 1155.0.*",
+              "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan10, new String[]{});
 
-    query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept\n"
-            + " on emp.department_id = dept.department_id "
-            + " group by emp.employee_id "
-            + " having emp.employee_id = 7";
-    String[] expectedPlan11 = {"StreamAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 1.0.*",
-            "HashJoin\\(condition.*\\).*rowcount = 1.0,.*",
-            "Filter\\(condition=\\[=\\(\\$1, 7\\)\\]\\).*rowcount = 1.0.*",
-            "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*",
-            "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan11, new String[]{});
+      query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept\n"
+              + " on emp.department_id = dept.department_id "
+              + " group by emp.employee_id "
+              + " having emp.employee_id = 7";
+      String[] expectedPlan11 = {"StreamAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 1.0.*",
+              "HashJoin\\(condition.*\\).*rowcount = 1.0,.*",
+              "Filter\\(condition=\\[=\\(\\$1, 7\\)\\]\\).*rowcount = 1.0.*",
+              "Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*",
+              "Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan11, new String[]{});
+    } finally {
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
+    }
   }
 
   @Test
@@ -337,35 +353,41 @@
     test("ALTER SESSION SET `store.format` = 'parquet'");
     test("ALTER SESSION SET `planner.statistics.use` = true");
     final String tmpLocation = "/multilevel/parquet";
-    // copy the data into the temporary location
-    test("DROP TABLE dfs.tmp.parquetStale");
-    test("CREATE TABLE dfs.tmp.parquetStale AS SELECT o_orderkey, o_custkey, o_orderstatus, " +
-            "o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment from dfs.`%s`", tmpLocation);
-    String query = "select count(distinct o_orderkey) from dfs.tmp.parquetStale";
-    verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
-    test("REFRESH TABLE METADATA dfs.tmp.parquetStale");
-    // Verify we recompute statistics once a new file/directory is added. Update the directory some
-    // time after ANALYZE so that the timestamps are different.
-    Thread.sleep(1000);
-    final String Q4 = "/multilevel/parquet/1996/Q4";
-    test("CREATE TABLE dfs.tmp.`parquetStale/1996/Q5` AS SELECT o_orderkey, o_custkey, o_orderstatus, " +
-            "o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment from dfs.`%s`", Q4);
-    // query should use STALE statistics
-    String[] expectedStalePlan = {"StreamAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 119.0.*",
-        "Scan.*rowcount = 130.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedStalePlan, new String[]{});
-    // Query should use Parquet Metadata, since statistics not available. In this case, NDV is computed as
-    // 1/10*rowcount (Calcite default). Hence, NDV is 13.0 instead of the correct 119.0
-    test("DROP TABLE dfs.tmp.`parquetStale/.stats.drill`");
-    String[] expectedPlan1 = {"HashAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 13.0.*",
-        "Scan.*rowcount = 130.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan1, new String[]{});
-    // query should use the new statistics. NDV remains unaffected since we copy the Q4 into Q5
-    verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
-    String[] expectedPlan2 = {"StreamAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 119.0.*",
-        "Scan.*rowcount = 130.0.*"};
-    PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan2, new String[]{});
-    test("DROP TABLE dfs.tmp.`parquetStale/1996/Q5`");
+    try {
+      // copy the data into the temporary location
+      test("DROP TABLE dfs.tmp.parquetStale");
+      test("CREATE TABLE dfs.tmp.parquetStale AS SELECT o_orderkey, o_custkey, o_orderstatus, " +
+              "o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment from dfs.`%s`", tmpLocation);
+      String query = "select count(distinct o_orderkey) from dfs.tmp.parquetStale";
+      verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
+      test("REFRESH TABLE METADATA dfs.tmp.parquetStale");
+      // Verify we recompute statistics once a new file/directory is added. Update the directory some
+      // time after ANALYZE so that the timestamps are different.
+      Thread.sleep(1000);
+      final String Q4 = "/multilevel/parquet/1996/Q4";
+      test("CREATE TABLE dfs.tmp.`parquetStale/1996/Q5` AS SELECT o_orderkey, o_custkey, o_orderstatus, " +
+              "o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment from dfs.`%s`", Q4);
+      // query should use STALE statistics
+      String[] expectedStalePlan = {"StreamAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 119.0.*",
+          "Scan.*rowcount = 130.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedStalePlan, new String[]{});
+      // Query should use Parquet Metadata, since statistics not available. In this case, NDV is computed as
+      // 1/10*rowcount (Calcite default). Hence, NDV is 13.0 instead of the correct 119.0
+      test("DROP TABLE dfs.tmp.`parquetStale/.stats.drill`");
+      String[] expectedPlan1 = {"HashAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 13.0.*",
+          "Scan.*rowcount = 130.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan1, new String[]{});
+      // query should use the new statistics. NDV remains unaffected since we copy the Q4 into Q5
+      verifyAnalyzeOutput("ANALYZE TABLE dfs.tmp.parquetStale COMPUTE STATISTICS", "9");
+      String[] expectedPlan2 = {"StreamAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 119.0.*",
+          "Scan.*rowcount = 130.0.*"};
+      PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan2, new String[]{});
+      test("DROP TABLE dfs.tmp.`parquetStale/1996/Q5`");
+    } finally {
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
+      resetSessionOption("planner.statistics.use");
+    }
   }
 
   // Test basic histogram creation functionality for int, bigint, double, date, timestamp and boolean data types.
@@ -436,9 +458,10 @@
       String[] expectedPlan6 = {"Filter\\(condition.*\\).*rowcount = 1.0,.*",
         "Scan.*columns=\\[`store_id`\\].*rowcount = 1128.0.*"};
       PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan6, new String[]{});
-
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
+      resetSessionOption("planner.statistics.use");
     }
   }
 
@@ -462,7 +485,8 @@
               .baselineValues("`c_acctbal`", 11)
               .go();
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
     }
   }
 
@@ -498,11 +522,11 @@
               .go();
 
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
     }
   }
 
-
   @Test
   public void testHistogramWithIntervalPredicate() throws Exception {
     try {
@@ -516,7 +540,8 @@
       String[] expectedPlan1 = {"Filter\\(condition.*\\).*rowcount = 59?.*,.*", "Scan.*columns=\\[`o_orderdate`\\].*rowcount = 15000.0.*"};
       PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan1, new String[]{});
     } finally {
-      test("ALTER SESSION SET `planner.slice_target` = " + ExecConstants.SLICE_TARGET_DEFAULT);
+      resetSessionOption("planner.slice_target");
+      resetSessionOption("store.format");
     }
   }
 
diff --git a/exec/vector/src/main/codegen/templates/NullableValueVectors.java b/exec/vector/src/main/codegen/templates/NullableValueVectors.java
index 29cbe39..e32ecc9 100644
--- a/exec/vector/src/main/codegen/templates/NullableValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/NullableValueVectors.java
@@ -579,21 +579,21 @@
     /**
      * Fill in missing values up to, but not including, the given
      * index.
-     * 
+     *
      * @param index the index about to be written, or the total
      * vector length about to be set
      */
-    
+
     @VisibleForTesting
     protected void fillEmpties(int index) {
       values.getMutator().fillEmpties(lastSet, index);
       while (index > bits.getValueCapacity()) {
         bits.reAlloc();
       }
-      
+
       // Set last set to the given index; which the caller
       // will write to
-      
+
       lastSet = index;
     }
 
@@ -753,7 +753,7 @@
       values.getMutator().setValueCount(valueCount);
       bits.getMutator().setValueCount(valueCount);
     }
-    
+
     <#if type.major == "VarLen">
     /** Enables this wrapper container class to participate in bulk mutator logic */
     private final class VarLenBulkInputCallbackImpl implements VarLenBulkInput.BulkInputCallback<VarLenBulkEntry> {
@@ -853,7 +853,7 @@
     public void setSetCount(int n) {
       <#if type.major = "VarLen">lastSet = n - 1;</#if>
     }
-    
+
     // For nullable vectors, exchanging buffers (done elsewhere)
     // requires also exchanging mutator state (done here.)
 
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/NullableVector.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/NullableVector.java
index e4e8cbc..099f113 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/NullableVector.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/NullableVector.java
@@ -25,7 +25,11 @@
 
   public interface Mutator extends ValueVector.Mutator {
 
-    // Used by the vector accessors to force the last set value.
+    /**
+     * Used by the vector accessors to force the last set value.
+     * @param n the value of the last set field used to
+     * fill empties
+     */
 
     void setSetCount(int n);
   }
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/impl/VectorPrinter.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/impl/VectorPrinter.java
index 564e784..5830b30 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/impl/VectorPrinter.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/impl/VectorPrinter.java
@@ -34,7 +34,7 @@
     if (capacity == 0) {
       return;
     }
-    int length = Math.min(maxPrint, capacity);
+    int length = Math.max(maxPrint, vector.getAccessor().getValueCount());
     printOffsets(vector, 0, length);
   }