diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
index c421c73..3a16353 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
@@ -162,7 +162,6 @@
   public static final PositiveLongValidator HASHJOIN_RUNTIME_FILTER_MAX_WAITING_TIME = new PositiveLongValidator(HASHJOIN_RUNTIME_FILTER_MAX_WAITING_TIME_KEY, Character.MAX_VALUE, null);
 
 
-
   // Hash Aggregate Options
   public static final String HASHAGG_NUM_PARTITIONS_KEY = "exec.hashagg.num_partitions";
   public static final LongValidator HASHAGG_NUM_PARTITIONS_VALIDATOR = new RangeLongValidator(HASHAGG_NUM_PARTITIONS_KEY, 1, 128,
@@ -189,6 +188,12 @@
   public static final BooleanValidator HASHAGG_FALLBACK_ENABLED_VALIDATOR = new BooleanValidator(HASHAGG_FALLBACK_ENABLED_KEY,
       new OptionDescription("Hash Aggregates ignore memory limits when enabled (true). When disabled (false), Hash Aggregates fail when memory is set too low."));
 
+  // Partitioner options
+  public static final String PARTITIONER_MEMORY_REDUCTION_THRESHOLD_KEY = "exec.partition.mem_throttle";
+  public static final LongValidator PARTITIONER_MEMORY_REDUCTION_THRESHOLD_VALIDATOR =
+      new RangeLongValidator(PARTITIONER_MEMORY_REDUCTION_THRESHOLD_KEY, 0, Integer.MAX_VALUE,
+      new OptionDescription("Linearly reduces partition sender buffer row count after this number of receivers. Default is 0 (disabled). (Since Drill 1.18)"));
+
   public static final String SSL_PROVIDER = "drill.exec.ssl.provider"; // valid values are "JDK", "OPENSSL" // default JDK
   public static final String SSL_PROTOCOL = "drill.exec.ssl.protocol"; // valid values are SSL, SSLV2, SSLV3, TLS, TLSV1, TLSv1.1, TLSv1.2(default)
   public static final String SSL_KEYSTORE_TYPE = "drill.exec.ssl.keyStoreType";
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
index b4790c8..2c580da 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
@@ -29,6 +29,8 @@
 import org.apache.drill.exec.expr.CodeGenerator;
 import org.apache.drill.exec.server.options.OptionSet;
 import org.codehaus.commons.compiler.CompileException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Implements the "plain Java" method of code generation and
@@ -76,10 +78,9 @@
  * The setting to prefer plain Java is ignored for any remaining generated
  * classes not marked as plain Java capable.
  */
-
 public class ClassBuilder {
 
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ClassBuilder.class);
+  private static final Logger logger = LoggerFactory.getLogger(ClassBuilder.class);
   public static final String CODE_DIR_OPTION = CodeCompiler.COMPILE_BASE + ".code_dir";
 
   private final DrillConfig config;
@@ -95,7 +96,6 @@
     // point your debugger to the directory set below, and you
     // can step into the code for debugging. Code is not saved
     // be default because doing so is expensive and unnecessary.
-
     codeDir = new File(config.getString(CODE_DIR_OPTION));
   }
 
@@ -109,7 +109,6 @@
    * @return the class that the code generator defines
    * @throws ClassTransformationException
    */
