[
https://issues.apache.org/jira/browse/DRILL-6032?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16497242#comment-16497242
]
ASF GitHub Bot commented on DRILL-6032:
---------------------------------------
ilooner closed pull request #1101: DRILL-6032: Made the batch sizing for
HashAgg more accurate.
URL: https://github.com/apache/drill/pull/1101
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
index 47f1017b34..374dcdd3a4 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
@@ -25,7 +25,6 @@
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.common.expression.ErrorCollector;
import org.apache.drill.common.expression.ErrorCollectorImpl;
-import org.apache.drill.common.expression.FunctionHolderExpression;
import org.apache.drill.common.expression.IfExpression;
import org.apache.drill.common.expression.LogicalExpression;
import org.apache.drill.common.logical.data.NamedExpression;
@@ -65,7 +64,7 @@
private HashAggregator aggregator;
private RecordBatch incoming;
- private LogicalExpression[] aggrExprs;
+ private ValueVectorWriteExpression[] aggrExprs;
private TypedFieldId[] groupByOutFieldIds;
private TypedFieldId[] aggrOutFieldIds; // field ids for the outgoing
batch
private final List<Comparator> comparators;
@@ -231,7 +230,7 @@ private HashAggregator createAggregatorInternal() throws
SchemaChangeException,
int numGroupByExprs = (popConfig.getGroupByExprs() != null) ?
popConfig.getGroupByExprs().size() : 0;
int numAggrExprs = (popConfig.getAggrExprs() != null) ?
popConfig.getAggrExprs().size() : 0;
- aggrExprs = new LogicalExpression[numAggrExprs];
+ aggrExprs = new ValueVectorWriteExpression[numAggrExprs];
groupByOutFieldIds = new TypedFieldId[numGroupByExprs];
aggrOutFieldIds = new TypedFieldId[numAggrExprs];
@@ -255,7 +254,6 @@ private HashAggregator createAggregatorInternal() throws
SchemaChangeException,
groupByOutFieldIds[i] = container.add(vv);
}
- int extraNonNullColumns = 0; // each of SUM, MAX and MIN gets an extra
bigint column
for (i = 0; i < numAggrExprs; i++) {
NamedExpression ne = popConfig.getAggrExprs().get(i);
final LogicalExpression expr =
@@ -273,15 +271,10 @@ private HashAggregator createAggregatorInternal() throws
SchemaChangeException,
continue;
}
- if ( expr instanceof FunctionHolderExpression ) {
- String funcName = ((FunctionHolderExpression) expr).getName();
- if ( funcName.equals("sum") || funcName.equals("max") ||
funcName.equals("min") ) {extraNonNullColumns++;}
- }
final MaterializedField outputField =
MaterializedField.create(ne.getRef().getAsNamePart().getName(),
expr.getMajorType());
@SuppressWarnings("resource")
ValueVector vv = TypeHelper.getNewVector(outputField,
oContext.getAllocator());
aggrOutFieldIds[i] = container.add(vv);
-
aggrExprs[i] = new ValueVectorWriteExpression(aggrOutFieldIds[i], expr,
true);
}
@@ -301,7 +294,7 @@ private HashAggregator createAggregatorInternal() throws
SchemaChangeException,
aggrExprs,
cgInner.getWorkspaceTypes(),
groupByOutFieldIds,
- this.container, extraNonNullColumns * 8 /* sizeof(BigInt) */);
+ container);
return agg;
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
index 4c54080169..1a33577120 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
@@ -17,20 +17,16 @@
*/
package org.apache.drill.exec.physical.impl.aggregate;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-import javax.inject.Named;
-
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
import org.apache.drill.common.exceptions.RetryAfterSpillException;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.common.expression.ExpressionPosition;
import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.common.expression.FunctionHolderExpression;
import org.apache.drill.common.expression.LogicalExpression;
-
+import org.apache.drill.common.map.CaseInsensitiveMap;
+import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.cache.VectorSerializer.Writer;
import org.apache.drill.exec.compile.sig.RuntimeOverridden;
@@ -38,7 +34,8 @@
import org.apache.drill.exec.exception.OutOfMemoryException;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.expr.TypeHelper;
-
+import org.apache.drill.exec.expr.ValueVectorReadExpression;
+import org.apache.drill.exec.expr.ValueVectorWriteExpression;
import org.apache.drill.exec.memory.BaseAllocator;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.FragmentContext;
@@ -51,45 +48,42 @@
import org.apache.drill.exec.physical.impl.common.HashTable;
import org.apache.drill.exec.physical.impl.common.HashTableConfig;
import org.apache.drill.exec.physical.impl.common.HashTableStats;
+import org.apache.drill.exec.physical.impl.common.HashTableTemplate;
import org.apache.drill.exec.physical.impl.common.IndexPointer;
-
import org.apache.drill.exec.record.RecordBatchSizer;
-
import org.apache.drill.exec.physical.impl.spill.SpillSet;
import org.apache.drill.exec.planner.physical.AggPrelBase;
-
+import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.MaterializedField;
-
import org.apache.drill.exec.record.RecordBatch;
-import org.apache.drill.exec.record.BatchSchema;
-
-import org.apache.drill.exec.record.VectorContainer;
-
-import org.apache.drill.exec.record.TypedFieldId;
-
import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+import org.apache.drill.exec.record.TypedFieldId;
+import org.apache.drill.exec.record.VectorContainer;
import org.apache.drill.exec.record.VectorWrapper;
import org.apache.drill.exec.record.WritableBatch;
-
import org.apache.drill.exec.vector.AllocationHelper;
-
import org.apache.drill.exec.vector.FixedWidthVector;
import org.apache.drill.exec.vector.ObjectVector;
import org.apache.drill.exec.vector.ValueVector;
-
import org.apache.drill.exec.vector.VariableWidthVector;
+import org.slf4j.Marker;
+import org.slf4j.MarkerFactory;
+
+import javax.inject.Named;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
import static org.apache.drill.exec.record.RecordBatch.MAX_BATCH_SIZE;
public abstract class HashAggTemplate implements HashAggregator {
- protected static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(HashAggregator.class);
-
- private static final int VARIABLE_MAX_WIDTH_VALUE_SIZE = 50;
- private static final int VARIABLE_MIN_WIDTH_VALUE_SIZE = 8;
+ protected static final Marker hashAggDebug1 =
MarkerFactory.getMarker("HASH_AGG_DEBUG_1");
+ protected static final Marker hashAggDebug2 =
MarkerFactory.getMarker("HASH_AGG_DEBUG_2");
+ protected static final Marker hashAggDebugSpill =
MarkerFactory.getMarker("HASH_AGG_DEBUG_SPILL");
- private static final boolean EXTRA_DEBUG_1 = false;
- private static final boolean EXTRA_DEBUG_2 = false;
- private static final boolean EXTRA_DEBUG_SPILL = false;
+ protected static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(HashAggregator.class);
// Fields needed for partitioning (the groups into partitions)
private int numPartitions = 0; // must be 2 to the power of bitsInMask (set
in setup())
@@ -112,15 +106,12 @@
private int earlyPartition = 0; // which partition to return early
private boolean retrySameIndex = false; // in case put failed during 1st
phase - need to output early, then retry
private boolean useMemoryPrediction = false; // whether to use memory
prediction to decide when to spill
- private long estMaxBatchSize = 0; // used for adjusting #partitions and
deciding when to spill
- private long estRowWidth = 0; // the size of the internal "row" (keys +
values + extra columns)
- private long estValuesRowWidth = 0; // the size of the internal values (
values + extra )
- private long estOutputRowWidth = 0; // the size of the output "row" (no
extra columns)
- private long estValuesBatchSize = 0; // used for "reserving" memory for the
Values batch to overcome an OOM
- private long estOutgoingAllocSize = 0; // used for "reserving" memory for
the Outgoing Output Values to overcome an OOM
+ private long estAccumulationBatchSize = 0; // used for adjusting #partitions
and deciding when to spill
+ private long estAccumulationBatchRowWidth = 0; // the size of the output
"row" (no extra columns)
+ private long estBatchHolderValuesBatchSize = 0; // used for "reserving"
memory for the Values batch to overcome an OOM
+ private long estOutgoingBatchValuesSize = 0; // used for "reserving" memory
for the Outgoing Output Values to overcome an OOM
private long reserveValueBatchMemory; // keep "reserve memory" for Values
Batch
private long reserveOutgoingMemory; // keep "reserve memory" for the
Outgoing (Values only) output
- private int maxColumnWidth = VARIABLE_MIN_WIDTH_VALUE_SIZE; // to control
memory allocation for varchars
private long minBatchesPerPartition; // for tuning - num partitions and
spill decision
private long plannedBatches = 0; // account for planned, but not yet
allocated batches
@@ -140,6 +131,9 @@
private OperatorContext oContext;
private BufferAllocator allocator;
+ private Map<String, Integer> keySizes;
+ // The size estimates for varchar value columns. The keys are the index of
the varchar value columns.
+ private Map<Integer, Integer> varcharValueSizes;
private HashTable htables[];
private ArrayList<BatchHolder> batchHolders[];
private int outBatchIndex[];
@@ -156,7 +150,6 @@
private static class SpilledPartition { public int spilledBatches; public
String spillFile; int cycleNum; int origPartn; int prevOrigPartn; }
private ArrayList<SpilledPartition> spilledPartitionsList;
- private int operatorId; // for the spill file name
private IndexPointer htIdxHolder; // holder for the Hashtable's internal
index returned by put()
private IndexPointer outStartIdxHolder;
@@ -165,6 +158,7 @@
private TypedFieldId[] groupByOutFieldIds;
private MaterializedField[] materializedValueFields;
+ private ValueVectorWriteExpression[] valueExprs;
private boolean allFlushed = false;
private boolean buildComplete = false;
private boolean handlingSpills = false; // True once starting to process
spill files
@@ -216,19 +210,19 @@ public BatchHolder() {
// Create a type-specific ValueVector for this value
vector = TypeHelper.getNewVector(outputField, allocator);
- // Try to allocate space to store BATCH_SIZE records. Key stored at
index i in HashTable has its workspace
+ // Try to allocate space to store MAX_BATCH_SIZE records. Key stored
at index i in HashTable has its workspace
// variables (such as count, sum etc) stored at index i in HashAgg.
HashTable and HashAgg both have
// BatchHolders. Whenever a BatchHolder in HashAgg reaches its
capacity, a new BatchHolder is added to
- // HashTable. If HashAgg can't store BATCH_SIZE records in a
BatchHolder, it leaves empty slots in current
+ // HashTable. If HashAgg can't store MAX_BATCH_SIZE records in a
BatchHolder, it leaves empty slots in current
// BatchHolder in HashTable, causing the HashTable to be space
inefficient. So it is better to allocate space
- // to fit as close to as BATCH_SIZE records.
+ // to fit as close to as MAX_BATCH_SIZE records.
if (vector instanceof FixedWidthVector) {
- ((FixedWidthVector) vector).allocateNew(HashTable.BATCH_SIZE);
+ ((FixedWidthVector) vector).allocateNew(MAX_BATCH_SIZE);
} else if (vector instanceof VariableWidthVector) {
- // This case is never used .... a varchar falls under ObjectVector
which is allocated on the heap !
- ((VariableWidthVector) vector).allocateNew(maxColumnWidth,
HashTable.BATCH_SIZE);
+ // TODO once varchars are no longer stored on heap while
aggregating we will also have to account for their sizes here
+ throw new IllegalStateException();
} else if (vector instanceof ObjectVector) {
- ((ObjectVector) vector).allocateNew(HashTable.BATCH_SIZE);
+ ((ObjectVector) vector).allocateNew(MAX_BATCH_SIZE);
} else {
vector.allocateNew();
}
@@ -263,9 +257,7 @@ private void outputValues(IndexPointer outStartIdxHolder,
IndexPointer outNumRec
for (int i = batchOutputCount; i <= maxOccupiedIdx; i++) {
try { outputRecordValues(i, batchOutputCount); }
catch (SchemaChangeException sc) { throw new
UnsupportedOperationException(sc);}
- if (EXTRA_DEBUG_2) {
- logger.debug("Outputting values to output index: {}",
batchOutputCount);
- }
+ logger.debug(hashAggDebug2, "Outputting values to output index: {}",
batchOutputCount);
batchOutputCount++;
outNumRecordsHolder.value++;
}
@@ -300,16 +292,15 @@ public void outputRecordValues(@Named("htRowIdx") int
htRowIdx, @Named("outRowId
}
@Override
- public void setup(HashAggregate hashAggrConfig, HashTableConfig htConfig,
FragmentContext context, OperatorContext oContext,
- RecordBatch incoming, HashAggBatch outgoing,
LogicalExpression[] valueExprs, List<TypedFieldId> valueFieldIds,
- TypedFieldId[] groupByOutFieldIds, VectorContainer
outContainer, int extraRowBytes) throws SchemaChangeException, IOException {
+ public void setup(HashAggregate hashAggrConfig, HashTableConfig htConfig,
+ FragmentContext context, OperatorContext oContext,
+ RecordBatch incoming, HashAggBatch outgoing,
+ ValueVectorWriteExpression[] valueExprs,
List<TypedFieldId> valueFieldIds,
+ TypedFieldId[] groupByOutFieldIds, VectorContainer
outContainer) throws SchemaChangeException, IOException {
- if (valueExprs == null || valueFieldIds == null) {
- throw new IllegalArgumentException("Invalid aggr value exprs or
workspace variables.");
- }
- if (valueFieldIds.size() < valueExprs.length) {
- throw new IllegalArgumentException("Wrong number of workspace
variables.");
- }
+ Preconditions.checkNotNull(valueFieldIds);
+ Preconditions.checkNotNull(valueExprs);
+ Preconditions.checkArgument(valueFieldIds.size() >= valueExprs.length,
"Wrong number of workspace variables.");
this.context = context;
this.stats = oContext.getStats();
@@ -318,8 +309,8 @@ public void setup(HashAggregate hashAggrConfig,
HashTableConfig htConfig, Fragme
this.incoming = incoming;
this.outgoing = outgoing;
this.outContainer = outContainer;
- this.operatorId = hashAggrConfig.getOperatorId();
this.useMemoryPrediction =
context.getOptions().getOption(ExecConstants.HASHAGG_USE_MEMORY_PREDICTION_VALIDATOR);
+ this.valueExprs = valueExprs;
is2ndPhase = hashAggrConfig.getAggPhase() ==
AggPrelBase.OperatorPhase.PHASE_2of2;
isTwoPhase = hashAggrConfig.getAggPhase() !=
AggPrelBase.OperatorPhase.PHASE_1of1;
@@ -374,10 +365,6 @@ public void setup(HashAggregate hashAggrConfig,
HashTableConfig htConfig, Fragme
this.groupByOutFieldIds = groupByOutFieldIds; // retain these for
delayedSetup, and to allow recreating hash tables (after a spill)
numGroupByOutFields = groupByOutFieldIds.length;
- // Start calculating the row widths (with the extra columns; the rest
would be done in updateEstMaxBatchSize() )
- estRowWidth = extraRowBytes;
- estValuesRowWidth = extraRowBytes;
-
doSetup(incoming);
}
@@ -397,24 +384,22 @@ private void delayedSetup() {
}
numPartitions = BaseAllocator.nextPowerOfTwo(numPartitions); // in case
not a power of 2
- if ( schema == null ) { estValuesBatchSize = estOutgoingAllocSize =
estMaxBatchSize = 0; } // incoming was an empty batch
- else {
- // Estimate the max batch size; should use actual data (e.g. lengths of
varchars)
- updateEstMaxBatchSize(incoming);
- }
+ // Estimate the max batch size; should use actual data (e.g. lengths of
varchars)
+ updateEstMaxBatchSize(incoming);
+
// create "reserved memory" and adjust the memory limit down
- reserveValueBatchMemory = reserveOutgoingMemory = estValuesBatchSize ;
- long newMemoryLimit = allocator.getLimit() - reserveValueBatchMemory -
reserveOutgoingMemory ;
- long memAvail = newMemoryLimit - allocator.getAllocatedMemory();
- if ( memAvail <= 0 ) { throw new OutOfMemoryException("Too little memory
available"); }
- allocator.setLimit(newMemoryLimit);
+ if (!restoreReservedMemory()) {
+ throw new OutOfMemoryException("Too little memory available");
+ }
+
+ long memAvail = allocator.getLimit() - allocator.getAllocatedMemory();
if ( !canSpill ) { // single phase, or spill disabled by configuation
numPartitions = 1; // single phase should use only a single partition
(to save memory)
} else { // two phase
// Adjust down the number of partitions if needed - when the memory
available can not hold as
// many batches (configurable option), plus overhead (e.g. hash table,
links, hash values))
- while ( numPartitions * ( estMaxBatchSize * minBatchesPerPartition + 2 *
1024 * 1024) > memAvail ) {
+ while (numPartitions * (estAccumulationBatchSize *
minBatchesPerPartition) > memAvail) {
numPartitions /= 2;
if ( numPartitions < 2) {
if (is2ndPhase) {
@@ -465,7 +450,7 @@ private void delayedSetup() {
for (int i = 0; i < numPartitions; i++ ) {
try {
this.htables[i] =
baseHashTable.createAndSetupHashTable(groupByOutFieldIds, numPartitions);
- this.htables[i].setMaxVarcharSize(maxColumnWidth);
+ this.htables[i].setKeySizes(keySizes);
} catch (ClassTransformationException e) {
throw UserException.unsupportedError(e)
.message("Code generation error - likely an error in the code.")
@@ -516,46 +501,128 @@ private void initializeSetup(RecordBatch newIncoming)
throws SchemaChangeExcepti
* @param incoming
*/
private void updateEstMaxBatchSize(RecordBatch incoming) {
- if ( estMaxBatchSize > 0 ) { return; } // no handling of a schema (or
varchar) change
// Use the sizer to get the input row width and the length of the longest
varchar column
- RecordBatchSizer sizer = new RecordBatchSizer(incoming);
- logger.trace("Incoming sizer: {}",sizer);
- // An empty batch only has the schema, can not tell actual length of
varchars
- // else use the actual varchars length, each capped at 50 (to match the
space allocation)
- long estInputRowWidth = sizer.rowCount() == 0 ? sizer.stdRowWidth() :
sizer.netRowWidthCap50();
+ final RecordBatchSizer incomingColumnSizes = new
RecordBatchSizer(incoming);
+ final Map<String, RecordBatchSizer.ColumnSize> columnSizeMap =
incomingColumnSizes.columns();
+ keySizes = CaseInsensitiveMap.newHashMap();
+ varcharValueSizes = Maps.newHashMap();
+
+ logger.trace("Incoming sizer: {}",incomingColumnSizes);
+
+ for (int columnIndex = 0; columnIndex < numGroupByOutFields;
columnIndex++) {
+ final VectorWrapper vectorWrapper =
outContainer.getValueVector(columnIndex);
+ final String columnName = vectorWrapper.getField().getName();
+ final int columnSize = columnSizeMap.get(columnName).getStdOrEstSize();
+ keySizes.put(columnName, columnSize);
+ estAccumulationBatchRowWidth += columnSize;
+ }
+
+ // estBatchHolderValuesRowWidth represents the size of a row in value
BatchHolder that is used for accumulations.
+ long estBatchHolderValuesRowWidth = 0;
+ // estOutputValuesRowWidth represents the size of the value rows in the
batch that is actually output.
+ // Does not include extra phantom columns from DRILL-5728.
+ long estOutputValuesRowWidth = 0;
+
+ // TODO DRILL-5728 - This code is necessary because the generated
BatchHolders add extra BigInt columns
+ // to store a flag indicating if a row is null or not. Ideally we should
not generate these
+ // extra BigInt columns, but fixing this is a big task. Once DRILL-5728 is
fixed this code block
+ // should be deleted
+ {
+ final TypeProtos.MajorType majorType = TypeProtos.MajorType.newBuilder()
+ .setMinorType(TypeProtos.MinorType.BIGINT)
+ .build();
+ final int bigIntSize = TypeHelper.getSize(majorType);
+
+ for (ValueVectorWriteExpression valueExpr : valueExprs) {
+ final LogicalExpression childExpression = valueExpr.getChild();
+
+ if (childExpression instanceof FunctionHolderExpression) {
+ final FunctionHolderExpression funcExpression =
((FunctionHolderExpression) childExpression);
+ final String funcName = funcExpression.getName();
+
+ if (funcName.equals("sum") || funcName.equals("max") ||
funcName.equals("min")) {
+ estBatchHolderValuesRowWidth += bigIntSize;
+ estAccumulationBatchRowWidth += bigIntSize;
+ }
+ }
+ }
+ }
- // Get approx max (varchar) column width to get better memory allocation
- maxColumnWidth = Math.max(sizer.maxAvgColumnSize(),
VARIABLE_MIN_WIDTH_VALUE_SIZE);
- maxColumnWidth = Math.min(maxColumnWidth, VARIABLE_MAX_WIDTH_VALUE_SIZE);
+ // We need to store the estimated size of varchar columns that are being
aggregated so that we can estimate the out batch size
+ // TODO after varchars are no longer stored on heap we will also have to
use these estimates for the BatchHolders.
+ for (int valueIndex = 0; valueIndex < valueExprs.length; valueIndex++) {
+ final ValueVectorWriteExpression valueExpr = valueExprs[valueIndex];
+ final LogicalExpression childExpression = valueExpr.getChild();
+
+ if (childExpression instanceof FunctionHolderExpression) {
+ final FunctionHolderExpression funcExpression =
((FunctionHolderExpression) childExpression);
+
+ for (LogicalExpression logicalExpression: funcExpression.args) {
+ if (logicalExpression instanceof ValueVectorReadExpression) {
+ final ValueVectorReadExpression valueExpression =
+ (ValueVectorReadExpression) logicalExpression;
+
+ if
(valueExpression.getMajorType().getMinorType().equals(TypeProtos.MinorType.VARCHAR))
{
+ // If the argument to the aggregation function was a varchar, we
need to save the size estimate
+ // for the column
+ final VectorWrapper columnWrapper =
incoming.getValueAccessorById(null, valueExpression.getFieldId().getFieldIds());
+ final String columnName = columnWrapper.getField().getName();
+ final RecordBatchSizer.ColumnSize columnSizer =
columnSizeMap.get(columnName);
+ varcharValueSizes.put(numGroupByOutFields + valueIndex,
columnSizer.estSize);
+ break;
+ }
+ }
+ }
+ }
+ }
- //
- // Calculate the estimated max (internal) batch (i.e. Keys batch + Values
batch) size
- // (which is used to decide when to spill)
- // Also calculate the values batch size (used as a reserve to overcome an
OOM)
- //
- Iterator<VectorWrapper<?>> outgoingIter = outContainer.iterator();
- int fieldId = 0;
- while (outgoingIter.hasNext()) {
- ValueVector vv = outgoingIter.next().getValueVector();
- MaterializedField mr = vv.getField();
- int fieldSize = vv instanceof VariableWidthVector ? maxColumnWidth :
- TypeHelper.getSize(mr.getType());
- estRowWidth += fieldSize;
- estOutputRowWidth += fieldSize;
- if ( fieldId < numGroupByOutFields ) { fieldId++; }
- else { estValuesRowWidth += fieldSize; }
+ for (int columnIndex = numGroupByOutFields; columnIndex <
outContainer.getNumberOfColumns(); columnIndex++) {
+ VectorWrapper vectorWrapper = outContainer.getValueVector(columnIndex);
+ RecordBatchSizer.ColumnSize columnSize = new
RecordBatchSizer.ColumnSize(vectorWrapper.getValueVector());
+
+ if (columnSize.hasStdSize()) {
+ // If the column is fixed width we know the size, otherwise the column
is a varchar which is
+ // stored on heap in the BatchHolders so we can assume that the amount
of direct memory it consumes is 0.
+ estAccumulationBatchRowWidth += columnSize.getStdSize();
+ estBatchHolderValuesRowWidth += columnSize.getStdSize();
+ estOutputValuesRowWidth += columnSize.getStdSize();
+ } else {
+ // Include the size of varchar values which are transfered back to
direct memory when copied from the BatchHolder to
+ // the out container.
+ estOutputValuesRowWidth += varcharValueSizes.get(columnIndex);
+ }
}
- // multiply by the max number of rows in a batch to get the final
estimated max size
- estMaxBatchSize = Math.max(estRowWidth, estInputRowWidth) * MAX_BATCH_SIZE;
- // (When there are no aggr functions, use '1' as later code relies on this
size being non-zero)
- estValuesBatchSize = Math.max(estValuesRowWidth, 1) * MAX_BATCH_SIZE;
- estOutgoingAllocSize = estValuesBatchSize; // initially assume same size
- logger.trace("{} phase. Estimated internal row width: {} Values row width:
{} batch size: {} memory limit: {} max column width: {}",
-
isTwoPhase?(is2ndPhase?"2nd":"1st"):"Single",estRowWidth,estValuesRowWidth,estMaxBatchSize,allocator.getLimit(),maxColumnWidth);
+ // Hash table overhead may double what is actually required.
+ long hashTableOverhead =
RecordBatchSizer.multiplyByFactors(HashTableTemplate.ENTRY_OVERHEAD_BYTES *
MAX_BATCH_SIZE);
+
+ // This is the size of an accumulation batch. An accumulation batch
includes the keys stored in the hashtable and the values stored in the
BatchHolder.
+ // It also includes the overhead of the links, indices, and hash values
stored in the hashtable
+ estAccumulationBatchSize =
RecordBatchSizer.multiplyByFactors(estAccumulationBatchRowWidth *
MAX_BATCH_SIZE);
+ estAccumulationBatchSize += hashTableOverhead;
+ estAccumulationBatchSize = Math.max(estAccumulationBatchSize, 1);
+ // WARNING! (When there are no aggr functions, use '1' as later code
relies on this size being non-zero)
+ // Note: estBatchHolderValuesBatchSize cannot be 0 because a zero value
for estBatchHolderValuesBatchSize will cause reserveValueBatchMemory to have a
value of 0. And
+ // a reserveValueBatchMemory value of 0 has multiple meanings in different
contexts. So estBatchHolderValuesBatchSize has an enforced minimum value of 1,
without this
+ // estBatchHolderValuesBatchSize could have a value of 0 in the case were
there are no value columns and all the columns are key columns.
+
+ // This includes the values stored in the BatchHolder and the has table
overhead
+ estBatchHolderValuesBatchSize =
RecordBatchSizer.multiplyByFactors(estBatchHolderValuesRowWidth *
MAX_BATCH_SIZE);
+ estBatchHolderValuesBatchSize += hashTableOverhead;
+ estBatchHolderValuesBatchSize = Math.max(estBatchHolderValuesBatchSize, 1);
+ // This includes the values stored in the outContainer. Note the phantom
columns are not included in this size
+ // because the out container does not have phantom columns, only the
BatchHolder does. Also note that we are not
+ // including the size of the key columns in this estimate because the key
columns stored in the hash table are transferred
+ // to the out container, and the size of the key columns are already
accounted for in estAccumulationBatchSize
+ estOutgoingBatchValuesSize =
Math.max(RecordBatchSizer.multiplyByFactors(estOutputValuesRowWidth *
MAX_BATCH_SIZE), 1);;
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("{} phase. Estimated internal row width: {} Values row
width: {} batch size: {} memory limit: {}",
+ isTwoPhase ? (is2ndPhase ? "2nd" : "1st") : "Single",
estAccumulationBatchRowWidth, estBatchHolderValuesRowWidth,
estAccumulationBatchSize, allocator.getLimit());
+ }
- if ( estMaxBatchSize > allocator.getLimit() ) {
- logger.warn("HashAggregate: Estimated max batch size {} is larger than
the memory limit {}",estMaxBatchSize,allocator.getLimit());
+ if ( estAccumulationBatchSize > allocator.getLimit() ) {
+ logger.warn("HashAggregate: Estimated max batch size {} is larger than
the memory limit {}", estAccumulationBatchSize,allocator.getLimit());
}
}
@@ -579,16 +646,11 @@ public AggOutcome doWork() {
delayedSetup();
}
- //
- // loop through existing records in this batch, aggregating the values
as necessary.
- //
- if (EXTRA_DEBUG_1) {
- logger.debug("Starting outer loop of doWork()...");
- }
+ // loop through existing records in this batch, aggregating the values
as necessary.
+ logger.debug(hashAggDebug1, "Starting outer loop of doWork()...");
+
while (underlyingIndex < currentBatchRecordCount) {
- if (EXTRA_DEBUG_2) {
- logger.debug("Doing loop with values underlying {}, current {}",
underlyingIndex, currentIndex);
- }
+ logger.debug(hashAggDebug2, "Doing loop with values underlying {},
current {}", underlyingIndex, currentIndex);
checkGroupAndAggrValues(currentIndex);
if ( retrySameIndex ) { retrySameIndex = false; } // need to retry
this row (e.g. we had an OOM)
@@ -596,15 +658,13 @@ public AggOutcome doWork() {
// If adding a group discovered a memory pressure during 1st phase,
then start
// outputing some partition downstream in order to free memory.
- if ( earlyOutput ) {
+ if (earlyOutput) {
outputCurrentBatch();
return AggOutcome.RETURN_OUTCOME;
}
}
- if (EXTRA_DEBUG_1) {
- logger.debug("Processed {} records", underlyingIndex);
- }
+ logger.debug(hashAggDebug1, "Processed {} records", underlyingIndex);
// Cleanup the previous batch since we are done processing it.
for (VectorWrapper<?> v : incoming) {
@@ -615,26 +675,14 @@ public AggOutcome doWork() {
// from one of the spill files (The spill case is handled differently
here to avoid
// collecting stats on the spilled records)
//
- long memAllocBeforeNext = allocator.getAllocatedMemory();
if ( handlingSpills ) {
outcome = incoming.next(); // get it from the SpilledRecordBatch
} else {
// Get the next RecordBatch from the incoming (i.e. upstream operator)
outcome = outgoing.next(0, incoming);
}
- long memAllocAfterNext = allocator.getAllocatedMemory();
- long incomingBatchSize = memAllocAfterNext - memAllocBeforeNext;
-
- // If incoming batch is bigger than our estimate - adjust the estimate
to match
- if ( estMaxBatchSize < incomingBatchSize) {
- logger.debug("Found a bigger next {} batch: {} , prior estimate was:
{}, mem allocated {}",handlingSpills ? "spill" : "incoming",
- incomingBatchSize, estMaxBatchSize, memAllocAfterNext);
- estMaxBatchSize = incomingBatchSize;
- }
- if (EXTRA_DEBUG_1) {
- logger.debug("Received IterOutcome of {}", outcome);
- }
+ logger.debug(hashAggDebug1, "Received IterOutcome of {}", outcome);
// Handle various results from getting the next batch
switch (outcome) {
@@ -643,9 +691,7 @@ public AggOutcome doWork() {
return AggOutcome.RETURN_OUTCOME;
case OK_NEW_SCHEMA:
- if (EXTRA_DEBUG_1) {
- logger.debug("Received new schema. Batch has {} records.",
incoming.getRecordCount());
- }
+ logger.debug(hashAggDebug1, "Received new schema. Batch has {}
records.", incoming.getRecordCount());
this.cleanup();
// TODO: new schema case needs to be handled appropriately
return AggOutcome.UPDATE_AGGREGATOR;
@@ -655,9 +701,7 @@ public AggOutcome doWork() {
resetIndex(); // initialize index (a new batch needs to be processed)
- if (EXTRA_DEBUG_1) {
- logger.debug("Continue to start processing the next batch");
- }
+ logger.debug(hashAggDebug1, "Continue to start processing the next
batch");
break;
case NONE:
@@ -711,21 +755,29 @@ private void useReservedOutgoingMemory() {
* Restore the reserve memory (both)
*
*/
- private void restoreReservedMemory() {
+ private boolean restoreReservedMemory() {
+ boolean success = true;
+
if ( 0 == reserveOutgoingMemory ) { // always restore OutputValues first
(needed for spilling)
long memAvail = allocator.getLimit() - allocator.getAllocatedMemory();
- if ( memAvail > estOutgoingAllocSize) {
- allocator.setLimit(allocator.getLimit() - estOutgoingAllocSize);
- reserveOutgoingMemory = estOutgoingAllocSize;
+ if ( memAvail > estOutgoingBatchValuesSize) {
+ allocator.setLimit(allocator.getLimit() - estOutgoingBatchValuesSize);
+ reserveOutgoingMemory = estOutgoingBatchValuesSize;
+ } else {
+ success = false;
}
}
if ( 0 == reserveValueBatchMemory ) {
long memAvail = allocator.getLimit() - allocator.getAllocatedMemory();
- if ( memAvail > estValuesBatchSize) {
- allocator.setLimit(allocator.getLimit() - estValuesBatchSize);
- reserveValueBatchMemory = estValuesBatchSize;
+ if ( memAvail > estBatchHolderValuesBatchSize) {
+ allocator.setLimit(allocator.getLimit() -
estBatchHolderValuesBatchSize);
+ reserveValueBatchMemory = estBatchHolderValuesBatchSize;
+ } else {
+ success = false;
}
}
+
+ return success;
}
/**
* Allocate space for the returned aggregate columns
@@ -733,28 +785,34 @@ private void restoreReservedMemory() {
* @param records
*/
private void allocateOutgoing(int records) {
- // Skip the keys and only allocate for outputting the workspace values
- // (keys will be output through splitAndTransfer)
- Iterator<VectorWrapper<?>> outgoingIter = outContainer.iterator();
- for (int i = 0; i < numGroupByOutFields; i++) {
- outgoingIter.next();
- }
-
// try to preempt an OOM by using the reserved memory
useReservedOutgoingMemory();
long allocatedBefore = allocator.getAllocatedMemory();
- while (outgoingIter.hasNext()) {
+ for (int columnIndex = numGroupByOutFields; columnIndex <
outContainer.getNumberOfColumns(); columnIndex++) {
+ final VectorWrapper wrapper = outContainer.getValueVector(columnIndex);
@SuppressWarnings("resource")
- ValueVector vv = outgoingIter.next().getValueVector();
+ final ValueVector vv = wrapper.getValueVector();
+
+ final RecordBatchSizer.ColumnSize columnSizer = new
RecordBatchSizer.ColumnSize(wrapper.getValueVector());
+ int columnSize;
+
+ if (columnSizer.hasStdSize()) {
+ // For fixed width vectors we know the size of each record
+ columnSize = columnSizer.getStdSize();
+ } else {
+ // For var chars we need to use the input estimate
+ columnSize = varcharValueSizes.get(columnIndex);
+ }
- AllocationHelper.allocatePrecomputedChildCount(vv, records,
maxColumnWidth, 0);
+ // TODO currently the childValCount is set to 0 since HashAgg does not
support aggregating repeated types. If we add support
+ // for repeated types we should use the elementCount estimated from the
input batch as the value count.
+ AllocationHelper.allocatePrecomputedChildCount(vv, records, columnSize,
0);
}
long memAdded = allocator.getAllocatedMemory() - allocatedBefore;
- if ( memAdded > estOutgoingAllocSize ) {
- logger.trace("Output values allocated {} but the estimate was only {}.
Adjusting ...",memAdded,estOutgoingAllocSize);
- estOutgoingAllocSize = memAdded;
+ if ( memAdded > estOutgoingBatchValuesSize) {
+ logger.trace("Output values allocated {} but the estimate was only {}.",
memAdded, estOutgoingBatchValuesSize);
}
// try to restore the reserve
restoreReservedMemory();
@@ -816,7 +874,6 @@ public void cleanup() {
if ( newIncoming != null ) { newIncoming.close(); }
spillSet.close(); // delete the spill directory(ies)
htIdxHolder = null;
- materializedValueFields = null;
outStartIdxHolder = null;
outNumRecordsHolder = null;
}
@@ -920,9 +977,7 @@ private void spillAPartition(int part) {
ArrayList<BatchHolder> currPartition = batchHolders[part];
rowsInPartition = 0;
- if ( EXTRA_DEBUG_SPILL ) {
- logger.debug("HashAggregate: Spilling partition {} current cycle {} part
size {}", part, cycleNum, currPartition.size());
- }
+ logger.debug(hashAggDebugSpill, "HashAggregate: Spilling partition {}
current cycle {} part size {}", part, cycleNum, currPartition.size());
if ( currPartition.size() == 0 ) { return; } // in case empty - nothing to
spill
@@ -956,21 +1011,8 @@ private void spillAPartition(int part) {
this.htables[part].outputKeys(currOutBatchIndex, this.outContainer,
outStartIdxHolder.value, outNumRecordsHolder.value, numPendingOutput);
// set the value count for outgoing batch value vectors
- /* int i = 0; */
for (VectorWrapper<?> v : outgoing) {
v.getValueVector().getMutator().setValueCount(numOutputRecords);
- /*
- // print out the first row to be spilled ( varchar, varchar, bigint )
- try {
- if (i++ < 2) {
- NullableVarCharVector vv = ((NullableVarCharVector)
v.getValueVector());
- logger.info("FIRST ROW = {}", vv.getAccessor().get(0));
- } else {
- NullableBigIntVector vv = ((NullableBigIntVector)
v.getValueVector());
- logger.info("FIRST ROW = {}", vv.getAccessor().get(0));
- }
- } catch (Exception e) { logger.info("While printing the first row -
Got an exception = {}",e); }
- */
}
outContainer.setRecordCount(numPendingOutput);
@@ -997,10 +1039,7 @@ private void addBatchHolder(int part) {
BatchHolder bh = newBatchHolder();
batchHolders[part].add(bh);
- if (EXTRA_DEBUG_1) {
- logger.debug("HashAggregate: Added new batch; num batches = {}.",
batchHolders[part].size());
- }
-
+ logger.debug(hashAggDebug1, "HashAggregate: Added new batch; num batches =
{}.", batchHolders[part].size());
bh.setup();
}
@@ -1103,10 +1142,8 @@ public AggIterOutcome outputCurrentBatch() {
if ( cycleNum == 4 ) { logger.warn("QUATERNARY SPILLING "); }
if ( cycleNum == 5 ) { logger.warn("QUINARY SPILLING "); }
}
- if ( EXTRA_DEBUG_SPILL ) {
- logger.debug("Start reading spilled partition {} (prev {}) from
cycle {} (with {} batches). More {} spilled partitions left.",
+ logger.debug(hashAggDebugSpill, "Start reading spilled partition {}
(prev {}) from cycle {} (with {} batches). More {} spilled partitions left.",
sp.origPartn, sp.prevOrigPartn, sp.cycleNum, sp.spilledBatches,
spilledPartitionsList.size());
- }
return AggIterOutcome.AGG_RESTART;
}
@@ -1128,11 +1165,11 @@ public AggIterOutcome outputCurrentBatch() {
currPartition.get(currOutBatchIndex).outputValues(outStartIdxHolder,
outNumRecordsHolder);
int numOutputRecords = outNumRecordsHolder.value;
- if (EXTRA_DEBUG_1) {
- logger.debug("After output values: outStartIdx = {}, outNumRecords =
{}", outStartIdxHolder.value, outNumRecordsHolder.value);
- }
+ logger.debug(hashAggDebug1, "After output values: outStartIdx = {},
outNumRecords = {}", outStartIdxHolder.value, outNumRecordsHolder.value);
+
+ this.htables[partitionToReturn].outputKeys(currOutBatchIndex,
outContainer, outStartIdxHolder.value, outNumRecordsHolder.value,
numPendingOutput);
- this.htables[partitionToReturn].outputKeys(currOutBatchIndex,
this.outContainer, outStartIdxHolder.value, outNumRecordsHolder.value,
numPendingOutput);
+ outContainer.buildSchema(BatchSchema.SelectionVectorMode.NONE);
// set the value count for outgoing batch value vectors
for (VectorWrapper<?> v : outgoing) {
@@ -1141,41 +1178,31 @@ public AggIterOutcome outputCurrentBatch() {
this.outcome = IterOutcome.OK;
- if ( EXTRA_DEBUG_SPILL && is2ndPhase ) {
- logger.debug("So far returned {} + SpilledReturned {} total {} (spilled
{})",rowsNotSpilled,rowsSpilledReturned,
- rowsNotSpilled+rowsSpilledReturned,
- rowsSpilled);
+ if (is2ndPhase) {
+ logger.debug(hashAggDebugSpill, "So far returned {} + SpilledReturned {}
total {} (spilled {})",
+ rowsNotSpilled, rowsSpilledReturned, rowsNotSpilled +
rowsSpilledReturned, rowsSpilled);
}
lastBatchOutputCount = numOutputRecords;
outBatchIndex[partitionToReturn]++;
// if just flushed the last batch in the partition
if (outBatchIndex[partitionToReturn] == currPartition.size()) {
+ logger.debug(hashAggDebugSpill, "HashAggregate: {} Flushed partition {}
with {} batches total {} rows",
+ earlyOutput ? "(Early)" : "", partitionToReturn,
outBatchIndex[partitionToReturn], rowsInPartition);
- if ( EXTRA_DEBUG_SPILL ) {
- logger.debug("HashAggregate: {} Flushed partition {} with {} batches
total {} rows",
- earlyOutput ? "(Early)" : "",
- partitionToReturn, outBatchIndex[partitionToReturn],
rowsInPartition);
- }
rowsInPartition = 0; // reset to count for the next partition
// deallocate memory used by this partition, and re-initialize
reinitPartition(partitionToReturn);
if ( earlyOutput ) {
-
- if ( EXTRA_DEBUG_SPILL ) {
- logger.debug("HASH AGG: Finished (early) re-init partition {}, mem
allocated: {}", earlyPartition, allocator.getAllocatedMemory());
- }
+ logger.debug(hashAggDebugSpill, "HASH AGG: Finished (early) re-init
partition {}, mem allocated: {}", earlyPartition,
allocator.getAllocatedMemory());
outBatchIndex[earlyPartition] = 0; // reset, for next time
earlyOutput = false ; // done with early output
}
else if ( (partitionToReturn + 1 == numPartitions) &&
spilledPartitionsList.isEmpty() ) { // last partition ?
-
allFlushed = true; // next next() call will return NONE
-
logger.trace("HashAggregate: All batches flushed.");
-
// cleanup my internal state since there is nothing more to return
this.cleanup();
}
@@ -1213,11 +1240,25 @@ private String getOOMErrorMsg(String prefix) {
errmsg = "Too little memory available to operator to facilitate
spilling.";
} else { // a bug ?
errmsg = prefix + " OOM at " + (is2ndPhase ? "Second Phase" : "First
Phase") + ". Partitions: " + numPartitions +
- ". Estimated batch size: " + estMaxBatchSize + ". values size: " +
estValuesBatchSize + ". Output alloc size: " + estOutgoingAllocSize;
+ ". Estimated batch size: " + estAccumulationBatchSize + ". values size:
" + estBatchHolderValuesBatchSize + ". Output alloc size: " +
estOutgoingBatchValuesSize;
if ( plannedBatches > 0 ) { errmsg += ". Planned batches: " +
plannedBatches; }
if ( rowsSpilled > 0 ) { errmsg += ". Rows spilled so far: " +
rowsSpilled; }
}
errmsg += " Memory limit: " + allocator.getLimit() + " so far allocated: "
+ allocator.getAllocatedMemory() + ". ";
+ errmsg += "\nBatch holders in memory:\n";
+
+ // Print the number of batches we are holding in memory for each partition
+ for (int partition = 0; partition < batchHolders.length; partition++) {
+ List<BatchHolder> partitionHolders = batchHolders[partition];
+
+ int num = 0;
+
+ if (partitionHolders != null) {
+ num = partitionHolders.size();
+ }
+
+ errmsg += "Partition " + partition + " num batches: " + num + "\n";
+ }
return errmsg;
}
@@ -1229,37 +1270,10 @@ private void checkGroupAndAggrValues(int
incomingRowIdx) {
assert incomingRowIdx >= 0;
assert ! earlyOutput;
- /** for debugging
- Object tmp = (incoming).getValueAccessorById(0,
BigIntVector.class).getValueVector();
- BigIntVector vv0 = null;
- BigIntHolder holder = null;
-
- if (tmp != null) {
- vv0 = ((BigIntVector) tmp);
- holder = new BigIntHolder();
- holder.value = vv0.getAccessor().get(incomingRowIdx) ;
- }
- */
- /*
- if ( handlingSpills && ( incomingRowIdx == 0 ) ) {
- // for debugging -- show the first row from a spilled batch
- Object tmp0 =
(incoming).getValueAccessorById(NullableVarCharVector.class,
0).getValueVector();
- Object tmp1 =
(incoming).getValueAccessorById(NullableVarCharVector.class,
1).getValueVector();
- Object tmp2 =
(incoming).getValueAccessorById(NullableBigIntVector.class, 2).getValueVector();
-
- if (tmp0 != null && tmp1 != null && tmp2 != null) {
- NullableVarCharVector vv0 = ((NullableVarCharVector) tmp0);
- NullableVarCharVector vv1 = ((NullableVarCharVector) tmp1);
- NullableBigIntVector vv2 = ((NullableBigIntVector) tmp2);
- logger.debug("The first row = {} , {} , {}",
vv0.getAccessor().get(incomingRowIdx), vv1.getAccessor().get(incomingRowIdx),
vv2.getAccessor().get(incomingRowIdx));
- }
- }
- */
// The hash code is computed once, then its lower bits are used to
determine the
// partition to use, and the higher bits determine the location in the
hash table.
int hashCode;
try {
- // htables[0].updateBatches();
hashCode = htables[0].getHashCode(incomingRowIdx);
} catch (SchemaChangeException e) {
throw new UnsupportedOperationException("Unexpected schema change", e);
@@ -1335,14 +1349,12 @@ private void checkGroupAndAggrValues(int
incomingRowIdx) {
logger.trace("MEMORY CHECK AGG: allocated now {}, added {}, total
(with HT) added {}", allocator.getAllocatedMemory(),
aggValuesAddedMem, totalAddedMem);
// resize the batch estimates if needed (e.g., varchars may take more
memory than estimated)
- if (totalAddedMem > estMaxBatchSize) {
- logger.trace("Adjusting Batch size estimate from {} to {}",
estMaxBatchSize, totalAddedMem);
- estMaxBatchSize = totalAddedMem;
+ if (totalAddedMem > estAccumulationBatchSize) {
+ logger.trace("Estimated memory needed {}, actual memory used {}",
estAccumulationBatchSize, totalAddedMem);
needToCheckIfSpillIsNeeded = true;
}
- if (aggValuesAddedMem > estValuesBatchSize) {
- logger.trace("Adjusting Values Batch size from {} to
{}",estValuesBatchSize, aggValuesAddedMem);
- estValuesBatchSize = aggValuesAddedMem;
+ if (aggValuesAddedMem > estBatchHolderValuesBatchSize) {
+ logger.trace("Estimate memory needed {}, actual memory used {}",
estBatchHolderValuesBatchSize, aggValuesAddedMem);
needToCheckIfSpillIsNeeded = true;
}
} catch (OutOfMemoryException exc) {
@@ -1388,18 +1400,16 @@ private void spillIfNeeded(int currentPartition,
boolean forceSpill) {
long maxMemoryNeeded = 0;
if ( !forceSpill ) { // need to check the memory in order to decide
// calculate the (max) new memory needed now; plan ahead for at least
MIN batches
- maxMemoryNeeded = minBatchesPerPartition * Math.max(1, plannedBatches) *
(estMaxBatchSize + MAX_BATCH_SIZE * (4 + 4 /* links + hash-values */));
+ maxMemoryNeeded = minBatchesPerPartition * Math.max(1, plannedBatches) *
estAccumulationBatchSize;
// Add the (max) size of the current hash table, in case it will double
int maxSize = 1;
for (int insp = 0; insp < numPartitions; insp++) {
maxSize = Math.max(maxSize, batchHolders[insp].size());
}
- maxMemoryNeeded += MAX_BATCH_SIZE * 2 * 2 * 4 * maxSize; // 2 - double,
2 - max when %50 full, 4 - Uint4
// log a detailed debug message explaining why a spill may be needed
logger.trace("MEMORY CHECK: Allocated mem: {}, agg phase: {}, trying to
add to partition {} with {} batches. " + "Max memory needed {}, Est batch size
{}, mem limit {}",
- allocator.getAllocatedMemory(), isTwoPhase ? (is2ndPhase ? "2ND" :
"1ST") : "Single", currentPartition, batchHolders[currentPartition].size(),
maxMemoryNeeded,
- estMaxBatchSize, allocator.getLimit());
+ allocator.getAllocatedMemory(), isTwoPhase ? (is2ndPhase ? "2ND" :
"1ST") : "Single", currentPartition, batchHolders[currentPartition].size(),
maxMemoryNeeded, estAccumulationBatchSize, allocator.getLimit());
}
//
// Spill if (forced, or) the allocated memory plus the memory needed
exceed the memory limit.
@@ -1455,10 +1465,7 @@ private void spillIfNeeded(int currentPartition, boolean
forceSpill) {
// 1st phase need to return a partition early in order to free some
memory
earlyOutput = true;
earlyPartition = victimPartition;
-
- if ( EXTRA_DEBUG_SPILL ) {
- logger.debug("picked partition {} for early output",
victimPartition);
- }
+ logger.debug(hashAggDebugSpill, "picked partition {} for early
output", victimPartition);
}
}
}
@@ -1492,7 +1499,7 @@ private void updateStats(HashTable[] htables) {
}
if ( rowsReturnedEarly > 0 ) {
stats.setLongStat(Metric.SPILL_MB, // update stats - est. total MB
returned early
- (int) Math.round( rowsReturnedEarly * estOutputRowWidth / 1024.0D /
1024.0));
+ (int) Math.round( rowsReturnedEarly * estAccumulationBatchRowWidth /
1024.0D / 1024.0));
}
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggregator.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggregator.java
index 3384e671d5..bad3ea370a 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggregator.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggregator.java
@@ -20,10 +20,10 @@
import java.io.IOException;
import java.util.List;
-import org.apache.drill.common.expression.LogicalExpression;
import org.apache.drill.exec.compile.TemplateClassDefinition;
import org.apache.drill.exec.exception.ClassTransformationException;
import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.expr.ValueVectorWriteExpression;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.physical.config.HashAggregate;
@@ -34,7 +34,6 @@
import org.apache.drill.exec.record.VectorContainer;
public interface HashAggregator {
-
TemplateClassDefinition<HashAggregator> TEMPLATE_DEFINITION =
new TemplateClassDefinition<HashAggregator>(HashAggregator.class,
HashAggTemplate.class);
@@ -46,8 +45,10 @@
// OK - batch returned, NONE - end of data, RESTART - call again
enum AggIterOutcome { AGG_OK, AGG_NONE, AGG_RESTART }
- void setup(HashAggregate hashAggrConfig, HashTableConfig htConfig,
FragmentContext context, OperatorContext oContext, RecordBatch incoming,
HashAggBatch outgoing,
- LogicalExpression[] valueExprs, List<TypedFieldId> valueFieldIds,
TypedFieldId[] keyFieldIds, VectorContainer outContainer, int extraRowBytes)
throws SchemaChangeException, IOException, ClassTransformationException;
+ void setup(HashAggregate hashAggrConfig, HashTableConfig htConfig,
FragmentContext context,
+ OperatorContext oContext, RecordBatch incoming, HashAggBatch
outgoing,
+ ValueVectorWriteExpression[] valueExprs, List<TypedFieldId>
valueFieldIds, TypedFieldId[] keyFieldIds,
+ VectorContainer outContainer) throws SchemaChangeException,
IOException, ClassTransformationException;
IterOutcome getOutcome();
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
index d28fe498e4..e72733696a 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
@@ -24,6 +24,8 @@
import org.apache.drill.exec.record.VectorContainer;
import org.apache.drill.common.exceptions.RetryAfterSpillException;
+import java.util.Map;
+
public interface HashTable {
TemplateClassDefinition<HashTable> TEMPLATE_DEFINITION =
new TemplateClassDefinition<>(HashTable.class, HashTableTemplate.class);
@@ -69,7 +71,7 @@ void setup(HashTableConfig htConfig, BufferAllocator
allocator, RecordBatch inco
void reset();
- void setMaxVarcharSize(int size);
+ void setKeySizes(Map<String, Integer> keySizes);
boolean outputKeys(int batchIdx, VectorContainer outContainer, int
outStartIndex, int numRecords, int numExpectedRecords);
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
index 272d782e5b..606196deb4 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
@@ -19,9 +19,13 @@
import java.util.ArrayList;
import java.util.Iterator;
+import java.util.Map;
import javax.inject.Named;
+import com.google.common.base.Preconditions;
+import org.apache.drill.common.map.CaseInsensitiveMap;
+import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.common.types.TypeProtos.MinorType;
import org.apache.drill.common.types.Types;
import org.apache.drill.exec.compile.sig.RuntimeOverridden;
@@ -46,6 +50,13 @@
private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(HashTable.class);
private static final boolean EXTRA_DEBUG = false;
+ // startIndices, hashValues, and links overhead
+ public static final int ENTRY_OVERHEAD_BYTES =
+
TypeHelper.getSize(TypeProtos.MajorType.newBuilder().setMinorType(MinorType.INT).build())
* 3;
+
+ // TODO remove these defaults. We should require key sizes to be specified.
+ private static final int DEFAULT_VAR_CHAR_SIZE = 8;
+
private static final int EMPTY_SLOT = -1;
// A hash 'bucket' consists of the start index to indicate start of a hash
chain
@@ -99,7 +110,7 @@
private int resizingTime = 0;
- private int maxVarcharSize = 8; // for varchar allocation
+ private Map<String, Integer> keySizes; // for varchar allocation
// This class encapsulates the links, keys and values for up to BATCH_SIZE
// *unique* records. Thus, suppose there are N incoming record batches, each
@@ -117,7 +128,6 @@
private IntVector hashValues;
private int maxOccupiedIdx = -1;
-// private int batchOutputCount = 0;
private int batchIndex = 0;
@@ -129,7 +139,8 @@ public BatchHolder(int idx) {
boolean success = false;
try {
for (VectorWrapper<?> w : htContainerOrig) {
- ValueVector vv = TypeHelper.getNewVector(w.getField(), allocator);
+ final ValueVector vv = TypeHelper.getNewVector(w.getField(),
allocator);
+ final String name = vv.getField().getName();
htContainer.add(vv); // add to container before actual allocation
(to allow clearing in case of an OOM)
// Capacity for "hashValues" and "links" vectors is BATCH_SIZE
records. It is better to allocate space for
@@ -140,9 +151,10 @@ public BatchHolder(int idx) {
if (vv instanceof FixedWidthVector) {
((FixedWidthVector) vv).allocateNew(BATCH_SIZE);
} else if (vv instanceof VariableWidthVector) {
+ final int keySize = getKeySize(name, DEFAULT_VAR_CHAR_SIZE);
long beforeMem = allocator.getAllocatedMemory();
- ((VariableWidthVector) vv).allocateNew(maxVarcharSize *
BATCH_SIZE, BATCH_SIZE);
- logger.trace("HT allocated {} for varchar of max width
{}",allocator.getAllocatedMemory() - beforeMem, maxVarcharSize);
+ ((VariableWidthVector) vv).allocateNew(keySize * BATCH_SIZE,
BATCH_SIZE);
+ logger.trace("HT allocated {} for varchar size
{}",allocator.getAllocatedMemory() - beforeMem, keySize);
} else {
vv.allocateNew();
}
@@ -159,6 +171,22 @@ public BatchHolder(int idx) {
}
}
+ private int getKeySize(final String name, final int defaultSize)
+ {
+ if (keySizes == null) {
+ return defaultSize;
+ }
+
+ final Integer size = keySizes.get(name);
+
+ if (size == null) {
+ final String message = String.format(keySizes + " Could not find size
for column %s", name);
+ throw new IllegalStateException(message);
+ }
+
+ return size;
+ }
+
private void init(IntVector links, IntVector hashValues, int size) {
for (int i = 0; i < size; i++) {
links.getMutator().set(i, EMPTY_SLOT);
@@ -323,12 +351,6 @@ private void rehash(int numbuckets, IntVector
newStartIndices, int batchStartIdx
}
private boolean outputKeys(VectorContainer outContainer, int
outStartIndex, int numRecords, int numExpectedRecords) {
-
- /** for debugging
- BigIntVector vv0 = getValueVector(0);
- BigIntHolder holder = new BigIntHolder();
- */
-
// set the value count for htContainer's value vectors before the
transfer ..
setValueCount();
@@ -352,26 +374,6 @@ private boolean outputKeys(VectorContainer outContainer,
int outStartIndex, int
}
}
-/*
- logger.debug("Attempting to output keys for batch index: {} from index
{} to maxOccupiedIndex {}.",
- this.batchIndex, 0, maxOccupiedIdx);
- for (int i = batchOutputCount; i <= maxOccupiedIdx; i++) {
- if (outputRecordKeys(i, batchOutputCount) ) {
- if (EXTRA_DEBUG) logger.debug("Outputting keys to output index: {}",
batchOutputCount) ;
-
- // debugging
- // holder.value = vv0.getAccessor().get(i);
- // if (holder.value == 100018 || holder.value == 100021) {
- // logger.debug("Outputting key = {} at index - {} to outgoing
index = {}.", holder.value, i,
- // batchOutputCount);
- // }
-
- batchOutputCount++;
- } else {
- return false;
- }
- }
- */
return true;
}
@@ -805,7 +807,12 @@ private IntVector allocMetadataVector(int size, int
initialValue) {
}
@Override
- public void setMaxVarcharSize(int size) { maxVarcharSize = size; }
+ public void setKeySizes(Map<String, Integer> keySizes) {
+ Preconditions.checkNotNull(keySizes);
+
+ this.keySizes = CaseInsensitiveMap.newHashMap();
+ this.keySizes.putAll(keySizes);
+ }
// These methods will be code-generated in the context of the outer class
protected abstract void doSetup(@Named("incomingBuild") RecordBatch
incomingBuild, @Named("incomingProbe") RecordBatch incomingProbe) throws
SchemaChangeException;
@@ -813,5 +820,4 @@ private IntVector allocMetadataVector(int size, int
initialValue) {
protected abstract int getHashBuild(@Named("incomingRowIdx") int
incomingRowIdx, @Named("seedValue") int seedValue) throws SchemaChangeException;
protected abstract int getHashProbe(@Named("incomingRowIdx") int
incomingRowIdx, @Named("seedValue") int seedValue) throws SchemaChangeException;
-
}
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchSizer.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchSizer.java
index f5c77ce55a..4f48facf89 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchSizer.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatchSizer.java
@@ -17,16 +17,17 @@
*/
package org.apache.drill.exec.record;
-import java.util.Set;
import java.util.Map;
+import java.util.Set;
import org.apache.drill.common.map.CaseInsensitiveMap;
import org.apache.drill.common.types.TypeProtos.DataMode;
import org.apache.drill.common.types.TypeProtos.MinorType;
-import org.apache.drill.exec.expr.TypeHelper;
import org.apache.drill.exec.memory.AllocationManager.BufferLedger;
import org.apache.drill.exec.memory.BaseAllocator;
+import org.apache.drill.exec.physical.impl.xsort.managed.SortMemoryManager;
import org.apache.drill.exec.record.selection.SelectionVector2;
+import org.apache.drill.exec.vector.FixedWidthVector;
import org.apache.drill.exec.vector.UInt4Vector;
import org.apache.drill.exec.vector.ValueVector;
import org.apache.drill.exec.vector.complex.AbstractMapVector;
@@ -44,6 +45,11 @@
public class RecordBatchSizer {
+ // TODO consolidate common memory estimation helpers
+ public static final double PAYLOAD_FROM_BUFFER =
SortMemoryManager.PAYLOAD_FROM_BUFFER;
+ public static final double FRAGMENTATION_FACTOR = 1.0 / PAYLOAD_FROM_BUFFER;
+ public static final double BUFFER_FROM_PAYLOAD =
SortMemoryManager.BUFFER_FROM_PAYLOAD;
+
/**
* Column size information.
*/
@@ -57,7 +63,7 @@
* columns.
*/
- public int stdSize;
+ public int stdSize = -1;
/**
* Actual average column width as determined from actual memory use. This
@@ -108,7 +114,15 @@
public final float estElementCountPerArray;
public final boolean isVariableWidth;
+ public ColumnSize(ValueVector v) {
+ this(v, "");
+ }
+
public ColumnSize(ValueVector v, String prefix) {
+ if (v instanceof FixedWidthVector) {
+ stdSize = ((FixedWidthVector)v).getValueWidth();
+ }
+
this.prefix = prefix;
valueCount = v.getAccessor().getValueCount();
metadata = v.getField();
@@ -133,19 +147,46 @@ public ColumnSize(ValueVector v, String prefix) {
// No standard size for Union type
dataSize = v.getPayloadByteCount(valueCount);
break;
+ case GENERIC_OBJECT:
+ // Object vectors do not consume direct memory so their known size and
+ // estSize are 0.
+ stdSize = 0;
+ break;
default:
dataSize = v.getPayloadByteCount(valueCount);
- try {
- stdSize = TypeHelper.getSize(metadata.getType()) * elementCount;
- } catch (Exception e) {
- // For unsupported types, just set stdSize to 0.
- stdSize = 0;
- }
}
estSize = safeDivide(dataSize, valueCount);
netSize = v.getPayloadByteCount(valueCount);
}
+ /**
+ * If we can determine the knownSize, return that. Otherwise return the
estSize.
+ * @return The knownSize or estSize.
+ */
+ public int getStdOrEstSize()
+ {
+ if (hasStdSize()) {
+ // We know the exact size of the column, return it.
+ return stdSize;
+ }
+
+ return estSize;
+ }
+
+ public boolean hasStdSize()
+ {
+ return stdSize != -1;
+ }
+
+ public int getStdSize()
+ {
+ if (!hasStdSize()) {
+ throw new IllegalStateException("Unknown size for column: " +
metadata);
+ }
+
+ return stdSize;
+ }
+
@SuppressWarnings("resource")
private int buildRepeated(ValueVector v) {
@@ -252,8 +293,6 @@ public static ColumnSize getColumn(ValueVector v, String
prefix) {
return new ColumnSize(v, prefix);
}
- public static final int MAX_VECTOR_SIZE = ValueVector.MAX_BUFFER_SIZE; // 16
MiB
-
private Map<String, ColumnSize> columnSizes =
CaseInsensitiveMap.newHashMap();
/**
@@ -281,7 +320,6 @@ public static ColumnSize getColumn(ValueVector v, String
prefix) {
* vectors are partially full; prevents overestimating row width.
*/
private int netRowWidth;
- private int netRowWidthCap50;
private boolean hasSv2;
private int sv2Size;
private int avgDensity;
@@ -371,25 +409,14 @@ public void applySv2() {
computeEstimates();
}
- /**
- * Round up (if needed) to the next power of 2 (only up to 64)
- * @param arg Number to round up (must be < 64)
- * @return power of 2 result
- */
- private int roundUpToPowerOf2(int arg) {
- if ( arg <= 2 ) { return 2; }
- if ( arg <= 4 ) { return 4; }
- if ( arg <= 8 ) { return 8; }
- if ( arg <= 16 ) { return 16; }
- if ( arg <= 32 ) { return 32; }
- return 64;
- }
-
private void measureColumn(ValueVector v, String prefix) {
ColumnSize colSize = new ColumnSize(v, prefix);
- columnSizes.put(v.getField().getName(), colSize);
- stdRowWidth += colSize.stdSize;
+
+ if (colSize.hasStdSize()) {
+ stdRowWidth += colSize.stdSize;
+ }
+
netBatchSize += colSize.dataSize;
maxSize = Math.max(maxSize, colSize.dataSize);
if (colSize.metadata.isNullable()) {
@@ -414,10 +441,8 @@ private void measureColumn(ValueVector v, String prefix) {
v.collectLedgers(ledgers);
}
+ columnSizes.put(v.getField().getName(), colSize);
netRowWidth += colSize.estSize;
- netRowWidthCap50 += ! colSize.isVariableWidth ? colSize.estSize :
- 8 /* offset vector */ +
roundUpToPowerOf2(Math.min(colSize.estSize,50));
- // above change 8 to 4 after DRILL-5446 is fixed
}
private void expandMap(AbstractMapVector mapVector, String prefix) {
@@ -454,19 +479,17 @@ public static int safeDivide(long num, long denom) {
public int netRowWidth() { return netRowWidth; }
public Map<String, ColumnSize> columns() { return columnSizes; }
- /**
- * Compute the "real" width of the row, taking into account each varchar
column size
- * (historically capped at 50, and rounded up to power of 2 to match drill
buf allocation)
- * and null marking columns.
- * @return "real" width of the row
- */
- public int netRowWidthCap50() { return netRowWidthCap50 + nullableCount; }
public long actualSize() { return accountedMemorySize; }
public boolean hasSv2() { return hasSv2; }
public int avgDensity() { return avgDensity; }
public long netSize() { return netBatchSize; }
public int maxAvgColumnSize() { return maxSize / rowCount; }
+ public static long multiplyByFactors(long size)
+ {
+ return (long) (((double) size) * FRAGMENTATION_FACTOR *
BUFFER_FROM_PAYLOAD);
+ }
+
@Override
public String toString() {
StringBuilder buf = new StringBuilder();
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/FragmentsRunner.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/FragmentsRunner.java
index 2e5f2dd06e..9213f1671e 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/FragmentsRunner.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/FragmentsRunner.java
@@ -61,7 +61,7 @@
private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(FragmentsRunner.class);
private static final ControlsInjector injector =
ControlsInjectorFactory.getInjector(FragmentsRunner.class);
- private static final long RPC_WAIT_IN_MSECS_PER_FRAGMENT = 5000;
+ private static final long RPC_WAIT_IN_MSECS_PER_FRAGMENT = 30000;
private final WorkerBee bee;
private final UserClientConnection initiatingClient;
diff --git a/exec/java-exec/src/main/resources/drill-module.conf
b/exec/java-exec/src/main/resources/drill-module.conf
index 5659c82ab0..060ee39a84 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -427,8 +427,8 @@ drill.exec.options: {
exec.enable_union_type: false,
exec.errors.verbose: false,
exec.hashagg.mem_limit: 0,
- exec.hashagg.min_batches_per_partition: 2,
- exec.hashagg.num_partitions: 32,
+ exec.hashagg.min_batches_per_partition: 1,
+ exec.hashagg.num_partitions: 16,
exec.hashagg.use_memory_prediction: true,
exec.impersonation.inbound_policies: "[]",
exec.java.compiler.exp_in_method_size: 50,
diff --git
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java
index 3c1088459c..ebb649057a 100644
---
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java
+++
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java
@@ -46,13 +46,10 @@
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
-import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
import java.io.File;
-import java.io.IOException;
import java.io.PrintWriter;
import java.nio.file.Paths;
import java.util.List;
@@ -212,7 +209,7 @@ public void testGroupByMultiFields() throws Exception {
testBuilder.baselineValues(new Object[] { (long)i, (long)0, (long)0,
(long)numOccurrances});
}
- testBuilder.go();
+ testBuilder.go();
}
@Test
diff --git
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java
index 3de5519658..cc214d6862 100644
---
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java
+++
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java
@@ -59,4 +59,9 @@ public void test8() throws Exception{
testPhysicalFromFile("agg/hashagg/q8.json");
}
+ // Test aggregating varchars
+ @Test
+ public void testQ9() throws Exception {
+ testPhysicalFromFile("agg/hashagg/q9.json");
+ }
}
diff --git
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggrSpill.java
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggrSpill.java
index f517b1d0c3..689af8871f 100644
---
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggrSpill.java
+++
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggrSpill.java
@@ -53,61 +53,63 @@
@Rule
public final BaseDirTestWatcher dirTestWatcher = new BaseDirTestWatcher();
- /**
- * A template for Hash Aggr spilling tests
- *
- * @throws Exception
- */
- private void testSpill(long maxMem, long numPartitions, long minBatches,
int maxParallel, boolean fallback ,boolean predict,
- String sql, long expectedRows, int cycle, int
fromPart, int toPart) throws Exception {
- LogFixture.LogFixtureBuilder logBuilder = LogFixture.builder()
- .toConsole()
- .logger("org.apache.drill", Level.WARN);
-
- ClusterFixtureBuilder builder = ClusterFixture.builder(dirTestWatcher)
- .sessionOption(ExecConstants.HASHAGG_MAX_MEMORY_KEY,maxMem)
-
.sessionOption(ExecConstants.HASHAGG_NUM_PARTITIONS_KEY,numPartitions)
-
.sessionOption(ExecConstants.HASHAGG_MIN_BATCHES_PER_PARTITION_KEY,minBatches)
- .configProperty(ExecConstants.SYS_STORE_PROVIDER_LOCAL_ENABLE_WRITE,
false)
- .sessionOption(PlannerSettings.FORCE_2PHASE_AGGR_KEY,true)
- .sessionOption(ExecConstants.HASHAGG_FALLBACK_ENABLED_KEY, fallback)
-
.sessionOption(ExecConstants.HASHAGG_USE_MEMORY_PREDICTION_KEY,predict)
- .maxParallelization(maxParallel)
- .saveProfiles();
- String sqlStr = sql != null ? sql : // if null then use this default
query
- "SELECT empid_s17, dept_i, branch_i, AVG(salary_i) FROM
`mock`.`employee_1200K` GROUP BY empid_s17, dept_i, branch_i";
-
- try (LogFixture logs = logBuilder.build();
- ClusterFixture cluster = builder.build();
- ClientFixture client = cluster.clientFixture()) {
- runAndDump(client, sqlStr, expectedRows, cycle, fromPart,toPart);
- }
- }
- /**
- * Test "normal" spilling: Only 2 (or 3) partitions (out of 4) would
require spilling
- * ("normal spill" means spill-cycle = 1 )
- *
- * @throws Exception
- */
- @Test
- public void testSimpleHashAggrSpill() throws Exception {
- testSpill(68_000_000, 16, 2, 2, false, true, null,
- 1_200_000, 1,2, 3
- );
- }
- /**
- * Test with "needed memory" prediction turned off
- * (i.e., do exercise code paths that catch OOMs from the Hash Table and
recover)
- *
- * @throws Exception
- */
- @Test
- public void testNoPredictHashAggrSpill() throws Exception {
- testSpill(58_000_000, 16, 2, 2, false, false /* no prediction */, null,
- 1_200_000, 1, 1, 1);
+ /**
+ * A template for Hash Aggr spilling tests
+ *
+ * @throws Exception
+ */
+ private void testSpill(long maxMem, long numPartitions, long minBatches, int
maxParallel,
+ boolean fallback, boolean predict, String sql, long
expectedRows,
+ int fromCycle, int toCycle, int fromPart, int toPart)
throws Exception {
+ LogFixture.LogFixtureBuilder logBuilder = LogFixture.builder()
+ .toConsole()
+ .logger("org.apache.drill", Level.WARN);
+
+ ClusterFixtureBuilder builder = ClusterFixture.builder(dirTestWatcher)
+ .sessionOption(ExecConstants.HASHAGG_MAX_MEMORY_KEY, maxMem)
+ .sessionOption(ExecConstants.HASHAGG_NUM_PARTITIONS_KEY, numPartitions)
+ .sessionOption(ExecConstants.HASHAGG_MIN_BATCHES_PER_PARTITION_KEY,
minBatches)
+ .configProperty(ExecConstants.SYS_STORE_PROVIDER_LOCAL_ENABLE_WRITE,
false)
+ .sessionOption(PlannerSettings.FORCE_2PHASE_AGGR_KEY, true)
+ .sessionOption(ExecConstants.HASHAGG_FALLBACK_ENABLED_KEY, fallback)
+ .sessionOption(ExecConstants.HASHAGG_USE_MEMORY_PREDICTION_KEY, predict)
+ .maxParallelization(maxParallel).saveProfiles();
+ String sqlStr = sql != null ? sql : // if null then use this default query
+ "SELECT empid_s17, dept_i, branch_i, AVG(salary_i) FROM
`mock`.`employee_1200K` GROUP BY empid_s17, dept_i, branch_i";
+
+ try (LogFixture logs = logBuilder.build();
+ ClusterFixture cluster = builder.build();
+ ClientFixture client = cluster.clientFixture()) {
+ runAndDump(client, sqlStr, expectedRows, fromCycle, toCycle, fromPart,
toPart);
}
+ }
+
+ /**
+ * Test "normal" spilling: Only 2 (or 3) partitions (out of 4) would require
spilling
+ * ("normal spill" means spill-cycle = 1 )
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testSimpleHashAggrSpill() throws Exception {
+ testSpill(80_000_000, 16, 2, 2, false,
+ true, null, 1_200_000, 1, 1, 1, 3);
+ }
+
+ /**
+ * Test with "needed memory" prediction turned off
+ * (i.e., do exercise code paths that catch OOMs from the Hash Table and
recover)
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testNoPredictHashAggrSpill() throws Exception {
+ testSpill(58_000_000, 16, 2, 2, false,
+ false /* no prediction */, null, 1_200_000, 1, 1, 1, 3);
+ }
- private void runAndDump(ClientFixture client, String sql, long expectedRows,
long spillCycle, long fromSpilledPartitions, long toSpilledPartitions) throws
Exception {
+ private void runAndDump(ClientFixture client, String sql, long expectedRows,
long fromSpillCycle, long toSpillCycle, long fromSpilledPartitions, long
toSpilledPartitions) throws
+ Exception {
QueryBuilder.QuerySummary summary = client.queryBuilder().sql(sql).run();
if (expectedRows > 0) {
assertEquals(expectedRows, summary.recordCount());
@@ -120,7 +122,7 @@ private void runAndDump(ClientFixture client, String sql,
long expectedRows, lon
// check for the first op only
ProfileParser.OperatorProfile hag0 = ops.get(0);
long opCycle =
hag0.getMetric(HashAggTemplate.Metric.SPILL_CYCLE.ordinal());
- assertEquals(spillCycle, opCycle);
+ assertTrue(fromSpillCycle <= opCycle && opCycle <= toSpillCycle);
long op_spilled_partitions =
hag0.getMetric(HashAggTemplate.Metric.SPILLED_PARTITIONS.ordinal());
assertTrue(op_spilled_partitions >= fromSpilledPartitions &&
op_spilled_partitions <= toSpilledPartitions);
}
@@ -133,8 +135,8 @@ private void runAndDump(ClientFixture client, String sql,
long expectedRows, lon
@Test
public void testHashAggrSecondaryTertiarySpill() throws Exception {
- testSpill(58_000_000, 16, 3, 1, false, true, "SELECT empid_s44, dept_i,
branch_i, AVG(salary_i) FROM `mock`.`employee_1100K` GROUP BY empid_s44,
dept_i, branch_i",
- 1_100_000, 3, 2, 2);
+ testSpill(58_000_000, 16, 2, 1, false, true, "SELECT empid_s44, dept_i,
branch_i, AVG(salary_i) FROM `mock`.`employee_1100K` GROUP BY empid_s44,
dept_i, branch_i",
+ 1_100_000, 3, 4, 2, 2);
}
/**
@@ -147,7 +149,7 @@ public void testHashAggrFailWithFallbackDisabed() throws
Exception {
try {
testSpill(34_000_000, 4, 5, 2, false /* no fallback */, true, null,
- 1_200_000, 0 /* no spill due to fallback to pre-1.11 */, 0, 0);
+ 1_200_000, 0 /* no spill due to fallback to pre-1.11 */, 0, 0, 0);
fail(); // in case the above test did not throw
} catch (Exception ex) {
assertTrue(ex instanceof UserRemoteException);
@@ -165,6 +167,6 @@ public void testHashAggrFailWithFallbackDisabed() throws
Exception {
@Test
public void testHashAggrSuccessWithFallbackEnabled() throws Exception {
testSpill(34_000_000, 4, 5, 2, true /* do fallback */, true, null,
- 1_200_000, 0 /* no spill due to fallback to pre-1.11 */, 0, 0);
+ 1_200_000, 0 /* no spill due to fallback to pre-1.11 */, 0, 0, 0);
}
}
diff --git
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinAdvanced.java
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinAdvanced.java
index b9b97c15cb..05a75b11b3 100644
---
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinAdvanced.java
+++
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinAdvanced.java
@@ -31,8 +31,6 @@
import java.io.File;
import java.io.FileWriter;
import java.nio.file.Paths;
-import java.util.concurrent.ExecutorService;
-
@Category(OperatorTest.class)
public class TestHashJoinAdvanced extends JoinTestBase {
diff --git
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestSortImpl.java
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestSortImpl.java
index a98547885f..3ccb2a0995 100644
---
a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestSortImpl.java
+++
b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestSortImpl.java
@@ -51,6 +51,7 @@
import org.apache.drill.test.rowSet.RowSetReader;
import org.apache.drill.test.rowSet.RowSetWriter;
import org.apache.drill.test.rowSet.SchemaBuilder;
+import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
diff --git
a/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java
b/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java
index c3ecaf118e..d64132f367 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java
@@ -36,7 +36,6 @@
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.drill.exec.compile.ClassBuilder;
-import org.apache.drill.exec.compile.CodeCompiler;
import org.apache.drill.test.DrillTestWrapper.TestServices;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.config.DrillProperties;
@@ -73,8 +72,6 @@
import org.apache.drill.exec.record.VectorWrapper;
import org.apache.drill.exec.vector.ValueVector;
-import org.apache.drill.test.ClusterFixture;
-import org.junit.ClassRule;
public class BaseTestQuery extends ExecTest {
private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(BaseTestQuery.class);
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q9.json
b/exec/java-exec/src/test/resources/agg/hashagg/q9.json
new file mode 100644
index 0000000000..35faed8bf8
--- /dev/null
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q9.json
@@ -0,0 +1,57 @@
+{
+ head : {
+ version : 1,
+ generator : {
+ type : "optiq",
+ info : "na"
+ },
+ type : "APACHE_DRILL_PHYSICAL"
+ },
+ graph : [ {
+ "pop" : "parquet-scan",
+ "@id" : 1,
+ "entries" : [ {
+ "path" : "tpch/nation.parquet"
+ } ],
+ "storage" : {
+ "type" : "file",
+ "connection" : "classpath:///"
+ },
+ "format" : {
+ "type" : "parquet"
+ }
+ }, {
+ pop : "project",
+ @id : 2,
+ exprs : [ {
+ ref : "$f0",
+ expr : "N_REGIONKEY"
+ }, {
+ ref : "$f1",
+ expr : "N_NAME"
+ }, {
+ ref : "$f2",
+ expr : "N_NATIONKEY"
+ } ],
+ child : 1
+ }, {
+ pop : "hash-aggregate",
+ @id : 3,
+ child : 2,
+ keys : [ {
+ ref : "$f0",
+ expr : "$f0"
+ } ],
+ exprs : [ {
+ ref : "Y",
+ expr : "sum($f2) "
+ }, {
+ ref : "Z",
+ expr : "max($f1) "
+ } ]
+ }, {
+ pop : "screen",
+ @id : 4,
+ child : 3
+ } ]
+}
diff --git a/exec/vector/src/main/codegen/templates/FixedValueVectors.java
b/exec/vector/src/main/codegen/templates/FixedValueVectors.java
index 79beb52e05..3284efe8c1 100644
--- a/exec/vector/src/main/codegen/templates/FixedValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/FixedValueVectors.java
@@ -298,6 +298,11 @@ public int getPayloadByteCount(int valueCount) {
return valueCount * ${type.width};
}
+ @Override
+ public int getValueWidth() {
+ return ${type.width};
+ }
+
private class TransferImpl implements TransferPair {
private ${minor.class}Vector to;
diff --git a/exec/vector/src/main/codegen/templates/NullableValueVectors.java
b/exec/vector/src/main/codegen/templates/NullableValueVectors.java
index bc13d62a2f..2c59ec7ba2 100644
--- a/exec/vector/src/main/codegen/templates/NullableValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/NullableValueVectors.java
@@ -221,6 +221,13 @@ public int getPayloadByteCount(int valueCount) {
return bits.getPayloadByteCount(valueCount) +
values.getPayloadByteCount(valueCount);
}
+ <#if type.major != "VarLen">
+ @Override
+ public int getValueWidth(){
+ return bits.getValueWidth() + ${type.width};
+ }
+ </#if>
+
<#if type.major == "VarLen">
@Override
public void allocateNew(int totalBytes, int valueCount) {
diff --git
a/exec/vector/src/main/java/org/apache/drill/exec/vector/BitVector.java
b/exec/vector/src/main/java/org/apache/drill/exec/vector/BitVector.java
index ca2be3a769..247355690d 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/BitVector.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/BitVector.java
@@ -211,6 +211,11 @@ public void zeroVector() {
data.setZero(0, data.capacity());
}
+ @Override
+ public int getValueWidth() {
+ return VALUE_WIDTH;
+ }
+
public void copyFrom(int inIndex, int outIndex, BitVector from) {
this.mutator.set(outIndex, from.accessor.get(inIndex));
}
diff --git
a/exec/vector/src/main/java/org/apache/drill/exec/vector/FixedWidthVector.java
b/exec/vector/src/main/java/org/apache/drill/exec/vector/FixedWidthVector.java
index 09bcdd8134..e151e30c81 100644
---
a/exec/vector/src/main/java/org/apache/drill/exec/vector/FixedWidthVector.java
+++
b/exec/vector/src/main/java/org/apache/drill/exec/vector/FixedWidthVector.java
@@ -30,4 +30,10 @@
* Zero out the underlying buffer backing this vector.
*/
void zeroVector();
+
+ /**
+ * The width of a record in bytes.
+ * @return The width of a record in bytes.
+ */
+ int getValueWidth();
}
diff --git
a/exec/vector/src/main/java/org/apache/drill/exec/vector/UntypedNullVector.java
b/exec/vector/src/main/java/org/apache/drill/exec/vector/UntypedNullVector.java
index 5565fa41a6..c46e3b854e 100644
---
a/exec/vector/src/main/java/org/apache/drill/exec/vector/UntypedNullVector.java
+++
b/exec/vector/src/main/java/org/apache/drill/exec/vector/UntypedNullVector.java
@@ -91,6 +91,11 @@ public DrillBuf reallocRaw(int newAllocationSize) {
throw new UnsupportedOperationException();
}
+ @Override
+ public int getValueWidth() {
+ return VALUE_WIDTH;
+ }
+
@Override
public void load(SerializedField metadata, DrillBuf buffer) {
Preconditions.checkArgument(this.field.getName().equals(metadata.getNamePart().getName()),
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Use RecordBatchSizer to estimate size of columns in HashAgg
> -----------------------------------------------------------
>
> Key: DRILL-6032
> URL: https://issues.apache.org/jira/browse/DRILL-6032
> Project: Apache Drill
> Issue Type: Improvement
> Reporter: Timothy Farkas
> Assignee: Timothy Farkas
> Priority: Major
> Fix For: 1.15.0
>
>
> We need to use the RecordBatchSize to estimate the size of columns in the
> Partition batches created by HashAgg.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
