drill-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (DRILL-5325) Implement sub-operator unit tests for managed external sort
Date Tue, 06 Jun 2017 22:41:18 GMT

    [ https://issues.apache.org/jira/browse/DRILL-5325?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16039802#comment-16039802
] 

ASF GitHub Bot commented on DRILL-5325:
---------------------------------------

Github user paul-rogers commented on a diff in the pull request:

    https://github.com/apache/drill/pull/808#discussion_r120498177
  
    --- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMemoryManager.java
---
    @@ -0,0 +1,506 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you under the Apache License, Version 2.0 (the
    + * "License"); you may not use this file except in compliance
    + * with the License.  You may obtain a copy of the License at
    + *
    + * http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +package org.apache.drill.exec.physical.impl.xsort.managed;
    +
    +public class SortMemoryManager {
    +
    +  /**
    +   * Maximum memory this operator may use. Usually comes from the
    +   * operator definition, but may be overridden by a configuration
    +   * parameter for unit testing.
    +   */
    +
    +  private final long memoryLimit;
    +
    +  /**
    +   * Estimated size of the records for this query, updated on each
    +   * new batch received from upstream.
    +   */
    +
    +  private int estimatedRowWidth;
    +
    +  /**
    +   * Size of the merge batches that this operator produces. Generally
    +   * the same as the merge batch size, unless low memory forces a smaller
    +   * value.
    +   */
    +
    +  private int expectedMergeBatchSize;
    +
    +  /**
    +   * Estimate of the input batch size based on the largest batch seen
    +   * thus far.
    +   */
    +  private int estimatedInputBatchSize;
    +
    +  /**
    +   * Maximum memory level before spilling occurs. That is, we can buffer input
    +   * batches in memory until we reach the level given by the buffer memory pool.
    +   */
    +
    +  private long bufferMemoryLimit;
    +
    +  /**
    +   * Maximum memory that can hold batches during the merge
    +   * phase.
    +   */
    +
    +  private long mergeMemoryLimit;
    +
    +  /**
    +   * The target size for merge batches sent downstream.
    +   */
    +
    +  private int preferredMergeBatchSize;
    +
    +  /**
    +   * The configured size for each spill batch.
    +   */
    +  private int preferredSpillBatchSize;
    +
    +  /**
    +   * Estimated number of rows that fit into a single spill batch.
    +   */
    +
    +  private int spillBatchRowCount;
    +
    +  /**
    +   * The estimated actual spill batch size which depends on the
    +   * details of the data rows for any particular query.
    +   */
    +
    +  private int expectedSpillBatchSize;
    +
    +  /**
    +   * The number of records to add to each output batch sent to the
    +   * downstream operator or spilled to disk.
    +   */
    +
    +  private int mergeBatchRowCount;
    +
    +  private SortConfig config;
    +
    +//  private long spillPoint;
    +
    +//  private long minMergeMemory;
    +
    +  private int estimatedInputSize;
    +
    +  private boolean potentialOverflow;
    +
    +  public SortMemoryManager(SortConfig config, long memoryLimit) {
    +    this.config = config;
    +
    +    // The maximum memory this operator can use as set by the
    +    // operator definition (propagated to the allocator.)
    +
    +    if (config.maxMemory() > 0) {
    +      this.memoryLimit = Math.min(memoryLimit, config.maxMemory());
    +    } else {
    +      this.memoryLimit = memoryLimit;
    +    }
    +
    +    preferredSpillBatchSize = config.spillBatchSize();;
    +    preferredMergeBatchSize = config.mergeBatchSize();
    +  }
    +
    +  /**
    +   * Update the data-driven memory use numbers including:
    +   * <ul>
    +   * <li>The average size of incoming records.</li>
    +   * <li>The estimated spill and output batch size.</li>
    +   * <li>The estimated number of average-size records per
    +   * spill and output batch.</li>
    +   * <li>The amount of memory set aside to hold the incoming
    +   * batches before spilling starts.</li>
    +   * </ul>
    +   * <p>
    +   * Under normal circumstances, the amount of memory available is much
    +   * larger than the input, spill or merge batch sizes. The primary question
    +   * is to determine how many input batches we can buffer during the load
    +   * phase, and how many spill batches we can merge during the merge
    +   * phase.
    +   *
    +   * @param batchSize the overall size of the current batch received from
    +   * upstream
    +   * @param batchRowWidth the width in bytes (including overhead) of each
    +   * row in the current input batch
    +   * @param batchRowCount the number of actual (not filtered) records in
    +   * that upstream batch
    +   */
    +
    +  public void updateEstimates(int batchSize, int batchRowWidth, int batchRowCount) {
    +
    +    // The record count should never be zero, but better safe than sorry...
    +
    +    if (batchRowCount == 0) {
    +      return; }
    +
    +
    +    // Update input batch estimates.
    +    // Go no further if nothing changed.
    +
    +    if (! updateInputEstimates(batchSize, batchRowWidth, batchRowCount)) {
    +      return;
    +    }
    +
    +    updateSpillSettings();
    +    updateMergeSettings();
    +    adjustForLowMemory();
    +    logSettings(batchRowCount);
    +  }
    +
    +  private boolean updateInputEstimates(int batchSize, int batchRowWidth, int batchRowCount)
{
    +
    +    // The row width may end up as zero if all fields are nulls or some
    +    // other unusual situation. In this case, assume a width of 10 just
    +    // to avoid lots of special case code.
    +
    +    if (batchRowWidth == 0) {
    +      batchRowWidth = 10;
    +    }
    +
    +    // We know the batch size and number of records. Use that to estimate
    +    // the average record size. Since a typical batch has many records,
    +    // the average size is a fairly good estimator. Note that the batch
    +    // size includes not just the actual vector data, but any unused space
    +    // resulting from power-of-two allocation. This means that we don't
    +    // have to do size adjustments for input batches as we will do below
    +    // when estimating the size of other objects.
    +
    +    // Record sizes may vary across batches. To be conservative, use
    +    // the largest size observed from incoming batches.
    +
    +    int origRowEstimate = estimatedRowWidth;
    +    estimatedRowWidth = Math.max(estimatedRowWidth, batchRowWidth);
    +
    +    // Maintain an estimate of the incoming batch size: the largest
    +    // batch yet seen. Used to reserve memory for the next incoming
    +    // batch. Because we are using the actual observed batch size,
    +    // the size already includes overhead due to power-of-two rounding.
    +
    +    long origInputBatchSize = estimatedInputBatchSize;
    +    estimatedInputBatchSize = Math.max(estimatedInputBatchSize, batchSize);
    +
    +    // Estimate the total size of each incoming batch plus sv2. Note that, due
    +    // to power-of-two rounding, the allocated sv2 size might be twice the data size.
    +
    +    estimatedInputSize = estimatedInputBatchSize + 4 * batchRowCount;
    +
    +    // Return whether anything changed.
    +
    +    return estimatedRowWidth != origRowEstimate || estimatedInputBatchSize != origInputBatchSize;
    +  }
    +
    +  /**
    +   * Determine the number of records to spill per spill batch. The goal is to
    +   * spill batches of either 64K records, or as many records as fit into the
    +   * amount of memory dedicated to each spill batch, whichever is less.
    +   */
    +
    +  private void updateSpillSettings() {
    +
    +    spillBatchRowCount = rowsPerBatch(preferredSpillBatchSize);
    +
    +    // Compute the actual spill batch size which may be larger or smaller
    +    // than the preferred size depending on the row width. Double the estimated
    +    // memory needs to allow for power-of-two rounding.
    +
    +    expectedSpillBatchSize = batchForRows(spillBatchRowCount);
    +
    +    // Determine the minimum memory needed for spilling. Spilling is done just
    +    // before accepting a spill batch, so we must spill if we don't have room for a
    +    // (worst case) input batch. To spill, we need room for the spill batch created
    +    // by merging the batches already in memory.
    +
    +    bufferMemoryLimit = memoryLimit - expectedSpillBatchSize;
    +  }
    +
    +  /**
    +   * Determine the number of records per batch per merge step. The goal is to
    +   * merge batches of either 64K records, or as many records as fit into the
    +   * amount of memory dedicated to each merge batch, whichever is less.
    +   */
    +
    +  private void updateMergeSettings() {
    +
    +    mergeBatchRowCount = rowsPerBatch(preferredMergeBatchSize);
    +    expectedMergeBatchSize = batchForRows(mergeBatchRowCount);
    +
    +    // The merge memory pool assumes we can spill all input batches. The memory
    +    // available to hold spill batches for merging is total memory minus the
    +    // expected output batch size.
    +
    +    mergeMemoryLimit = memoryLimit - expectedMergeBatchSize;
    +  }
    +
    +  /**
    +   * In a low-memory situation we have to approach the memory assignment
    +   * problem from a different angle. Memory is low enough that we can't
    +   * fit the incoming batches (of a size decided by the upstream operator)
    +   * and our usual spill or merge batch sizes. Instead, we have to
    +   * determine the largest spill and merge batch sizes possible given
    +   * the available memory, input batch size and row width. We shrink the
    +   * sizes of the batches we control to try to make things fit into limited
    +   * memory. At some point, however, if we cannot fit even two input
    +   * batches and even the smallest merge match, then we will run into an
    +   * out-of-memory condition and we log a warning.
    +   * <p>
    +   * Note that these calculations are a bit crazy: it is Drill that
    +   * decided to allocate the small memory, it is Drill that created the
    +   * large incoming batches, and so it is Drill that created the low
    +   * memory situation. Over time, a better fix for this condition is to
    +   * control memory usage at the query level so that the sort is guaranteed
    +   * to have sufficient memory. But, since we don't yet have the luxury
    +   * of making such changes, we just live with the situation as we find
    +   * it.
    +   */
    +
    +  private void adjustForLowMemory() {
    +
    +    long loadHeadroom = bufferMemoryLimit - 2 * estimatedInputSize;
    +    long mergeHeadroom = mergeMemoryLimit - 2 * expectedSpillBatchSize;
    +    if (loadHeadroom >= 0  &&  mergeHeadroom >= 0) {
    +      return;
    +    }
    +
    +    lowMemorySpillBatchSize();
    +    lowMemoryMergeBatchSize();
    +
    +    // Sanity check: if we've been given too little memory to make progress,
    +    // issue a warning but proceed anyway. Should only occur if something is
    +    // configured terribly wrong.
    +
    +    long minNeeds = 2 * estimatedInputSize + expectedSpillBatchSize;
    +    if (minNeeds > memoryLimit) {
    +      ExternalSortBatch.logger.warn("Potential memory overflow during load phase! " +
    +          "Minumum needed = {} bytes, actual available = {} bytes",
    +          minNeeds, memoryLimit);
    +      bufferMemoryLimit = 0;
    +      potentialOverflow = true;
    +    }
    +
    +    // Sanity check
    +
    +    minNeeds = 2 * expectedSpillBatchSize + expectedMergeBatchSize;
    +    if (minNeeds > memoryLimit) {
    +      ExternalSortBatch.logger.warn("Potential memory overflow during merge phase! "
+
    +          "Minumum needed = {} bytes, actual available = {} bytes",
    +          minNeeds, memoryLimit);
    +      mergeMemoryLimit = 0;
    +      potentialOverflow = true;
    +    }
    +  }
    +
    +  /**
    +   * If we are in a low-memory condition, then we might not have room for the
    +   * default spill batch size. In that case, pick a smaller size based on
    +   * the observation that we need two input batches and
    +   * one spill batch to make progress.
    +   */
    +
    +  private void lowMemorySpillBatchSize() {
    +
    +    // The "expected" size is with power-of-two rounding in some vectors.
    +    // We later work backwards to the row count assuming average internal
    +    // fragmentation.
    +
    +    // Must hold two input batches. Use (most of) the rest for the spill batch.
    +
    +    expectedSpillBatchSize = (int) (memoryLimit - 2 * estimatedInputSize);
    +
    +    // But, in the merge phase, we need two spill batches and one output batch.
    +    // (Assume that the spill and merge are equal sizes.)
    +    // Use 3/4 of memory for each batch (to allow power-of-two rounding:
    +
    +    expectedSpillBatchSize = (int) Math.min(expectedSpillBatchSize, memoryLimit/3);
    +
    +    // Never going to happen, but let's ensure we don't somehow create large batches.
    +
    +    expectedSpillBatchSize = Math.max(expectedSpillBatchSize, SortConfig.MIN_SPILL_BATCH_SIZE);
    +
    +    // Must hold at least one row to spill. That is, we can make progress if we
    +    // create spill files that consist of single-record batches.
    +
    +    expectedSpillBatchSize = Math.max(expectedSpillBatchSize, estimatedRowWidth);
    +
    +    // Work out the spill batch count needed by the spill code. Allow room for
    +    // power-of-two rounding.
    +
    +    spillBatchRowCount = rowsPerBatch(expectedSpillBatchSize);
    +
    +    // Finally, figure out when we must spill.
    +
    +    bufferMemoryLimit = memoryLimit - expectedSpillBatchSize;
    +  }
    +
    +  /**
    +   * For merge batch, we must hold at least two spill batches and
    +   * one output batch.
    +   */
    +
    +  private void lowMemoryMergeBatchSize() {
    +    expectedMergeBatchSize = (int) (memoryLimit - 2 * expectedSpillBatchSize);
    +    expectedMergeBatchSize = Math.max(expectedMergeBatchSize, SortConfig.MIN_MERGE_BATCH_SIZE);
    +    expectedMergeBatchSize = Math.max(expectedMergeBatchSize, estimatedRowWidth);
    +    mergeBatchRowCount = rowsPerBatch(expectedMergeBatchSize);
    +    mergeMemoryLimit = memoryLimit - expectedMergeBatchSize;
    +  }
    +
    +  /**
    +   * Log the calculated values. Turn this on if things seem amiss.
    +   * Message will appear only when the values change.
    +   */
    +
    +  private void logSettings(int actualRecordCount) {
    +
    +    ExternalSortBatch.logger.debug("Input Batch Estimates: record size = {} bytes; input
batch = {} bytes, {} records",
    +                 estimatedRowWidth, estimatedInputBatchSize, actualRecordCount);
    +    ExternalSortBatch.logger.debug("Merge batch size = {} bytes, {} records; spill file
size: {} bytes",
    +                 expectedSpillBatchSize, spillBatchRowCount, config.spillFileSize());
    +    ExternalSortBatch.logger.debug("Output batch size = {} bytes, {} records",
    +                 expectedMergeBatchSize, mergeBatchRowCount);
    +    ExternalSortBatch.logger.debug("Available memory: {}, buffer memory = {}, merge memory
= {}",
    +                 memoryLimit, bufferMemoryLimit, mergeMemoryLimit);
    +  }
    +
    +  public enum MergeAction { SPILL, MERGE, NONE }
    +
    +  public static class MergeTask {
    +    public MergeAction action;
    +    public int count;
    +
    +    public MergeTask(MergeAction action, int count) {
    +      this.action = action;
    +      this.count = count;
    +    }
    +  }
    +
    +  public MergeTask consolidateBatches(long allocMemory, int inMemCount, int spilledRunsCount)
{
    +
    +    // Determine additional memory needed to hold one batch from each
    +    // spilled run.
    +
    +    // If the on-disk batches and in-memory batches need more memory than
    +    // is available, spill some in-memory batches.
    +
    +    if (inMemCount > 0) {
    +      long mergeSize = spilledRunsCount * expectedSpillBatchSize;
    +      if (allocMemory + mergeSize > mergeMemoryLimit) {
    +        return new MergeTask(MergeAction.SPILL, 0);
    +      }
    +    }
    +
    +    // Maximum batches that fit into available memory.
    +
    +    int mergeLimit = (int) ((mergeMemoryLimit - allocMemory) / expectedSpillBatchSize);
    +
    +    // Can't merge more than the merge limit.
    +
    +    mergeLimit = Math.min(mergeLimit, config.mergeLimit());
    +
    +    // How many batches to merge?
    +
    +    int mergeCount = spilledRunsCount - mergeLimit;
    +    if (mergeCount <= 0) {
    +      return new MergeTask(MergeAction.NONE, 0);
    +    }
    +
    +    // We will merge. This will create yet another spilled
    +    // run. Account for that.
    +
    +    mergeCount += 1;
    +
    +    // Must merge at least 2 batches to make progress.
    +    // This is the the (at least one) excess plus the allowance
    +    // above for the new one.
    +
    +    // Can't merge more than the limit.
    +
    +    mergeCount = Math.min(mergeCount, config.mergeLimit());
    +
    +    // Do the merge, then loop to try again in case not
    +    // all the target batches spilled in one go.
    +
    +    return new MergeTask(MergeAction.MERGE, mergeCount);
    +  }
    +
    +  /**
    +   * Compute the number of rows per batch assuming that the batch is
    +   * subject to average internal fragmentation due to power-of-two
    +   * rounding on vectors.
    +   * <p>
    +   * <pre>[____|__$__]</pre>
    +   * In the above, the brackets represent the whole vector. The
    +   * first half is always full. When the first half filled, the second
    +   * half was allocated. On average, the second half will be half full.
    +   *
    +   * @param batchSize expected batch size, including internal fragmentation
    +   * @return number of rows that fit into the batch
    +   */
    +
    +  private int rowsPerBatch(int batchSize) {
    +    int rowCount = batchSize * 3 / 4 / estimatedRowWidth;
    +    return Math.max(1, Math.min(rowCount, Character.MAX_VALUE));
    +  }
    +
    +  /**
    +   * Compute the expected number of rows that fit into a given size
    +   * batch, accounting for internal fragmentation due to power-of-two
    +   * rounding on vector allocations.
    +   *
    +   * @param rowCount the desired number of rows in the batch
    +   * @return the size of resulting batch, including power-of-two
    +   * rounding.
    +   */
    +
    +  private int batchForRows(int rowCount) {
    +    return estimatedRowWidth * rowCount * 4 / 3;
    +  }
    +
    +  // Must spill if we are below the spill point (the amount of memory
    +  // needed to do the minimal spill.)
    +
    +  public boolean isSpillNeeded(long allocatedBytes, int incomingSize) {
    +    return allocatedBytes + incomingSize >= bufferMemoryLimit;
    +  }
    +
    +  public boolean hasMemoryMergeCapacity(long allocatedBytes, long neededForInMemorySort)
{
    +    return (freeMemory(allocatedBytes) >= neededForInMemorySort);
    +  }
    +
    +  public long freeMemory(long allocatedBytes) {
    +    return memoryLimit - allocatedBytes;
    +  }
    +
    +  public long getMergeMemoryLimit() { return mergeMemoryLimit; }
    +  public int getSpillBatchRowCount() { return spillBatchRowCount; }
    +  public int getMergeBatchRowCount() { return mergeBatchRowCount; }
    +
    +  // Primarily for testing
    +
    --- End diff --
    
    Fixed.


> Implement sub-operator unit tests for managed external sort
> -----------------------------------------------------------
>
>                 Key: DRILL-5325
>                 URL: https://issues.apache.org/jira/browse/DRILL-5325
>             Project: Apache Drill
>          Issue Type: Improvement
>          Components: Tools, Build & Test
>    Affects Versions: 1.11.0
>            Reporter: Paul Rogers
>            Assignee: Paul Rogers
>             Fix For: 1.11.0
>
>
> Validate the proposed sub-operator test framework, by creating low-level unit tests for
the managed version of the external sort.
> The external sort has a small number of existing tests, but those tests are quite superficial;
the "managed sort" project found many bugs. The managed sort itself was tested with ad-hoc
system-level tests created using the new "cluster fixture" framework. But, again, such tests
could not reach deep inside the sort code to exercise very specific conditions.
> As a result, we spent far too much time using QA functional tests to identify specific
code issues.
> Using the sub-opeator unit test framework, we can instead test each bit of functionality
at the unit test level.
> If doing so works, and is practical, it can serve as a model for other operator testing
projects.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Mime
View raw message