-
   public Class<?> getImplementationClass(CodeGenerator<?> cg) throws ClassTransformationException {
     try {
       return compileClass(cg);
@@ -133,23 +132,19 @@
     final long t1 = System.nanoTime();
 
     // Get the plain Java code.
-
     String code = cg.getGeneratedCode();
 
     // Get the class names (dotted, file path, etc.)
-
     String className = cg.getMaterializedClassName();
     ClassTransformer.ClassNames name = new ClassTransformer.ClassNames(className);
 
     // A key advantage of this method is that the code can be
     // saved and debugged, if needed.
-
     if (cg.isCodeToBeSaved()) {
       saveCode(code, name);
     }
 
     // Compile the code and load it into a class loader.
-
     CachedClassLoader classLoader = new CachedClassLoader();
     ClassCompilerSelector compilerSelector = new ClassCompilerSelector(classLoader, config, options);
     Map<String,byte[]> results = compilerSelector.compile(name, code);
@@ -165,7 +160,6 @@
                   (System.nanoTime() - t1 + 500_000) / 1_000_000);
 
     // Get the class from the class loader.
-
     try {
       return classLoader.findClass(className);
     } catch (ClassNotFoundException e) {
@@ -184,7 +178,6 @@
    * @param code the source code
    * @param name the class name
    */
-
   private void saveCode(String code, ClassNames name) {
 
     String pathName = name.slash + ".java";
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 843d20b..00d5bae 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
@@ -945,7 +945,7 @@
       // everything in memory
       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.",
+              + "Forcing legacy behavior of using unbounded memory in order to prevent regressions.",
           numPartitions,
           FileUtils.byteCountToDisplaySize(buildCalc.getMaxReservedMemory()),
           FileUtils.byteCountToDisplaySize(allocator.getLimit()));
@@ -984,7 +984,7 @@
     // is enabled
     if (reason == null) {
       boolean fallbackEnabled = context.getOptions()
-          .getOption(ExecConstants.HASHJOIN_FALLBACK_ENABLED_KEY).bool_val;
+          .getBoolean(ExecConstants.HASHJOIN_FALLBACK_ENABLED_KEY);
       if (fallbackEnabled) {
         logger.warn(
             "Spilling is disabled - not enough memory available for internal partitioning. Falling back"
@@ -992,8 +992,9 @@
       } else {
         throw UserException.resourceError().message(String.format(
             "Not enough memory for internal partitioning and fallback mechanism for "
-                + "HashJoin to use unbounded memory is disabled. Either enable fallback config %s using Alter "
-                + "session/system command or increase memory limit for Drillbit",
+                + "HashJoin to use unbounded memory is disabled.\n" +
+                "Either enable fallback option %s using ALTER "
+                + "SESSION/SYSTEM command or increase the memory limit for the Drillbit",
             ExecConstants.HASHJOIN_FALLBACK_ENABLED_KEY)).build(logger);
       }
     } else {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java
index 76c60e8..4f22c5c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java
@@ -48,8 +48,9 @@
   void initialize();
   void clear();
   List<? extends PartitionOutgoingBatch> getOutgoingBatches();
+
   /**
-   * Method to get PartitionOutgoingBatch based on the fact that there can be > 1 Partitioner
+   * Get PartitionOutgoingBatch based on the fact that there can be > 1 Partitioner
    * @param index
    * @return PartitionOutgoingBatch that matches index within Partitioner. This method can
    * return null if index does not fall within boundary of this Partitioner
@@ -58,4 +59,4 @@
   OperatorStats getStats();
 
   TemplateClassDefinition<Partitioner> TEMPLATE_DEFINITION = new TemplateClassDefinition<>(Partitioner.class, PartitionerTemplate.class);
-}
\ No newline at end of file
+}
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 a3149b8..95e62e0 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
@@ -24,10 +24,11 @@
 import javax.inject.Named;
 
 import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ExecConstants;
 import org.apache.drill.exec.compile.sig.RuntimeOverridden;
 import org.apache.drill.exec.exception.SchemaChangeException;
 import org.apache.drill.exec.expr.ClassGenerator;
-import org.apache.drill.exec.expr.TypeHelper;
+import org.apache.drill.exec.memory.BaseAllocator;
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.ops.AccountingDataTunnel;
 import org.apache.drill.exec.ops.ExchangeFragmentContext;
@@ -50,7 +51,6 @@
 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.vector.ValueVector;
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -104,9 +104,44 @@
     this.end = end;
     doSetup(context, incoming, null);
 
-    // Half the outgoing record batch size if the number of senders exceeds 1000 to reduce the total amount of memory
-    // allocated.
-    if (popConfig.getDestinations().size() > 1000) {
+    // Consider the system/session option to allow the buffer size to shrink
+    // linearly with the increase in slice count, over some limit:
+    // exec.partition.mem_throttle:
+    // The default is 0, which leaves the current logic unchanged.
+    // If set to a positive value, then when the slice count exceeds that
+    // amount, the buffer size per sender is reduced.
+    // The reduction factor is 1 / (slice count - threshold), with a minimum
+    // batch size of 256 records.
+    //
+    // So, if we set the threshold at 2, and run 10 slices, each slice will
+    // get 1024 / 8 = 256 records.
+    //
+    // This option controls memory, but at an obvious cost of increasing overhead.
+    // One could argue that this is a good thing. As the number of senders
+    // increases, the number of records going to each sender decreases, which
+    // increases the time that batches must accumulate before they are sent.
+    //
+    // If the option is enabled, and buffer size reduction kicks in, you'll
+    // find an info-level log message which details the reduction:
+    // exec.partition.mem_throttle is set to 2: 10 receivers,
+    //       reduced send buffer size from 1024 to 256 rows
+    //
+    // See  DRILL-7675, DRILL-7686.
+    int destinationCount = popConfig.getDestinations().size();
+    int reductionCutoff = oContext.getFragmentContext().getOptions().getInt(
+        ExecConstants.PARTITIONER_MEMORY_REDUCTION_THRESHOLD_KEY);
+    if (reductionCutoff > 0 && destinationCount >= reductionCutoff) {
+      int reducedBatchSize = Math.max(256,
+          (DEFAULT_RECORD_BATCH_SIZE + 1) / (destinationCount - reductionCutoff));
+      outgoingRecordBatchSize = BaseAllocator.nextPowerOfTwo(reducedBatchSize) - 1;
+      logger.info("{} is set to {}: {} receivers, reduced send buffer size from {} to {} rows",
+          ExecConstants.PARTITIONER_MEMORY_REDUCTION_THRESHOLD_KEY,
+          reductionCutoff, destinationCount,
+          DEFAULT_RECORD_BATCH_SIZE, outgoingRecordBatchSize);
+    } else if (destinationCount > 1000) {
+      // Half the outgoing record batch size if the number of senders exceeds 1000 to reduce the total amount of memory
+      // allocated.
+
       // Always keep the recordCount as (2^x) - 1 to better utilize the memory allocation in ValueVectors
       outgoingRecordBatchSize = (DEFAULT_RECORD_BATCH_SIZE + 1)/2 - 1;
     }
@@ -114,7 +149,7 @@
     int fieldId = 0;
     for (MinorFragmentEndpoint destination : popConfig.getDestinations()) {
       // create outgoingBatches only for subset of Destination Points
-      if ( fieldId >= start && fieldId < end ) {
+      if (fieldId >= start && fieldId < end) {
         logger.debug("start: {}, count: {}, fieldId: {}", start, end, fieldId);
         outgoingBatches.add(newOutgoingRecordBatch(stats, popConfig,
           context.getDataTunnel(destination.getEndpoint()), context, oContext.getAllocator(), destination.getId()));
@@ -149,7 +184,6 @@
    * generated inner class. Byte-code manipulation appears to fix up the byte codes
    * directly. The name is special, it must be "new" + inner class name.
    */
-
   protected OutgoingRecordBatch newOutgoingRecordBatch(
                                OperatorStats stats, HashPartitionSender operator, AccountingDataTunnel tunnel,
                                FragmentContext context, BufferAllocator allocator, int oppositeMinorFragmentId) {
@@ -259,24 +293,23 @@
     private final AccountingDataTunnel tunnel;
     private final HashPartitionSender operator;
     private final FragmentContext context;
-    private final BufferAllocator allocator;
-    private final VectorContainer vectorContainer = new VectorContainer();
+    private final VectorContainer vectorContainer;
     private final int oppositeMinorFragmentId;
     private final OperatorStats stats;
 
-    private boolean isLast = false;
-    private boolean dropAll = false;
+    private boolean isLast;
+    private boolean dropAll;
     private int recordCount;
     private int totalRecords;
 
     public OutgoingRecordBatch(OperatorStats stats, HashPartitionSender operator, AccountingDataTunnel tunnel,
                                FragmentContext context, BufferAllocator allocator, int oppositeMinorFragmentId) {
       this.context = context;
-      this.allocator = allocator;
       this.operator = operator;
       this.tunnel = tunnel;
       this.stats = stats;
       this.oppositeMinorFragmentId = oppositeMinorFragmentId;
+      this.vectorContainer = new VectorContainer(allocator);
     }
 
     protected void copy(int inIndex) throws IOException {
@@ -376,9 +409,7 @@
     }
 
     private void allocateOutgoingRecordBatch() {
-      for (VectorWrapper<?> v : vectorContainer) {
-        v.getValueVector().allocateNew();
-      }
+      vectorContainer.allocate(outgoingRecordBatchSize);
     }
 
     public void updateStats(FragmentWritableBatch writableBatch) {
@@ -391,12 +422,7 @@
      * Initialize the OutgoingBatch based on the current schema in incoming RecordBatch
      */
     public void initializeBatch() {
-      for (VectorWrapper<?> v : incoming) {
-        // create new vector
-        ValueVector outgoingVector = TypeHelper.getNewVector(v.getField(), allocator);
-        outgoingVector.setInitialCapacity(outgoingRecordBatchSize);
-        vectorContainer.add(outgoingVector);
-      }
+      vectorContainer.buildFrom(incoming.getSchema());
       allocateOutgoingRecordBatch();
       try {
         doSetup(incoming, vectorContainer);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
index 61850b7..6d8865d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/BatchSchema.java
@@ -123,7 +123,6 @@
    * Hence we should make use of {@link BatchSchema#isEquivalent(BatchSchema)} method instead since
    * {@link MaterializedField#isEquivalent(MaterializedField)} method is updated to remove the reference check.
    */
-
   @Override
   public boolean equals(Object obj) {
     if (this == obj) {
@@ -145,7 +144,6 @@
 
     // Compare names.
     // (DRILL-5525: actually compares all fields.)
-
     if (!fields.equals(other.fields)) {
       return false;
     }
@@ -153,7 +151,6 @@
     // Compare types
     // (DRILL-5525: this code is redundant because any differences
     // will fail above.)
-
     for (int i = 0; i < fields.size(); i++) {
       MajorType t1 = fields.get(i).getType();
       MajorType t2 = other.fields.get(i).getType();
@@ -181,7 +178,6 @@
    * the {@link MaterializedField#isEquivalent(MaterializedField)} rules,
    * false otherwise
    */
-
   public boolean isEquivalent(BatchSchema other) {
     if (this == other) {
       return true;
@@ -209,17 +205,15 @@
   private boolean majorTypeEqual(MajorType t1, MajorType t2) {
     if (t1.equals(t2)) {
       return true;
-    }
-    if (!t1.getMinorType().equals(t2.getMinorType())) {
+    } else if (!t1.getMinorType().equals(t2.getMinorType())) {
       return false;
-    }
-    if (!t1.getMode().equals(t2.getMode())) {
+    } else if (!t1.getMode().equals(t2.getMode())) {
       return false;
-    }
-    if (!Sets.newHashSet(t1.getSubTypeList()).equals(Sets.newHashSet(t2.getSubTypeList()))) {
+    } else if (!Sets.newHashSet(t1.getSubTypeList()).equals(Sets.newHashSet(t2.getSubTypeList()))) {
       return false;
+    } else {
+      return true;
     }
-    return true;
   }
 
   /**
@@ -237,7 +231,6 @@
    * @param otherSchema the schema to merge with this one
    * @return the new, merged, schema
    */
-
   public BatchSchema merge(BatchSchema otherSchema) {
     if (selectionVectorMode != SelectionVectorMode.NONE ||
         otherSchema.selectionVectorMode != SelectionVectorMode.NONE) {
@@ -249,4 +242,17 @@
     mergedFields.addAll(otherSchema.fields);
     return new BatchSchema(selectionVectorMode, mergedFields);
   }
+
+  /**
+   * Format the schema into a multi-line format. Useful when debugging a query with
+   * a very wide schema as the usual single-line format is far too hard to read.
+   */
+  public String format() {
+    StringBuilder buf = new StringBuilder();
+    buf.append("Batch Schema:\n");
+    for (MaterializedField field : fields) {
+      field.format(buf, 1);
+    }
+    return buf.toString();
+  }
 }
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 3796e5a..4ec0b8d 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
@@ -440,6 +440,14 @@
     return wrappers.size();
   }
 
+  public void allocate(int recordCount) {
+    for (VectorWrapper<?> w : wrappers) {
+      ValueVector v = w.getValueVector();
+      v.setInitialCapacity(recordCount);
+      v.allocateNew();
+    }
+  }
+
   public void allocateNew() {
     for (VectorWrapper<?> w : wrappers) {
       w.getValueVector().allocateNew();
@@ -559,4 +567,10 @@
       addOrGet(wrapper.getField());
     }
   }
+
+  public void buildFrom(BatchSchema sourceSchema) {
+    for (MaterializedField field : sourceSchema) {
+      add(TypeHelper.getNewVector(field, allocator));
+    }
+  }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/BaseOptionManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/BaseOptionManager.java
index a502e07..ee786c0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/BaseOptionManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/BaseOptionManager.java
@@ -69,6 +69,11 @@
   }
 
   @Override
+  public int getInt(String name) {
+    return (int) getLong(name);
+  }
+
+  @Override
   public long getLong(String name) {
     return getByType(name, Kind.LONG).num_val;
   }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/OptionSet.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/OptionSet.java
index e96d571..7e85a76 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/OptionSet.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/OptionSet.java
@@ -83,10 +83,19 @@
    * @throws IllegalArgumentException if the option is undefined or
    * is not of the correct data type
    */
-
   boolean getBoolean(String name);
 
   /**
+   * Return the value of a long option as an int
+   *
+   * @param name option name
+   * @return the long value
+   * @throws IllegalArgumentException if the option is undefined or
+   * is not of the correct data type
+   */
+  int getInt(String name);
+
+  /**
    * Return the value of a long option.
    *
    * @param name option name
@@ -94,7 +103,6 @@
    * @throws IllegalArgumentException if the option is undefined or
    * is not of the correct data type
    */
-
   long getLong(String name);
 
   /**
@@ -105,7 +113,6 @@
    * @throws IllegalArgumentException if the option is undefined or
    * is not of the correct data type
    */
-
   double getDouble(String name);
 
   /**
@@ -116,6 +123,5 @@
    * @throws IllegalArgumentException if the option is undefined or
    * is not of the correct data type
    */
-
   String getString(String name);
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
index 5d2598c..3eb643c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
@@ -188,6 +188,7 @@
       new OptionDefinition(ExecConstants.PARQUET_FLAT_BATCH_NUM_RECORDS_VALIDATOR, new OptionMetaData(OptionValue.AccessibleScopes.SYSTEM_AND_SESSION, true, true)),
       new OptionDefinition(ExecConstants.PARQUET_FLAT_BATCH_MEMORY_SIZE_VALIDATOR, new OptionMetaData(OptionValue.AccessibleScopes.SYSTEM_AND_SESSION, true, true)),
       new OptionDefinition(ExecConstants.PARQUET_COMPLEX_BATCH_NUM_RECORDS_VALIDATOR, new OptionMetaData(OptionValue.AccessibleScopes.SYSTEM_AND_SESSION, true, true)),
+      new OptionDefinition(ExecConstants.PARTITIONER_MEMORY_REDUCTION_THRESHOLD_VALIDATOR),
       new OptionDefinition(ExecConstants.JSON_READER_ALL_TEXT_MODE_VALIDATOR),
       new OptionDefinition(ExecConstants.JSON_WRITER_NAN_INF_NUMBERS_VALIDATOR),
       new OptionDefinition(ExecConstants.JSON_READER_NAN_INF_NUMBERS_VALIDATOR),
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/CopyUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/CopyUtil.java
index 28a98aa..f24ea09 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/CopyUtil.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/CopyUtil.java
@@ -28,15 +28,18 @@
 import com.sun.codemodel.JVar;
 
 public class CopyUtil {
-  public static void generateCopies(ClassGenerator<?> g, VectorAccessible batch, boolean hyper){
+
+  public static void generateCopies(ClassGenerator<?> g, VectorAccessible batch, boolean hyper) {
     // we have parallel ids for each value vector so we don't actually have to deal with managing the ids at all.
     int fieldId = 0;
 
     JExpression inIndex = JExpr.direct("inIndex");
     JExpression outIndex = JExpr.direct("outIndex");
-    for(VectorWrapper<?> vv : batch) {
+    for (VectorWrapper<?> vv : batch) {
       String copyMethod;
-      if (!Types.isFixedWidthType(vv.getField().getType()) || Types.isRepeated(vv.getField().getType()) || Types.isComplex(vv.getField().getType())) {
+      if (!Types.isFixedWidthType(vv.getField().getType()) ||
+           Types.isRepeated(vv.getField().getType()) ||
+           Types.isComplex(vv.getField().getType())) {
         copyMethod = "copyFromSafe";
       } else {
         copyMethod = "copyFrom";
@@ -53,19 +56,13 @@
           .build();
       JVar outVV = g.declareVectorValueSetupAndMember("outgoing", outFieldId);
 
-      if(hyper){
-
-        g.getEvalBlock().add(
-                outVV
+      if (hyper) {
+        g.getEvalBlock().add(outVV
                         .invoke(copyMethod)
-                        .arg(
-                                inIndex.band(JExpr.lit((int) Character.MAX_VALUE)))
+                        .arg(inIndex.band(JExpr.lit((int) Character.MAX_VALUE)))
                         .arg(outIndex)
-                        .arg(
-                                inVV.component(inIndex.shrz(JExpr.lit(16)))
-                        )
-        );
-      }else{
+                        .arg(inVV.component(inIndex.shrz(JExpr.lit(16)))));
+      } else {
         g.getEvalBlock().add(outVV.invoke(copyMethod).arg(inIndex).arg(outIndex).arg(inVV));
       }
 
@@ -73,5 +70,4 @@
       fieldId++;
     }
   }
-
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
index 962d74e..1cecbd3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
@@ -55,7 +55,7 @@
 import org.apache.drill.exec.rpc.UserClientConnection;
 import org.apache.drill.exec.server.DrillbitContext;
 import org.apache.drill.exec.server.FailureUtils;
-import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.server.options.OptionSet;
 import org.apache.drill.exec.testing.ControlsInjector;
 import org.apache.drill.exec.testing.ControlsInjectorFactory;
 import org.apache.drill.exec.util.Pointer;
@@ -65,6 +65,8 @@
 import org.apache.drill.exec.work.foreman.rm.QueryQueue.QueryQueueException;
 import org.apache.drill.exec.work.foreman.rm.QueryResourceManager;
 import org.codehaus.jackson.map.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.Date;
@@ -93,8 +95,8 @@
  */
 
 public class Foreman implements Runnable {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Foreman.class);
-  private static final org.slf4j.Logger queryLogger = org.slf4j.LoggerFactory.getLogger("query.logger");
+  private static final Logger logger = LoggerFactory.getLogger(Foreman.class);
+  private static final Logger queryLogger = LoggerFactory.getLogger("query.logger");
   private static final ControlsInjector injector = ControlsInjectorFactory.getInjector(Foreman.class);
 
   public enum ProfileOption { SYNC, ASYNC, NONE }
@@ -108,8 +110,7 @@
   private final QueryManager queryManager; // handles lower-level details of query execution
   private final DrillbitContext drillbitContext;
   private final UserClientConnection initiatingClient; // used to send responses
-  private boolean resume = false;
-  private final ProfileOption profileOption;
+  private boolean resume;
 
   private final QueryResourceManager queryRM;
 
@@ -122,7 +123,7 @@
   private String queryText;
 
   private RuntimeFilterRouter runtimeFilterRouter;
-  private boolean enableRuntimeFilter;
+  private final boolean enableRuntimeFilter;
 
   /**
    * Constructor. Sets up the Foreman, but does not initiate any execution.
@@ -154,11 +155,9 @@
     this.queryRM = drillbitContext.getResourceManager().newQueryRM(this);
     this.fragmentsRunner = new FragmentsRunner(bee, initiatingClient, drillbitContext, this);
     this.queryStateProcessor = new QueryStateProcessor(queryIdString, queryManager, drillbitContext, new ForemanResult());
-    this.profileOption = setProfileOption(queryContext.getOptions());
     this.enableRuntimeFilter = queryContext.getOptions().getOption(ExecConstants.HASHJOIN_ENABLE_RUNTIME_FILTER_KEY).bool_val;
   }
 
-
   /**
    * @return query id
    */
@@ -326,11 +325,15 @@
     }
   }
 
-  private ProfileOption setProfileOption(OptionManager options) {
-    if (! options.getOption(ExecConstants.ENABLE_QUERY_PROFILE_VALIDATOR)) {
+  private ProfileOption getProfileOption(QueryContext queryContext) {
+    if (queryContext.isSkipProfileWrite()) {
       return ProfileOption.NONE;
     }
-    if (options.getOption(ExecConstants.QUERY_PROFILE_DEBUG_VALIDATOR)) {
+    OptionSet options = queryContext.getOptions();
+    if (!options.getBoolean(ExecConstants.ENABLE_QUERY_PROFILE_OPTION)) {
+      return ProfileOption.NONE;
+    }
+    if (options.getBoolean(ExecConstants.QUERY_PROFILE_DEBUG_OPTION)) {
       return ProfileOption.SYNC;
     } else {
       return ProfileOption.ASYNC;
@@ -792,8 +795,8 @@
 
       // Debug option: write query profile before sending final results so that
       // the client can be certain the profile exists.
-      final boolean skipProfileWrite = queryContext.isSkipProfileWrite();
-      if (profileOption == ProfileOption.SYNC && !skipProfileWrite) {
+      ProfileOption profileOption = getProfileOption(queryContext);
+      if (profileOption == ProfileOption.SYNC) {
         queryManager.writeFinalProfile(uex);
       }
 
@@ -823,7 +826,7 @@
       // storage write; query completion occurs in parallel with profile
       // persistence.
 
-      if (profileOption == ProfileOption.ASYNC && !skipProfileWrite) {
+      if (profileOption == ProfileOption.ASYNC) {
         queryManager.writeFinalProfile(uex);
       }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java
index 0c140a4..66739de 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java
@@ -58,12 +58,15 @@
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
- * Each Foreman holds its own QueryManager.  This manages the events associated with execution of a particular query across all fragments.
+ * Each Foreman holds its own QueryManager. This manages the events associated
+ * with execution of a particular query across all fragments.
  */
 public class QueryManager implements AutoCloseable {
-  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryManager.class);
+  private static final Logger logger = LoggerFactory.getLogger(QueryManager.class);
 
   private final Map<DrillbitEndpoint, NodeTracker> nodeMap = Maps.newHashMap();
   private final QueryId queryId;
@@ -240,10 +243,10 @@
   public void close() throws Exception { }
 
   /*
-     * This assumes that the FragmentStatusListener implementation takes action when it hears
-     * that the target fragment has acknowledged the signal. As a result, this listener doesn't do anything
-     * but log messages.
-     */
+   * This assumes that the FragmentStatusListener implementation takes action when it hears
+   * that the target fragment has acknowledged the signal. As a result, this listener doesn't do anything
+   * but log messages.
+   */
   private static class SignalListener extends EndpointListener<Ack, FragmentHandle> {
     /**
      * An enum of possible signals that {@link SignalListener} listens to.
@@ -461,13 +464,10 @@
    * there is a node failure, we can then correctly track how many outstanding messages will never arrive.
    */
   private class NodeTracker {
-    private final DrillbitEndpoint endpoint;
     private final AtomicInteger totalFragments = new AtomicInteger(0);
     private final AtomicInteger completedFragments = new AtomicInteger(0);
 
-    public NodeTracker(final DrillbitEndpoint endpoint) {
-      this.endpoint = endpoint;
-    }
+    public NodeTracker(final DrillbitEndpoint endpoint) { }
 
     /**
      * Increments the number of fragment this node is running.
@@ -506,7 +506,6 @@
       }
       return true;
     }
-
   }
 
   /**
@@ -556,7 +555,6 @@
     }
   };
 
-
   public DrillbitStatusListener getDrillbitStatusListener() {
     return drillbitStatusListener;
   }
@@ -601,7 +599,6 @@
             new ForemanException(String.format("One more more nodes lost connectivity during query.  Identified nodes were [%s].",
                 failedNodeList)));
       }
-
     }
   };
 }
diff --git a/exec/java-exec/src/main/resources/drill-module.conf b/exec/java-exec/src/main/resources/drill-module.conf
index 9e306e6..46ae53a 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -541,6 +541,7 @@
     exec.java_compiler_janino_maxsize: 262144,
     exec.max_hash_table_size: 1073741824,
     exec.min_hash_table_size: 65536,
+    exec.partition.mem_throttle: 0,
     exec.persistent_table.umask: "002",
     exec.query.progress.update: true,
     exec.query_profile.debug_mode: false,
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/ClusterFixture.java b/exec/java-exec/src/test/java/org/apache/drill/test/ClusterFixture.java
index 80e2a26..c8b64a4 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/ClusterFixture.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/ClusterFixture.java
@@ -80,7 +80,6 @@
     {
       // Properties here mimic those in drill-root/pom.xml, Surefire plugin
       // configuration. They allow tests to run successfully in Eclipse.
-
       put(ExecConstants.SYS_STORE_PROVIDER_LOCAL_ENABLE_WRITE, false);
 
       // The CTTAS function requires that the default temporary workspace be
@@ -88,36 +87,30 @@
       // dfs.tmp. But, the test setup marks dfs.tmp as read-only. To work
       // around this, tests are supposed to use dfs. So, we need to
       // set the default temporary workspace to dfs.tmp.
-
       put(ExecConstants.DEFAULT_TEMPORARY_WORKSPACE, DFS_TMP_SCHEMA);
       put(ExecConstants.HTTP_ENABLE, false);
       put("drill.catastrophic_to_standard_out", true);
 
       // Verbose errors.
-
       put(ExecConstants.ENABLE_VERBOSE_ERRORS_KEY, true);
 
       // See Drillbit.close. The Drillbit normally waits a specified amount
       // of time for ZK registration to drop. But, embedded Drillbits normally
       // don't use ZK, so no need to wait.
-
       put(ExecConstants.ZK_REFRESH, 0);
 
       // This is just a test, no need to be heavy-duty on threads.
       // This is the number of server and client RPC threads. The
       // production default is DEFAULT_SERVER_RPC_THREADS.
-
       put(ExecConstants.BIT_SERVER_RPC_THREADS, 2);
 
       // No need for many scanners except when explicitly testing that
       // behavior. Production default is DEFAULT_SCAN_THREADS
-
       put(ExecConstants.SCAN_THREADPOOL_SIZE, 4);
 
       // Define a useful root location for the ZK persistent
       // storage. Profiles will go here when running in distributed
       // mode.
-
       put(ZookeeperPersistentStoreProvider.DRILL_EXEC_SYS_STORE_PROVIDER_ZK_BLOBROOT,
           "/tmp/drill/tests");
     }
@@ -147,9 +140,9 @@
       startDrillbits();
       applyOptions();
     } catch (Exception e) {
+
       // Translate exceptions to unchecked to avoid cluttering
       // tests. Failures will simply fail the test itself.
-
       throw new IllegalStateException("Cluster fixture setup failed", e);
     }
   }
@@ -158,7 +151,7 @@
    * Set the client properties to be used by client fixture.
    */
   private void setClientProps() {
-      clientProps = builder.clientProps;
+    clientProps = builder.clientProps;
   }
 
   public Properties getClientProps() {
@@ -166,17 +159,17 @@
   }
 
   private void configureZk() {
-    // Start ZK if requested.
 
+    // Start ZK if requested.
     String zkConnect;
     if (builder.zkHelper != null) {
-      // Case where the test itself started ZK and we're only using it.
 
+      // Case where the test itself started ZK and we're only using it.
       zkHelper = builder.zkHelper;
       ownsZK = false;
     } else if (builder.localZkCount > 0) {
-      // Case where we need a local ZK just for this test cluster.
 
+      // Case where we need a local ZK just for this test cluster.
       zkHelper = new ZookeeperHelper();
       zkHelper.startZookeeper(builder.localZkCount);
       ownsZK = true;
@@ -189,7 +182,6 @@
       // in config properties defined at run time. Drill does not allow
       // combining locally-set properties and a config file: it is one
       // or the other.
-
       if (builder.configBuilder().hasResource()) {
         throw new IllegalArgumentException("Cannot specify a local ZK while using an external config file.");
       }
@@ -202,27 +194,27 @@
   }
 
   private void createConfig() throws Exception {
+
     // Create a config
     // Because of the way DrillConfig works, we can set the ZK
     // connection string only if a property set is provided.
-
     config = builder.configBuilder.build();
 
     if (builder.usingZk) {
-      // Distribute drillbit using ZK (in-process or external)
 
+      // Distribute drillbit using ZK (in-process or external)
       serviceSet = null;
       usesZk = true;
     } else {
-      // Embedded Drillbit.
 
+      // Embedded Drillbit.
       serviceSet = RemoteServiceSet.getLocalServiceSet();
     }
   }
 
   private void startDrillbits() throws Exception {
-    // Start the Drillbits.
 
+    // Start the Drillbits.
     Preconditions.checkArgument(builder.bitCount > 0);
     int bitCount = builder.bitCount;
     for (int i = 0; i < bitCount; i++) {
@@ -230,7 +222,6 @@
       bit.run();
 
       // Bit name and registration.
-
       String name;
       if (builder.bitNames != null && i < builder.bitNames.length) {
         name = builder.bitNames[i];
@@ -239,7 +230,6 @@
         // Name the Drillbit by default. Most tests use one Drillbit,
         // so make the name simple: "drillbit." Only add a numeric suffix
         // when the test creates multiple bits.
-
         if (bitCount == 1) {
           name = DEFAULT_BIT_NAME;
         } else {
@@ -250,7 +240,6 @@
 
       // Remember the first Drillbit, this is the default one returned from
       // drillbit().
-
       if (i == 0) {
         defaultDrillbit = bit;
       }
@@ -330,7 +319,6 @@
    * @return a test client. Client will be closed when this cluster
    * fixture closes, or can be closed early
    */
-
   public ClientFixture client(String host, int port) {
     return clientBuilder()
       .property(DrillProperties.DRILLBIT_CONNECTION, String.format("%s:%d", host, port))
@@ -573,6 +561,8 @@
     props.setProperty(ExecConstants.UDF_DIRECTORY_ROOT, dirTestWatcher.getHomeDir().getAbsolutePath());
     props.setProperty(ExecConstants.SYS_STORE_PROVIDER_LOCAL_PATH, dirTestWatcher.getStoreDir().getAbsolutePath());
     props.setProperty(ExecConstants.UDF_DIRECTORY_FS, FileSystem.DEFAULT_FS);
+    // ALTER SESSION profiles are seldom interesting
+    props.setProperty(ExecConstants.SKIP_ALTER_SESSION_QUERY_PROFILE, Boolean.TRUE.toString());
 
     builder.configBuilder.configProps(props);
     return builder;
@@ -640,7 +630,6 @@
    * @param value the value to encode
    * @return the SQL-acceptable string equivalent
    */
-
   public static String stringify(Object value) {
     if (value == null) {
       return null;
@@ -654,8 +643,8 @@
   }
 
   public static String getResource(String resource) throws IOException {
-    // Unlike the Java routines, Guava does not like a leading slash.
 
+    // Unlike the Java routines, Guava does not like a leading slash.
     final URL url = Resources.getResource(trimSlash(resource));
     if (url == null) {
       throw new IOException(
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/ClusterFixtureBuilder.java b/exec/java-exec/src/test/java/org/apache/drill/test/ClusterFixtureBuilder.java
index 24beb3a..9bc5312 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/ClusterFixtureBuilder.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/ClusterFixtureBuilder.java
@@ -214,7 +214,6 @@
     usingZk = true;
 
     // Using ZK. Turn refresh wait back on.
-
     return configProperty(ExecConstants.ZK_REFRESH, DEFAULT_ZK_REFRESH);
   }
 
@@ -236,7 +235,6 @@
     usingZk = true;
 
     // Using ZK. Turn refresh wait back on.
-
     configProperty(ExecConstants.ZK_REFRESH, DEFAULT_ZK_REFRESH);
     return this;
   }
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java b/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java
index e1b5d99..40fe40e 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/record/MaterializedField.java
@@ -23,6 +23,7 @@
 import java.util.LinkedHashSet;
 import java.util.Objects;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.drill.common.types.TypeProtos.DataMode;
 import org.apache.drill.common.types.TypeProtos.MajorType;
 import org.apache.drill.common.types.TypeProtos.MinorType;
@@ -393,27 +394,9 @@
    */
   public String toString(boolean includeChildren) {
     final int maxLen = 10;
-    final StringBuilder builder = new StringBuilder();
-    builder
-      .append("[`")
-      .append(name)
-      .append("` (")
-      .append(type.getMinorType().name());
-
-    if (type.hasPrecision() && (type.getPrecision() > 0 || Types.isDecimalType(type))) {
-      builder.append("(");
-      builder.append(type.getPrecision());
-      if (type.hasScale() && type.getScale() > 0) {
-        builder.append(", ");
-        builder.append(type.getScale());
-      }
-      builder.append(")");
-    }
-
-    builder
-      .append(":")
-      .append(type.getMode().name())
-      .append(")");
+    final StringBuilder builder = new StringBuilder()
+      .append("[`");
+    prefix(builder);
 
     if (includeChildren) {
       if (type.getSubTypeCount() > 0) {
@@ -436,17 +419,63 @@
         .toString();
   }
 
+  private void prefix(StringBuilder builder) {
+    builder
+      .append(name)
+      .append("` (")
+      .append(type.getMinorType().name());
+
+    if (type.hasPrecision() && (type.getPrecision() > 0 || Types.isDecimalType(type))) {
+      builder.append("(");
+      builder.append(type.getPrecision());
+      if (type.hasScale() && type.getScale() > 0) {
+        builder.append(", ");
+        builder.append(type.getScale());
+      }
+      builder.append(")");
+    }
+
+    builder
+      .append(":")
+      .append(type.getMode().name())
+      .append(")");
+  }
+
   @Override
   public String toString() {
     return toString(true);
   }
 
+  public String format() {
+    final StringBuilder builder = new StringBuilder();
+    format(builder, 0);
+    return builder.toString();
+  }
+
+  /**
+   * Format the field in a multi-line format, with children (but not subtypes)
+   * indented. Useful for wide rows where the single-line format is too hard
+   * to read.
+   */
+  public void format(StringBuilder builder, int level) {
+    builder.append(StringUtils.repeat(' ', level));
+    prefix(builder);
+    if (children != null && ! children.isEmpty()) {
+      builder.append(":\n");
+      for (MaterializedField child : children) {
+        child.format(builder, level + 1);
+      }
+    } else {
+      builder.append("\n");
+    }
+  }
+
   /**
    * Return true if two fields have identical MinorType and Mode.
    */
   public boolean hasSameTypeAndMode(MaterializedField that) {
-    return (getType().getMinorType() == that.getType().getMinorType())
-        && (getType().getMode() == that.getType().getMode());
+    return getType().getMinorType() == that.getType().getMinorType()
+        && getType().getMode() == that.getType().getMode();
   }
 
   private String toString(Collection<?> collection, int maxLen) {