You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by paul-rogers <gi...@git.apache.org> on 2017/09/26 01:20:11 UTC
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
GitHub user paul-rogers opened a pull request:
https://github.com/apache/drill/pull/960
DRILL-5815: Option to set query memory as percent of total
This PR provides an alternative way to set the memory per query as a percent of system memory.
### Background
Drill is an in-memory query engine, optimized for speed. Historically, all operators used as much memory as needed to perform their work. The sort has supported spilling for a number of releases. In this release, hash agg now also supports spilling.
Once an operator can spill, we can define a memory “budget” for that operator. This has been done by setting a series of options, and making a number of assumptions:
* Define `planner.memory.max_query_memory_per_node` as the amount of memory, per node, to give to each query. The default (which most users never change) is 2 GB.
* Compute the number of buffering operators as:
* The number of buffering operators (sort and hash agg) across all major fragments (as shown in the visualized plan),
* Multiplied by the slice target, typically 70% of the number of CPUs on the node.
* Divide max query memory by the total number of buffering operators to get the memory per operator.
The problem is, with the default value of 2 GB, it is very easy to have sufficient cores, or sufficient buffering operators, that each operator gets a very thin slice of memory (10 or 20 MB).
To work around this, we introduced a new option, `planner.memory.min_memory_per_buffered_op`
which sets a floor on the per operator memory. The default is 40 MB. Thus, even if the above calculations would prefer to give an operator, say, 10 MB of memory, the floor will force the allocation to 40 MB. The result is that actual query use will far exceed the expected budget (by 4X in this example), but the query will run (assuming the necessary memory is, in fact, available.)
This work-around is mostly fine because Drill still has a large number of operators that use unlimited memory, so a bit extra by the limited operators will be lost in the noise.
The problem, now, is that if a machine is generous, and gives Drill 128 GB of memory, say, each query still gets only 2 GB, slices the per-operator memory too small, and either runs out of memory or runs slowly.
### Query Memory as a Percent of Total Memory
This PR adds another option, `planner.memory.percent_per_query`, which provides another way to allocate query memory.
With this option, Drill computes the memory per query per node as:
* `planner.memory.percent_per_query` * the total direct memory, or
* `planner.memory.max_query_memory_per_node`
whichever is greater. For small systems, `planner.memory.max_query_memory_per_node` dominates. For larger systems, `planner.memory.percent_per_query` dominates.
### Computation
To compute the proper number for the user’s workload:
* Determine the memory ratio needed for non-managed operators. (See above.) Call this *_u_* (for unmanaged.)
* Determine the target concurrency. Call this *_n_*.
* Set `planner.memory.percent_per_query` to:
```
planner.memory.percent_per_query = (1 - u) / n
```
### Default Value
Let’s use the computation rules to determine how we arrive at the default setting of 0.05 (5%).
* We allow half of the total memory for unmanaged operators. (*_u_* = 0.5)
* We assume a concurrency of 10. (*_n_* = 10)
* The default value is:
```
(1 - 0.5) / 10 = 0.05 = 5%
```
Why 50% for unmanaged? We have no solid metrics; but most queries do include hash joins and exchanges, so it seems prudent to give half memory those these unlimited operators. (Users may find they need an even larger allowance since the operators are, after all, unlimited in their memory usage.)
Why concurrency of 10? The out-of-the-box configuration of 8 GB direct, 2 GB per query allows a concurrency of 2-3. The Drill web site talks about concurrency in the 100s (on a very large cluster.) The (logarithmic) average of 10<sup>0</sup> and 10<sup>2</sup> is 10<sup>1</sup> or 10.
This is just a default; we expect users to tune the number for their site.
### Queue-Based Memory Assignment
Another PR introduces the idea of using Drill’s ZK-based queues to allocate memory. That mechanism works similarly, except rather than having to assume a concurrency number, the queueing mechanism enforces that number.
Therefore, the new option has no effect when throttling is enabled.
### Disabling the Feature
Perhaps some users prefer to use only the static memory allocation as in prior versions of Drill. Setting `planner.memory.percent_per_query` to 0 effectively disables this technique as it will always produce values lower than `max_query_memory_per_node`.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/paul-rogers/drill DRILL-5815
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/drill/pull/960.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #960
----
commit 8dbee604233b86ba18fee2b72b8a2a872e1d49aa
Author: Paul Rogers <pr...@maprtech.com>
Date: 2017-09-26T00:04:41Z
DRILL-5815: Option to set query memory as percent of total
commit e08356c6e5df1fcb511e64c6aea8bc2ba8047c9a
Author: Paul Rogers <pr...@maprtech.com>
Date: 2017-09-26T00:40:44Z
Added option definition
commit 9127824afee0e7cf701641147bdbcb842ae6bf07
Author: Paul Rogers <pr...@maprtech.com>
Date: 2017-09-26T01:16:21Z
Unit test fix
----
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142577230
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
+
+ for(final PhysicalOperator op : bufferedOpList) {
+ final long alloc = Math.max(opMinMem, op.getInitialAllocation());
+ op.setMaxAllocation(alloc);
+ }
+ }
+
+ /**
+ * Compute per-operator memory based on the computed per-node memory, the
+ * number of operators, and the computed number of fragments (which house
+ * the operators.) Enforces a floor on the amount of memory per operator.
+ *
+ * @param optionManager system option manager
+ * @param maxAllocPerNode computed query memory per node
+ * @param opCount number of buffering operators in this query
+ * @return the per-operator memory
+ */
+
+ public static long computeOperatorMemory(OptionSet optionManager, long maxAllocPerNode, int opCount) {
+ final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
+ final double cpuLoadAverage = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
+ final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpuLoadAverage, maxWidth);
--- End diff --
Wondering - this is the _max_ per node, while the actual number of minor fragments may be smaller; so the division below may leave less memory per operator than is possible ... (yeah, that's the original code, but I just thought of that issue; a future dynamic memory scheme would address such an issue.)
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142574038
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
--- End diff --
A side comment: Wish we could put the maxAllocPerNode figure into the query profile, so users (or testers) could see it.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r143316843
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
--- End diff --
Created DRILL-5852
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142573753
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java ---
@@ -351,7 +351,35 @@
MAX_QUERY_MEMORY_PER_NODE_KEY, 1024 * 1024, Long.MAX_VALUE);
/**
- * Minimum memory alocated to each buffered operator instance.
+ * Alternative way to compute per-query-per-node memory as a percent
+ * of the total available system memory.
+ * <p>
+ * Suggestion for computation.
+ * <ul>
+ * <li>Assume an allowance for non-managed operators. Default assumption:
+ * 50%</li>
+ * <li>Assume a desired number of concurrent queries. Default assumption:
+ * 10.</li>
+ * <li>The value of this parameter is<br>
+ * (1 - non-managed allowance) / concurrency</li>
--- End diff --
Would it be simpler if working in a 100 base for the percent; i.e. "1" here be "100", to work in percentage numbers as integers (0...100) instead of a decimal fraction ?
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by paul-rogers <gi...@git.apache.org>.
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142580204
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
+
+ for(final PhysicalOperator op : bufferedOpList) {
+ final long alloc = Math.max(opMinMem, op.getInitialAllocation());
+ op.setMaxAllocation(alloc);
+ }
+ }
+
+ /**
+ * Compute per-operator memory based on the computed per-node memory, the
+ * number of operators, and the computed number of fragments (which house
+ * the operators.) Enforces a floor on the amount of memory per operator.
+ *
+ * @param optionManager system option manager
+ * @param maxAllocPerNode computed query memory per node
+ * @param opCount number of buffering operators in this query
+ * @return the per-operator memory
+ */
+
+ public static long computeOperatorMemory(OptionSet optionManager, long maxAllocPerNode, int opCount) {
+ final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
+ final double cpuLoadAverage = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
+ final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpuLoadAverage, maxWidth);
+ final long maxOperatorAlloc = maxAllocPerNode / (opCount * maxWidthPerNode);
+ logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
+
+ // User configurable option to allow forcing minimum memory.
+ // Ensure that the buffered ops receive the minimum memory needed to make progress.
+ // Without this, the math might work out to allocate too little memory.
+
+ return Math.max(maxOperatorAlloc,
+ optionManager.getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP));
+ }
+
+ /**
+ * Per-node memory calculations based on a number of constraints.
+ * <p>
+ * Factored out into a separate method to allow unit testing.
+ * @param config Drill config
+ * @param optionManager system options
+ * @param directMemory amount of direct memory
+ * @return memory per query per node
+ */
+
+ @VisibleForTesting
+ public static long computeQueryMemory(DrillConfig config, OptionSet optionManager, long directMemory) {
+
+ // Memory computed as a percent of total memory.
+
+ long perQueryMemory = Math.round(directMemory *
+ optionManager.getOption(ExecConstants.PERCENT_MEMORY_PER_QUERY));
+
+ // But, must allow at least the amount given explicitly for
+ // backward compatibility.
+
+ perQueryMemory = Math.max(perQueryMemory,
+ optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE));
+
+ // Compute again as either the total direct memory, or the
+ // configured maximum top-level allocation (10 GB).
+
+ long maxAllocPerNode = Math.min(directMemory,
+ config.getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
+
+ // Final amount per node per query is the minimum of these two.
+
+ maxAllocPerNode = Math.min(maxAllocPerNode, perQueryMemory);
--- End diff --
Yes. That code is original; seems overly cautious. If we start with all memory and slice it up, it is not clear how we get a number larger than what we started with...
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/drill/pull/960
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by paul-rogers <gi...@git.apache.org>.
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142580136
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
+
+ for(final PhysicalOperator op : bufferedOpList) {
+ final long alloc = Math.max(opMinMem, op.getInitialAllocation());
+ op.setMaxAllocation(alloc);
+ }
+ }
+
+ /**
+ * Compute per-operator memory based on the computed per-node memory, the
+ * number of operators, and the computed number of fragments (which house
+ * the operators.) Enforces a floor on the amount of memory per operator.
+ *
+ * @param optionManager system option manager
+ * @param maxAllocPerNode computed query memory per node
+ * @param opCount number of buffering operators in this query
+ * @return the per-operator memory
+ */
+
+ public static long computeOperatorMemory(OptionSet optionManager, long maxAllocPerNode, int opCount) {
+ final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
+ final double cpuLoadAverage = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
+ final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpuLoadAverage, maxWidth);
+ final long maxOperatorAlloc = maxAllocPerNode / (opCount * maxWidthPerNode);
+ logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
+
+ // User configurable option to allow forcing minimum memory.
+ // Ensure that the buffered ops receive the minimum memory needed to make progress.
+ // Without this, the math might work out to allocate too little memory.
+
+ return Math.max(maxOperatorAlloc,
+ optionManager.getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP));
+ }
+
+ /**
+ * Per-node memory calculations based on a number of constraints.
+ * <p>
+ * Factored out into a separate method to allow unit testing.
+ * @param config Drill config
+ * @param optionManager system options
+ * @param directMemory amount of direct memory
+ * @return memory per query per node
+ */
+
+ @VisibleForTesting
+ public static long computeQueryMemory(DrillConfig config, OptionSet optionManager, long directMemory) {
+
+ // Memory computed as a percent of total memory.
+
+ long perQueryMemory = Math.round(directMemory *
+ optionManager.getOption(ExecConstants.PERCENT_MEMORY_PER_QUERY));
+
+ // But, must allow at least the amount given explicitly for
+ // backward compatibility.
+
+ perQueryMemory = Math.max(perQueryMemory,
+ optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE));
+
+ // Compute again as either the total direct memory, or the
+ // configured maximum top-level allocation (10 GB).
+
+ long maxAllocPerNode = Math.min(directMemory,
+ config.getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
--- End diff --
Yeah, I don't know the original reason well enough to tinker with it. Might be possible to go over. We can test, just change the value of that `TOP_LEVEL_MAX_ALLOC` option to a bigger number on a large machine and see if anything breaks. If not, we can remove the limit.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by paul-rogers <gi...@git.apache.org>.
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142579717
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java ---
@@ -351,7 +351,35 @@
MAX_QUERY_MEMORY_PER_NODE_KEY, 1024 * 1024, Long.MAX_VALUE);
/**
- * Minimum memory alocated to each buffered operator instance.
+ * Alternative way to compute per-query-per-node memory as a percent
+ * of the total available system memory.
+ * <p>
+ * Suggestion for computation.
+ * <ul>
+ * <li>Assume an allowance for non-managed operators. Default assumption:
+ * 50%</li>
+ * <li>Assume a desired number of concurrent queries. Default assumption:
+ * 10.</li>
+ * <li>The value of this parameter is<br>
+ * (1 - non-managed allowance) / concurrency</li>
--- End diff --
I thought about that. Then I remembered that we use ratios elsewhere such as `planner.cpu_load_average`, `planner.affinity_factor`, and `planner.memory.hash_agg_table_factor`. I didn't see any place we use the percent-as-integer form.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by paul-rogers <gi...@git.apache.org>.
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142579793
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
--- End diff --
Great idea! Please file a JIRA for it. I learned how to do this (and get the value into the UI) for another task, so this improvement should be easy.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142576399
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
+
+ for(final PhysicalOperator op : bufferedOpList) {
+ final long alloc = Math.max(opMinMem, op.getInitialAllocation());
+ op.setMaxAllocation(alloc);
+ }
+ }
+
+ /**
+ * Compute per-operator memory based on the computed per-node memory, the
+ * number of operators, and the computed number of fragments (which house
+ * the operators.) Enforces a floor on the amount of memory per operator.
+ *
+ * @param optionManager system option manager
+ * @param maxAllocPerNode computed query memory per node
+ * @param opCount number of buffering operators in this query
+ * @return the per-operator memory
+ */
+
+ public static long computeOperatorMemory(OptionSet optionManager, long maxAllocPerNode, int opCount) {
+ final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
+ final double cpuLoadAverage = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
+ final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpuLoadAverage, maxWidth);
+ final long maxOperatorAlloc = maxAllocPerNode / (opCount * maxWidthPerNode);
+ logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
+
+ // User configurable option to allow forcing minimum memory.
+ // Ensure that the buffered ops receive the minimum memory needed to make progress.
+ // Without this, the math might work out to allocate too little memory.
+
+ return Math.max(maxOperatorAlloc,
+ optionManager.getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP));
+ }
+
+ /**
+ * Per-node memory calculations based on a number of constraints.
+ * <p>
+ * Factored out into a separate method to allow unit testing.
+ * @param config Drill config
+ * @param optionManager system options
+ * @param directMemory amount of direct memory
+ * @return memory per query per node
+ */
+
+ @VisibleForTesting
+ public static long computeQueryMemory(DrillConfig config, OptionSet optionManager, long directMemory) {
+
+ // Memory computed as a percent of total memory.
+
+ long perQueryMemory = Math.round(directMemory *
+ optionManager.getOption(ExecConstants.PERCENT_MEMORY_PER_QUERY));
+
+ // But, must allow at least the amount given explicitly for
+ // backward compatibility.
+
+ perQueryMemory = Math.max(perQueryMemory,
+ optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE));
+
+ // Compute again as either the total direct memory, or the
+ // configured maximum top-level allocation (10 GB).
+
+ long maxAllocPerNode = Math.min(directMemory,
+ config.getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
+
+ // Final amount per node per query is the minimum of these two.
+
+ maxAllocPerNode = Math.min(maxAllocPerNode, perQueryMemory);
--- End diff --
So this ensures that the query-per-node does not get more than the whole direct memory (or that 10GB).
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by paul-rogers <gi...@git.apache.org>.
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142580052
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
+
+ for(final PhysicalOperator op : bufferedOpList) {
+ final long alloc = Math.max(opMinMem, op.getInitialAllocation());
+ op.setMaxAllocation(alloc);
+ }
+ }
+
+ /**
+ * Compute per-operator memory based on the computed per-node memory, the
+ * number of operators, and the computed number of fragments (which house
+ * the operators.) Enforces a floor on the amount of memory per operator.
+ *
+ * @param optionManager system option manager
+ * @param maxAllocPerNode computed query memory per node
+ * @param opCount number of buffering operators in this query
+ * @return the per-operator memory
+ */
+
+ public static long computeOperatorMemory(OptionSet optionManager, long maxAllocPerNode, int opCount) {
+ final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
+ final double cpuLoadAverage = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
+ final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpuLoadAverage, maxWidth);
--- End diff --
Good point. The queue-based memory assignment feature uses a different model: it counts the actual operators per node and uses the max across all nodes. For a simple one-fragment query, the difference in the memory given to each operator is huge: all of the memory rather than a small slice. We might want to port that logic to the non-queued case as well.
---
[GitHub] drill issue #960: DRILL-5815: Option to set query memory as percent of total
Posted by arina-ielchiieva <gi...@git.apache.org>.
Github user arina-ielchiieva commented on the issue:
https://github.com/apache/drill/pull/960
+1, please resolve the conflicts.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by paul-rogers <gi...@git.apache.org>.
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142579925
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
--- End diff --
Another excellent idea, please file a JIRA for this one as well. Getting that right will be a bit tricky as it won't be as easy to say that, "below the memory-per-query-per-node we do this, above it we do that." Once the other operators kick in, the memory per sort/hash agg drops so a step above, say, 2 GB might actually *decrease* the memory per sort and hash agg.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by paul-rogers <gi...@git.apache.org>.
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r144926648
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
--- End diff --
MEP? Do you mean Drill 1.13?
The dynamic memory idea needs thought. In fact, this is exactly what we do today: each operator grabs the memory it needs; and sends messages downstream if it is memory-constrained. The reason we are doing so much work is that the existing technique is very, very hard to get right.
We are moving toward fixed budgets. Once that works, we can set a fixed budget per fragment and "slosh" memory between operators in the same fragment based on an empirical study of memory use.
It will be far, far harder to slosh memory between fragments as a fragment lifecycle, with buffering operators, is caret-shaped: ramp up to a peak, then ramp down. Coordination is needed to know, if a fragment is using only 100 MB of 1 GB, that this is because the fragment just started, and is ramping up (to full 1 GB use), vs. the operator is ramping down and won't be using additional memory.
Since all of this is a big project, the present change tries to do the best we can with the situation we currently find ourselves in.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142576288
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
+
+ for(final PhysicalOperator op : bufferedOpList) {
+ final long alloc = Math.max(opMinMem, op.getInitialAllocation());
+ op.setMaxAllocation(alloc);
+ }
+ }
+
+ /**
+ * Compute per-operator memory based on the computed per-node memory, the
+ * number of operators, and the computed number of fragments (which house
+ * the operators.) Enforces a floor on the amount of memory per operator.
+ *
+ * @param optionManager system option manager
+ * @param maxAllocPerNode computed query memory per node
+ * @param opCount number of buffering operators in this query
+ * @return the per-operator memory
+ */
+
+ public static long computeOperatorMemory(OptionSet optionManager, long maxAllocPerNode, int opCount) {
+ final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
+ final double cpuLoadAverage = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
+ final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpuLoadAverage, maxWidth);
+ final long maxOperatorAlloc = maxAllocPerNode / (opCount * maxWidthPerNode);
+ logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
+
+ // User configurable option to allow forcing minimum memory.
+ // Ensure that the buffered ops receive the minimum memory needed to make progress.
+ // Without this, the math might work out to allocate too little memory.
+
+ return Math.max(maxOperatorAlloc,
+ optionManager.getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP));
+ }
+
+ /**
+ * Per-node memory calculations based on a number of constraints.
+ * <p>
+ * Factored out into a separate method to allow unit testing.
+ * @param config Drill config
+ * @param optionManager system options
+ * @param directMemory amount of direct memory
+ * @return memory per query per node
+ */
+
+ @VisibleForTesting
+ public static long computeQueryMemory(DrillConfig config, OptionSet optionManager, long directMemory) {
+
+ // Memory computed as a percent of total memory.
+
+ long perQueryMemory = Math.round(directMemory *
+ optionManager.getOption(ExecConstants.PERCENT_MEMORY_PER_QUERY));
+
+ // But, must allow at least the amount given explicitly for
+ // backward compatibility.
+
+ perQueryMemory = Math.max(perQueryMemory,
+ optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE));
+
+ // Compute again as either the total direct memory, or the
+ // configured maximum top-level allocation (10 GB).
+
+ long maxAllocPerNode = Math.min(directMemory,
+ config.getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
--- End diff --
So a query can not go above 10GB per node ? What about a cluster with 256GB per node installed, running a single giant query -- why not use more ? (Yes this was in the original code, but I did not know then that it meant 10GB ...)
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by paul-rogers <gi...@git.apache.org>.
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142579551
--- Diff: exec/java-exec/src/main/resources/drill-module.conf ---
@@ -240,7 +240,7 @@ drill.exec: {
}
},
sort: {
- purge.threshold : 1000,
+ purge.threshold: 1000,
--- End diff --
Rebased onto master and resolved the merge conflict in favor of Tim's change.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142574389
--- Diff: exec/java-exec/src/main/resources/drill-module.conf ---
@@ -240,7 +240,7 @@ drill.exec: {
}
},
sort: {
- purge.threshold : 1000,
+ purge.threshold: 1000,
--- End diff --
Careful with the merge here -- Tim changed this number to "10" (see DRILL-5755), and it should end up as "10", not "1000" which would destroy Tim's fix !!!
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r142575862
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
--- End diff --
An improvement idea: Also include the Hash Join (and the Exchange?) operators in the division math here. Indeed these operators pay no attention to the memory limit, however this would work around the arbitrary fixed %50 assumption; instead dynamically going %100 when there are no Hash Joins, or much smaller when there are many Hash Joins.
---
[GitHub] drill pull request #960: DRILL-5815: Option to set query memory as percent o...
Posted by Ben-Zvi <gi...@git.apache.org>.
Github user Ben-Zvi commented on a diff in the pull request:
https://github.com/apache/drill/pull/960#discussion_r143317064
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java ---
@@ -56,30 +59,90 @@ public static void setupBufferedOpsMemoryAllocations(final PhysicalPlan plan, fi
}
// if there are any sorts, compute the maximum allocation, and set it on them
- if (bufferedOpList.size() > 0) {
- final OptionManager optionManager = queryContext.getOptions();
- double cpu_load_average = optionManager.getOption(ExecConstants.CPU_LOAD_AVERAGE);
- final long maxWidth = optionManager.getOption(ExecConstants.MAX_WIDTH_PER_NODE);
- final long maxWidthPerNode = ExecConstants.MAX_WIDTH_PER_NODE.computeMaxWidth(cpu_load_average,maxWidth);
- long maxAllocPerNode = Math.min(DrillConfig.getMaxDirectMemory(),
- queryContext.getConfig().getLong(RootAllocatorFactory.TOP_LEVEL_MAX_ALLOC));
- maxAllocPerNode = Math.min(maxAllocPerNode,
- optionManager.getOption(ExecConstants.MAX_QUERY_MEMORY_PER_NODE_KEY).num_val);
- final long maxOperatorAlloc = maxAllocPerNode / (bufferedOpList.size() * maxWidthPerNode);
- logger.debug("Max buffered operator alloc: {}", maxOperatorAlloc);
-
- // User configurable option to allow forcing minimum memory.
- // Ensure that the buffered ops receive the minimum memory needed to make progress.
- // Without this, the math might work out to allocate too little memory.
- final long opMinMem = queryContext.getOptions().getOption(ExecConstants.MIN_MEMORY_PER_BUFFERED_OP_KEY).num_val;
-
- for(final PhysicalOperator op : bufferedOpList) {
-
- long alloc = Math.max(maxOperatorAlloc, op.getInitialAllocation());
- alloc = Math.max(alloc, opMinMem);
- op.setMaxAllocation(alloc);
- }
- }
plan.getProperties().hasResourcePlan = true;
+ if (bufferedOpList.isEmpty()) {
+ return;
+ }
+
+ // Setup options, etc.
+
+ final OptionManager optionManager = queryContext.getOptions();
+ final long directMemory = DrillConfig.getMaxDirectMemory();
+
+ // Compute per-node, per-query memory.
+
+ final long maxAllocPerNode = computeQueryMemory(queryContext.getConfig(), optionManager, directMemory);
+ logger.debug("Memory per query per node: {}", maxAllocPerNode);
+
+ // Now divide up the memory by slices and operators.
+
+ final long opMinMem = computeOperatorMemory(optionManager, maxAllocPerNode, bufferedOpList.size());
--- End diff --
This seems like a short term idea (MEP 4 ?) ; else later the Hash Join would become a "buffered operator", and possibly we'd use a dynamic memory scheme (i.e., allocate minimum to each operator, and add more to those in need) which would cover this issue better.
---