DRILL-5716: Queue-driven memory allocation

* Creates new core resource management and query queue abstractions.
* Adds queue information to the Protobuf layer.
* Foreman and Planner changes

- Abstracts memory management out to the new resource management layer.
This means deferring generating the physical plan JSON to later in the
process after memory planning.

* Web UI changes

* Adds queue information to the main page and the profile page to each
query.
* Also sorts the list of options displayed in the Web UI.

- Added memory reserve

A new config parameter, exec.queue.memory_reserve_ratio, sets aside a
slice of total memory for operators that do not participate in the
memory assignment process. The default is 20% testing will tell us if
that value should be larger or smaller.

* Additional minor fixes

- Code cleanup.
- Added mechanism to abandon lease release during shutdown.
- Log queue configuration only when the config changes, rather than on
every query.
- Apply Boaz’ option to enforce a minimum memory allocation per
operator.
- Additional logging to help testers see what is happening.

closes #928

3 files changed, 22 insertions, 8 deletions

diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/BatchPrinter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/BatchPrinter.java
index 7e32a4d97..9382f4a46 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/BatchPrinter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/BatchPrinter.java
@@ -34,7 +34,7 @@ import com.google.common.collect.Lists;
 public class BatchPrinter {
   public static void printHyperBatch(VectorAccessible batch, SelectionVector4 sv4) {
     List<String> columns = Lists.newArrayList();
-    for (VectorWrapper vw : batch) {
+    for (VectorWrapper<?> vw : batch) {
       columns.add(vw.getValueVectors()[0].getField().getName());
     }
     int width = columns.size();
@@ -47,7 +47,7 @@ public class BatchPrinter {
         System.out.printf("|\n");
         System.out.println(StringUtils.repeat("-", width * 17 + 1));
       }
-      for (VectorWrapper vw : batch) {
+      for (VectorWrapper<?> vw : batch) {
         Object o = vw.getValueVectors()[sv4.get(j) >>> 16].getAccessor().getObject(sv4.get(j) & 65535);
         String value;
         if (o == null) {
@@ -67,7 +67,7 @@ public class BatchPrinter {
   public static void printBatch(VectorAccessible batch) {
     List<String> columns = Lists.newArrayList();
     List<ValueVector> vectors = Lists.newArrayList();
-    for (VectorWrapper vw : batch) {
+    for (VectorWrapper<?> vw : batch) {
       columns.add(vw.getValueVector().getField().getName());
       vectors.add(vw.getValueVector());
     }
@@ -101,7 +101,7 @@ public class BatchPrinter {
   public static void printBatch(VectorAccessible batch, SelectionVector2 sv2) {
     List<String> columns = Lists.newArrayList();
     List<ValueVector> vectors = Lists.newArrayList();
-    for (VectorWrapper vw : batch) {
+    for (VectorWrapper<?> vw : batch) {
       columns.add(vw.getValueVector().getField().getName());
       vectors.add(vw.getValueVector());
     }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java
index 48724a445..6a805c142 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java
@@ -33,7 +33,22 @@ public class MemoryAllocationUtilities {
   private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MemoryAllocationUtilities.class);
 
   /**
-   * Helper method to setup SortMemoryAllocations
+   * Helper method to setup Memory Allocations
+   * <p>
+   * Plan the memory for buffered operators (the only ones that can spill in this release)
+   * based on assumptions. These assumptions are the amount of memory per node to give
+   * to each query and the number of sort operators per node.
+   * <p>
+   * The reason the total
+   * memory is an assumption is that we have know knowledge of the number of queries
+   * that can run, so we need the user to tell use that information by configuring the
+   * amount of memory to be assumed available to each query.
+   * <p>
+   * The number of sorts per node could be calculated, but we instead simply take
+   * the worst case: the maximum per-query, per-node parallization and assume that
+   * all sorts appear in all fragments &mdash; a gross oversimplification, but one
+   * that Drill has long made.
+   * <p>
    * since this method can be used in multiple places adding it in this class
    * rather than keeping it in Foreman
    * @param plan
@@ -50,7 +65,7 @@ public class MemoryAllocationUtilities {
     // look for external sorts
     final List<PhysicalOperator> bufferedOpList = new LinkedList<>();
     for (final PhysicalOperator op : plan.getSortedOperators()) {
-      if ( op.isBufferedOperator() ) {
+      if (op.isBufferedOperator()) {
         bufferedOpList.add(op);
       }
     }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/TestUtilities.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/TestUtilities.java
index 0200dc5a0..a9f178a66 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/TestUtilities.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/TestUtilities.java
@@ -112,7 +112,7 @@ public class TestUtilities {
    * @param jsonBatches : list of input strings, each element represent a batch. Each string could either
    *                    be in the form of "[{...}, {...}, ..., {...}]", or in the form of "{...}".
    * @param fragContext : fragment context
-   * @param columnsToRead : list of schema pathes to read from JSON reader.
+   * @param columnsToRead : list of schema paths to read from JSON reader.
    * @return
    */
   public static Iterator<RecordReader> getJsonReadersFromBatchString(List<String> jsonBatches, FragmentContext fragContext, List<SchemaPath> columnsToRead) {
@@ -145,5 +145,4 @@ public class TestUtilities {
     }
     return readers.iterator();
   }
-
 }