You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by pr...@apache.org on 2017/10/01 01:14:19 UTC
drill git commit: DRILL-5443: Rollup of external sort fixes
Repository: drill
Updated Branches:
refs/heads/master 5df49ab9c -> e2a28f7f1
DRILL-5443: Rollup of external sort fixes
- DRILL-5758: the “record batch sizer” did not handle repeated columns correctly.
- Enabled managed sort by default
- Fix check style warning
- Fix for DRILL-5670
Estimation for size of spill batch read from disk was off. For some
reason, Drill needs an amount of memory 2x the data size. The previous
estimate was 1.5x. That error, accumulated over 47 columns, was enough
to cause an OOM.
- Code cleanup discovered during the investigation.
- Exception if reAlloc tries to double a zero-size vector
- DRILL-5804: Fixes issues with zero-length vector allocations.
- Better estimates array cardinality when it is fractional.
- Uses fractional cardinality to allocate new arrays.
- Prevents an infinite loop on reAlloc if the array starts empty.
- Fixed unit test issue
- Change batch size variables from int to long
closes #932
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/e2a28f7f
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/e2a28f7f
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/e2a28f7f
Branch: refs/heads/master
Commit: e2a28f7f1b329d20f19f14c165362af4d716539e
Parents: 5df49ab
Author: Paul Rogers <pr...@maprtech.com>
Authored: Thu Aug 31 17:24:43 2017 -0700
Committer: Paul Rogers <pr...@maprtech.com>
Committed: Sat Sep 30 18:00:43 2017 -0700
----------------------------------------------------------------------
.../physical/impl/spill/RecordBatchSizer.java | 143 ++++++++++---------
.../physical/impl/xsort/managed/BatchGroup.java | 6 +-
.../impl/xsort/managed/BufferedBatches.java | 4 +-
.../impl/xsort/managed/MergeSortWrapper.java | 1 -
.../managed/PriorityQueueCopierWrapper.java | 6 +-
.../physical/impl/xsort/managed/SortImpl.java | 7 +-
.../impl/xsort/managed/SortMemoryManager.java | 6 +-
.../impl/xsort/managed/SortMetrics.java | 2 +-
.../drill/exec/record/VectorInitializer.java | 25 +++-
.../src/main/resources/drill-module.conf | 3 -
.../impl/xsort/managed/SortTestUtilities.java | 1 -
.../managed/TestExternalSortInternals.java | 6 +-
.../impl/xsort/managed/TestShortArrays.java | 116 +++++++++++++++
.../drill/test/rowSet/AbstractRowSet.java | 2 +-
.../drill/test/rowSet/AbstractSingleRowSet.java | 2 +-
.../drill/test/rowSet/IndirectRowSet.java | 2 +-
.../org/apache/drill/test/rowSet/RowSet.java | 2 +-
.../codegen/templates/FixedValueVectors.java | 21 ++-
.../codegen/templates/NullableValueVectors.java | 1 +
.../codegen/templates/RepeatedValueVectors.java | 3 +
.../drill/exec/vector/AllocationHelper.java | 21 +++
21 files changed, 277 insertions(+), 103 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java
index 3e021ca..5808f45 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java
@@ -59,10 +59,10 @@ public class RecordBatchSizer {
/**
* Assumed size from Drill metadata. Note that this information is
- * 100% bogus. Do not use it.
+ * 100% bogus for variable-width columns. Do not use it for such
+ * columns.
*/
- @Deprecated
public int stdSize;
/**
@@ -74,42 +74,39 @@ public class RecordBatchSizer {
public final int estSize;
/**
- * Number of times the value here (possibly repeated) appears in
- * the record batch.
+ * Number of occurrences of the value in the batch. This is trivial
+ * for top-level scalars: it is the record count. For a top-level
+ * repeated vector, this is the number of arrays, also the record
+ * count. For a value nested inside a repeated map, it is the
+ * total number of values across all maps, and may be less than,
+ * greater than (but unlikely) same as the row count.
*/
public final int valueCount;
/**
- * The number of elements in the value vector. Consider two cases.
- * A required or nullable vector has one element per row, so the
- * <tt>entryCount</tt> is the same as the <tt>valueCount</tt> (which,
- * in turn, is the same as the row count.) But, if this vector is an
- * array, then the <tt>valueCount</tt> is the number of columns, while
- * <tt>entryCount</tt> is the total number of elements in all the arrays
- * that make up the columns, so <tt>entryCount</tt> will be different than
- * the <tt>valueCount</tt> (normally larger, but possibly smaller if most
- * arrays are empty.
- * <p>
- * Finally, the column may be part of another list. In this case, the above
- * logic still applies, but the <tt>valueCount</tt> is the number of entries
- * in the outer array, not the row count.
+ * Total number of elements for a repeated type, or 1 if this is
+ * a non-repeated type. That is, a batch of 100 rows may have an
+ * array with 10 elements per row. In this case, the element count
+ * is 1000.
*/
- public int entryCount;
+ public final int elementCount;
+
public int dataSize;
/**
- * The estimated, average number of elements. For a repeated type,
+ * The estimated, average number of elements per parent value.
+ * Always 1 for a non-repeated type. For a repeated type,
* this is the average entries per array (per repeated element).
*/
- public int estElementCount;
+ public final float estElementCountPerArray;
public final boolean isVariableWidth;
- public ColumnSize(ValueVector v, String prefix, int valueCount) {
+ public ColumnSize(ValueVector v, String prefix) {
this.prefix = prefix;
- this.valueCount = valueCount;
+ valueCount = v.getAccessor().getValueCount();
metadata = v.getField();
isVariableWidth = v instanceof VariableWidthVector;
@@ -117,10 +114,12 @@ public class RecordBatchSizer {
// data stored in the vectors.
if (v.getField().getDataMode() == DataMode.REPEATED) {
- buildRepeated(v);
+ elementCount = buildRepeated(v);
+ estElementCountPerArray = valueCount == 0 ? 0 : elementCount * 1.0f / valueCount;
+ } else {
+ elementCount = 1;
+ estElementCountPerArray = 1;
}
- estElementCount = 1;
- entryCount = 1;
switch (metadata.getType().getMinorType()) {
case LIST:
buildList(v);
@@ -132,19 +131,19 @@ public class RecordBatchSizer {
break;
default:
dataSize = v.getPayloadByteCount(valueCount);
- stdSize = TypeHelper.getSize(metadata.getType());
+ stdSize = TypeHelper.getSize(metadata.getType()) * elementCount;
}
- estSize = roundUp(dataSize, valueCount);
+ estSize = safeDivide(dataSize, valueCount);
}
- private void buildRepeated(ValueVector v) {
+ @SuppressWarnings("resource")
+ private int buildRepeated(ValueVector v) {
// Repeated vectors are special: they have an associated offset vector
// that changes the value count of the contained vectors.
- @SuppressWarnings("resource")
UInt4Vector offsetVector = ((RepeatedValueVector) v).getOffsetVector();
- buildArray(offsetVector);
+ int childCount = offsetVector.getAccessor().get(valueCount);
if (metadata.getType().getMinorType() == MinorType.MAP) {
// For map, the only data associated with the map vector
@@ -152,20 +151,15 @@ public class RecordBatchSizer {
dataSize = offsetVector.getPayloadByteCount(valueCount);
}
+ return childCount;
}
private void buildList(ValueVector v) {
@SuppressWarnings("resource")
UInt4Vector offsetVector = ((RepeatedListVector) v).getOffsetVector();
- buildArray(offsetVector);
dataSize = offsetVector.getPayloadByteCount(valueCount);
}
- private void buildArray(UInt4Vector offsetVector) {
- entryCount = offsetVector.getAccessor().get(valueCount);
- estElementCount = roundUp(entryCount, valueCount);
- }
-
@Override
public String toString() {
StringBuilder buf = new StringBuilder()
@@ -178,10 +172,10 @@ public class RecordBatchSizer {
.append(", count: ")
.append(valueCount);
if (metadata.getDataMode() == DataMode.REPEATED) {
- buf.append(", total entries: ")
- .append(entryCount)
+ buf.append(", elements: ")
+ .append(elementCount)
.append(", per-array: ")
- .append(estElementCount);
+ .append(estElementCountPerArray);
}
buf .append(", std size: ")
.append(stdSize)
@@ -196,7 +190,9 @@ public class RecordBatchSizer {
/**
* Add a single vector initializer to a collection for the entire batch.
* Uses the observed column size information to predict the size needed
- * when allocating a new vector for the same data.
+ * when allocating a new vector for the same data. Adds a hint only for
+ * variable-width or repeated types; no extra information is needed for
+ * fixed width, non-repeated columns.
*
* @param initializer the vector initializer to hold the hints
* for this column
@@ -225,9 +221,9 @@ public class RecordBatchSizer {
if (width > 0) {
// Estimated width is width of entire column. Divide
// by element count to get per-element size.
- initializer.variableWidthArray(name, width / estElementCount, estElementCount);
+ initializer.variableWidthArray(name, width / estElementCountPerArray, estElementCountPerArray);
} else {
- initializer.fixedWidthArray(name, estElementCount);
+ initializer.fixedWidthArray(name, estElementCountPerArray);
}
}
else if (width > 0) {
@@ -236,6 +232,8 @@ public class RecordBatchSizer {
}
}
+ public static final int MAX_VECTOR_SIZE = ValueVector.MAX_BUFFER_SIZE; // 16 MiB
+
private List<ColumnSize> columnSizes = new ArrayList<>();
/**
@@ -252,7 +250,7 @@ public class RecordBatchSizer {
* Actual batch size summing all buffers used to store data
* for the batch.
*/
- private int accountedMemorySize;
+ private long accountedMemorySize;
/**
* Actual row width computed by dividing total batch memory by the
* record count.
@@ -270,23 +268,26 @@ public class RecordBatchSizer {
private Set<BufferLedger> ledgers = Sets.newIdentityHashSet();
- private int netBatchSize;
-
- public RecordBatchSizer(RecordBatch batch) {
- this(batch,
- (batch.getSchema().getSelectionVectorMode() == BatchSchema.SelectionVectorMode.TWO_BYTE) ?
- batch.getSelectionVector2() : null);
- }
+ private long netBatchSize;
/**
* Maximum width of a column; used for memory estimation in case of Varchars
*/
+
public int maxSize;
+
/**
* Count the nullable columns; used for memory estimation
*/
+
public int nullableCount;
+
+ public RecordBatchSizer(RecordBatch batch) {
+ this(batch,
+ (batch.getSchema().getSelectionVectorMode() == BatchSchema.SelectionVectorMode.TWO_BYTE) ?
+ batch.getSelectionVector2() : null);
+ }
/**
* Create empirical metadata for a record batch given a vector accessible
* (basically, an iterator over the vectors in the batch.)
@@ -303,7 +304,7 @@ public class RecordBatchSizer {
* (basically, an iterator over the vectors in the batch) along with a
* selection vector for those records. The selection vector is used to
* pad the estimated row width with the extra two bytes needed per record.
- * The selection vector memory is added ot the total memory consumed by
+ * The selection vector memory is added to the total memory consumed by
* this batch.
*
* @param va iterator over the batch's vectors
@@ -313,7 +314,7 @@ public class RecordBatchSizer {
public RecordBatchSizer(VectorAccessible va, SelectionVector2 sv2) {
rowCount = va.getRecordCount();
for (VectorWrapper<?> vw : va) {
- measureColumn(vw.getValueVector(), "", rowCount);
+ measureColumn(vw.getValueVector(), "");
}
for (BufferLedger ledger : ledgers) {
@@ -321,7 +322,7 @@ public class RecordBatchSizer {
}
if (rowCount > 0) {
- grossRowWidth = roundUp(accountedMemorySize, rowCount);
+ grossRowWidth = safeDivide(accountedMemorySize, rowCount);
}
if (sv2 != null) {
@@ -334,9 +335,9 @@ public class RecordBatchSizer {
}
private void computeEstimates() {
- grossRowWidth = roundUp(accountedMemorySize, rowCount);
- netRowWidth = roundUp(netBatchSize, rowCount);
- avgDensity = roundUp(netBatchSize * 100, accountedMemorySize);
+ grossRowWidth = safeDivide(accountedMemorySize, rowCount);
+ netRowWidth = safeDivide(netBatchSize, rowCount);
+ avgDensity = safeDivide(netBatchSize * 100L, accountedMemorySize);
}
public void applySv2() {
@@ -364,9 +365,9 @@ public class RecordBatchSizer {
return 64;
}
- private void measureColumn(ValueVector v, String prefix, int valueCount) {
+ private void measureColumn(ValueVector v, String prefix) {
- ColumnSize colSize = new ColumnSize(v, prefix, valueCount);
+ ColumnSize colSize = new ColumnSize(v, prefix);
columnSizes.add(colSize);
stdRowWidth += colSize.stdSize;
netBatchSize += colSize.dataSize;
@@ -380,10 +381,10 @@ public class RecordBatchSizer {
switch (v.getField().getType().getMinorType()) {
case MAP:
- expandMap((AbstractMapVector) v, prefix + v.getField().getName() + ".", colSize.entryCount);
+ expandMap((AbstractMapVector) v, prefix + v.getField().getName() + ".");
break;
case LIST:
- expandList((RepeatedListVector) v, prefix + v.getField().getName() + ".", colSize.entryCount);
+ expandList((RepeatedListVector) v, prefix + v.getField().getName() + ".");
break;
default:
v.collectLedgers(ledgers);
@@ -395,9 +396,9 @@ public class RecordBatchSizer {
// above change 8 to 4 after DRILL-5446 is fixed
}
- private void expandMap(AbstractMapVector mapVector, String prefix, int valueCount) {
+ private void expandMap(AbstractMapVector mapVector, String prefix) {
for (ValueVector vector : mapVector) {
- measureColumn(vector, prefix, valueCount);
+ measureColumn(vector, prefix);
}
// For a repeated map, we need the memory for the offset vector (only).
@@ -408,15 +409,15 @@ public class RecordBatchSizer {
}
}
- private void expandList(RepeatedListVector vector, String prefix, int valueCount) {
- measureColumn(vector.getDataVector(), prefix, valueCount);
+ private void expandList(RepeatedListVector vector, String prefix) {
+ measureColumn(vector.getDataVector(), prefix);
// Determine memory for the offset vector (only).
vector.collectLedgers(ledgers);
}
- public static int roundUp(int num, int denom) {
+ public static int safeDivide(long num, long denom) {
if (denom == 0) {
return 0;
}
@@ -427,6 +428,8 @@ public class RecordBatchSizer {
public int stdRowWidth() { return stdRowWidth; }
public int grossRowWidth() { return grossRowWidth; }
public int netRowWidth() { return netRowWidth; }
+ public List<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)
@@ -434,14 +437,12 @@ public class RecordBatchSizer {
* @return "real" width of the row
*/
public int netRowWidthCap50() { return netRowWidthCap50 + nullableCount; }
- public int actualSize() { return accountedMemorySize; }
+ public long actualSize() { return accountedMemorySize; }
public boolean hasSv2() { return hasSv2; }
public int avgDensity() { return avgDensity; }
- public int netSize() { return netBatchSize; }
+ public long netSize() { return netBatchSize; }
public int maxAvgColumnSize() { return maxSize / rowCount; }
- public static final int MAX_VECTOR_SIZE = 16 * 1024 * 1024; // 16 MiB
-
@Override
public String toString() {
StringBuilder buf = new StringBuilder();
@@ -463,7 +464,7 @@ public class RecordBatchSizer {
buf.append(netRowWidth);
buf.append(", Density: ");
buf.append(avgDensity);
- buf.append("}");
+ buf.append("%}");
return buf.toString();
}
@@ -472,7 +473,7 @@ public class RecordBatchSizer {
* schema metadata. Use that metadata to create an instance of a class that
* allocates memory for new vectors based on the observed size information.
* The caller provides the row count; the size information here provides
- * column widths, number of elements in each array, etc.
+ * column widths and the number of elements in each array.
*/
public VectorInitializer buildVectorInitializer() {
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java
index f16bec6..d902e0d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java
@@ -75,9 +75,9 @@ public abstract class BatchGroup implements VectorAccessible, AutoCloseable {
public static class InputBatch extends BatchGroup {
private final SelectionVector2 sv2;
- private final int dataSize;
+ private final long dataSize;
- public InputBatch(VectorContainer container, SelectionVector2 sv2, BufferAllocator allocator, int dataSize) {
+ public InputBatch(VectorContainer container, SelectionVector2 sv2, BufferAllocator allocator, long dataSize) {
super(container, allocator);
this.sv2 = sv2;
this.dataSize = dataSize;
@@ -85,7 +85,7 @@ public abstract class BatchGroup implements VectorAccessible, AutoCloseable {
public SelectionVector2 getSv2() { return sv2; }
- public int getDataSize() { return dataSize; }
+ public long getDataSize() { return dataSize; }
@Override
public int getRecordCount() {
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BufferedBatches.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BufferedBatches.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BufferedBatches.java
index 5bf8073..c930877 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BufferedBatches.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BufferedBatches.java
@@ -79,7 +79,7 @@ public class BufferedBatches {
public int size() { return bufferedBatches.size(); }
@SuppressWarnings("resource")
- public void add(VectorAccessible incoming, int batchSize) {
+ public void add(VectorAccessible incoming, long batchSize) {
// Convert the incoming batch to the agreed-upon schema.
// No converted batch means we got an empty input batch.
// Converting the batch transfers memory ownership to our
@@ -165,7 +165,7 @@ public class BufferedBatches {
}
@SuppressWarnings("resource")
- private void bufferBatch(VectorContainer convertedBatch, SelectionVector2 sv2, int netSize) {
+ private void bufferBatch(VectorContainer convertedBatch, SelectionVector2 sv2, long netSize) {
BufferAllocator allocator = context.getAllocator();
RecordBatchData rbd = new RecordBatchData(convertedBatch, allocator);
try {
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSortWrapper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSortWrapper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSortWrapper.java
index 9fd995f..01135f0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSortWrapper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSortWrapper.java
@@ -37,7 +37,6 @@ import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
import org.apache.drill.exec.physical.impl.sort.SortRecordBatchBuilder;
import org.apache.drill.exec.physical.impl.xsort.managed.SortImpl.SortResults;
-import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.apache.drill.exec.record.VectorContainer;
import org.apache.drill.exec.record.selection.SelectionVector2;
import org.apache.drill.exec.record.selection.SelectionVector4;
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierWrapper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierWrapper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierWrapper.java
index 2d82cc1..88686e5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierWrapper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierWrapper.java
@@ -140,6 +140,7 @@ public class PriorityQueueCopierWrapper extends BaseSortWrapper {
.message("Unexpected schema change - likely code error.")
.build(logger);
}
+ logger.debug("Copier setup complete");
}
public BufferAllocator getAllocator() { return context.getAllocator(); }
@@ -261,8 +262,9 @@ public class PriorityQueueCopierWrapper extends BaseSortWrapper {
} else {
allocHelper.allocateBatch(outputContainer, targetRecordCount);
}
- logger.trace("Initial output batch allocation: {} bytes",
- holder.getAllocator().getAllocatedMemory() - start);
+ logger.trace("Initial output batch allocation: {} bytes, {} records",
+ holder.getAllocator().getAllocatedMemory() - start,
+ targetRecordCount);
Stopwatch w = Stopwatch.createStarted();
int count = holder.copier.next(targetRecordCount);
if (count > 0) {
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortImpl.java
index 7420402..96861ab 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortImpl.java
@@ -310,7 +310,7 @@ public class SortImpl {
* @return true if spilling is needed (and possible), false otherwise
*/
- private boolean isSpillNeeded(int incomingSize) {
+ private boolean isSpillNeeded(long incomingSize) {
if (bufferedBatches.size() >= config.getBufferedBatchLimit()) {
return true;
@@ -407,6 +407,9 @@ public class SortImpl {
* @return results iterator over the single input batch
*/
+ // Disabled temporarily
+
+ @SuppressWarnings("unused")
private SortResults singleBatchResult() {
List<InputBatch> batches = bufferedBatches.removeAll();
return new SingleBatchResults(batches.get(0), outputBatch);
@@ -499,9 +502,11 @@ public class SortImpl {
spilledRuns.size());
switch (task.action) {
case SPILL:
+ logger.debug("Consolidate: spill");
spillFromMemory();
break;
case MERGE:
+ logger.debug("Consolidate: merge {} batches", task.count);
mergeRuns(task.count);
break;
case NONE:
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMemoryManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMemoryManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMemoryManager.java
index cd03b70..6c7ce20 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMemoryManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMemoryManager.java
@@ -680,9 +680,11 @@ public class SortMemoryManager {
// spilled run.
// Maximum spill batches that fit into available memory.
+ // Use the maximum buffer size since spill batches seem to
+ // be read with almost 50% internal fragmentation.
int memMergeLimit = (int) ((mergeMemoryLimit - allocMemory) /
- spillBatchSize.expectedBufferSize);
+ spillBatchSize.maxBufferSize);
memMergeLimit = Math.max(0, memMergeLimit);
// If batches are in memory, and we need more memory to merge
@@ -751,7 +753,7 @@ public class SortMemoryManager {
// 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) {
+ public boolean isSpillNeeded(long allocatedBytes, long incomingSize) {
return allocatedBytes + incomingSize >= bufferMemoryLimit;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMetrics.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMetrics.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMetrics.java
index d51e007..1233de8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMetrics.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/SortMetrics.java
@@ -47,7 +47,7 @@ public class SortMetrics {
this.stats = stats;
}
- public void updateInputMetrics(int rowCount, int batchSize) {
+ public void updateInputMetrics(int rowCount, long batchSize) {
inputRecordCount += rowCount;
inputBatchCount++;
totalInputBytes += batchSize;
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorInitializer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorInitializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorInitializer.java
index 5dd348e..30c2a7a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorInitializer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/VectorInitializer.java
@@ -26,6 +26,8 @@ import org.apache.drill.exec.vector.ValueVector;
import org.apache.drill.exec.vector.complex.AbstractMapVector;
import org.apache.drill.exec.vector.complex.RepeatedMapVector;
+import com.google.common.annotations.VisibleForTesting;
+
/**
* Prototype mechanism to allocate vectors based on expected
* data sizes. This version uses a name-based map of fields
@@ -33,15 +35,20 @@ import org.apache.drill.exec.vector.complex.RepeatedMapVector;
* simply iterate over the schema rather than doing a per-field
* lookup. But, the mechanisms needed to do the efficient solution
* don't exist yet.
+ * <p>
+ * The element count is a float because it might be fractional.
+ * If every tenth array has an element, then the count per array
+ * is 0.1.
*/
public class VectorInitializer {
- private static class AllocationHint {
+ @VisibleForTesting
+ public static class AllocationHint {
public final int entryWidth;
- public final int elementCount;
+ public final float elementCount;
- private AllocationHint(int width, int elements) {
+ private AllocationHint(int width, float elements) {
entryWidth = width;
elementCount = elements;
}
@@ -72,12 +79,16 @@ public class VectorInitializer {
hints.put(name, new AllocationHint(width, 1));
}
- public void fixedWidthArray(String name, int elements) {
+ public void fixedWidthArray(String name, float elements) {
hints.put(name, new AllocationHint(0, elements));
}
- public void variableWidthArray(String name, int width, int elements) {
- hints.put(name, new AllocationHint(width, elements));
+ public void variableWidthArray(String name, float width, float elements) {
+ hints.put(name, new AllocationHint((int) Math.ceil(width), elements));
+ }
+
+ public AllocationHint hint(String name) {
+ return hints.get(name);
}
public void allocateBatch(VectorAccessible va, int recordCount) {
@@ -95,7 +106,7 @@ public class VectorInitializer {
allocateVector(vector, recordCount, hint);
}
// Set<BufferLedger> ledgers = new HashSet<>();
-// vector.getLedgers(ledgers);
+// vector.collectLedgers(ledgers);
// int size = 0;
// for (BufferLedger ledger : ledgers) {
// size += ledger.getAccountedSize();
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/main/resources/drill-module.conf
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/resources/drill-module.conf b/exec/java-exec/src/main/resources/drill-module.conf
index 0b8d605..a4597c0 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -242,9 +242,6 @@ drill.exec: {
sort: {
purge.threshold : 1000,
external: {
- // Drill uses the managed External Sort Batch by default.
- // Set this to true to use the legacy, unmanaged version.
- disable_managed: true,
// Limit on the number of batches buffered in memory.
// Primarily for testing.
// 0 = unlimited
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/SortTestUtilities.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/SortTestUtilities.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/SortTestUtilities.java
index 8ba34ef..f525b2a 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/SortTestUtilities.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/SortTestUtilities.java
@@ -102,7 +102,6 @@ public class SortTestUtilities {
}
int rowCount = outputRowCount();
VectorContainer dest = new VectorContainer();
- @SuppressWarnings("resource")
BatchMerger merger = copier.startMerge(schema.toBatchSchema(SelectionVectorMode.NONE),
batches, dest, rowCount, null);
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestExternalSortInternals.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestExternalSortInternals.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestExternalSortInternals.java
index 69e9e1b..1523dfe 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestExternalSortInternals.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestExternalSortInternals.java
@@ -472,9 +472,11 @@ public class TestExternalSortInternals extends DrillTest {
.put(ExecConstants.EXTERNAL_SORT_MERGE_LIMIT, mergeLimitConstraint)
.build();
SortConfig sortConfig = new SortConfig(drillConfig);
+
// Allow four spill batches, 8 MB each, plus one output of 16
// Allow for internal fragmentation
// 96 > (4 * 8 + 16) * 2
+
long memoryLimit = 96 * ONE_MEG;
SortMemoryManager memManager = new SortMemoryManager(sortConfig, memoryLimit);
@@ -486,7 +488,7 @@ public class TestExternalSortInternals extends DrillTest {
memManager.updateEstimates(batchSize, rowWidth, rowCount);
assertFalse(memManager.isLowMemory());
- int spillBatchBufferSize = memManager.getSpillBatchSize().expectedBufferSize;
+ int spillBatchBufferSize = memManager.getSpillBatchSize().maxBufferSize;
int inputBatchBufferSize = memManager.getInputBatchSize().expectedBufferSize;
// One in-mem batch, no merging.
@@ -502,7 +504,7 @@ public class TestExternalSortInternals extends DrillTest {
task = memManager.consolidateBatches(allocMemory, memBatches, 0);
assertEquals(MergeAction.NONE, task.action);
- // Spills if no room for spill and in-memory batches
+ // Spills if no room to merge spilled and in-memory batches
int spillCount = (int) Math.ceil((memManager.getMergeMemoryLimit() - allocMemory) / (1.0 * spillBatchBufferSize));
assertTrue(spillCount >= 1);
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestShortArrays.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestShortArrays.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestShortArrays.java
new file mode 100644
index 0000000..1c9d49e
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/managed/TestShortArrays.java
@@ -0,0 +1,116 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.drill.common.types.TypeProtos.DataMode;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.exec.physical.impl.spill.RecordBatchSizer;
+import org.apache.drill.exec.physical.impl.spill.RecordBatchSizer.ColumnSize;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.VectorInitializer;
+import org.apache.drill.exec.record.VectorInitializer.AllocationHint;
+import org.apache.drill.exec.vector.IntVector;
+import org.apache.drill.exec.vector.RepeatedIntVector;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.test.SubOperatorTest;
+import org.apache.drill.test.rowSet.RowSet;
+import org.apache.drill.test.rowSet.RowSet.SingleRowSet;
+import org.apache.drill.test.rowSet.RowSetBuilder;
+import org.apache.drill.test.rowSet.SchemaBuilder;
+import org.junit.Test;
+
+/**
+ * DRILL-5804.
+ * Code had a bug that if an array had low cardinality, the average,
+ * expressed as an int, was zero. We then allocated a zero length
+ * buffer, tried to double it, got another zero length buffer, and
+ * looped forever. This test verifies the fixes to avoid that case.
+ * @throws Exception
+ */
+
+
+public class TestShortArrays extends SubOperatorTest {
+
+ @Test
+ public void testSizer() {
+
+ // Create a row set with less than one item, on
+ // average, per array.
+
+ BatchSchema schema = new SchemaBuilder()
+ .add("a", MinorType.INT)
+ .addArray("b", MinorType.INT)
+ .build();
+ RowSetBuilder builder = fixture.rowSetBuilder(schema)
+ .add(1, new int[] {10});
+ for (int i = 2; i <= 10; i++) {
+ builder.add(i, new int[] {});
+ }
+ RowSet rows = builder.build();
+
+ // Run the record batch sizer on the resulting batch.
+
+ RecordBatchSizer sizer = new RecordBatchSizer(rows.container());
+ assertEquals(2, sizer.columns().size());
+ ColumnSize bCol = sizer.columns().get(1);
+ assertEquals(0.1, bCol.estElementCountPerArray, 0.01);
+ assertEquals(1, bCol.elementCount);
+
+ // Create a vector initializer using the sizer info.
+
+ VectorInitializer vi = sizer.buildVectorInitializer();
+ AllocationHint bHint = vi.hint("b");
+ assertNotNull(bHint);
+ assertEquals(bHint.elementCount, bCol.estElementCountPerArray, 0.001);
+
+ // Create a new batch, and new vector, using the sizer and
+ // intializer inferred from the previoius batch.
+
+ SingleRowSet empty = fixture.rowSet(schema);
+ vi.allocateBatch(empty.container(), 100);
+ assertEquals(2, empty.vectors().length);
+ @SuppressWarnings("resource")
+ ValueVector bVector = empty.vectors()[1];
+ assertTrue(bVector instanceof RepeatedIntVector);
+ assertEquals(16, ((RepeatedIntVector) bVector).getDataVector().getValueCapacity());
+
+ rows.clear();
+ empty.clear();
+ }
+
+ /**
+ * Test that a zero-length vector, on reAlloc, will default
+ * to 256 bytes. (Previously the code just doubled zero
+ * forever.)
+ */
+
+ @Test
+ public void testReAllocZeroSize() {
+ try (IntVector vector = new IntVector(
+ SchemaBuilder.columnSchema("a", MinorType.INT, DataMode.REQUIRED),
+ fixture.allocator())) {
+ vector.allocateNew(0);
+ vector.reAlloc();
+ assertEquals(256 / 4, vector.getValueCapacity());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractRowSet.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractRowSet.java b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractRowSet.java
index a32262a..6400a5b 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractRowSet.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractRowSet.java
@@ -153,7 +153,7 @@ public abstract class AbstractRowSet implements RowSet {
}
@Override
- public int size() {
+ public long size() {
throw new UnsupportedOperationException("getSize");
}
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractSingleRowSet.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractSingleRowSet.java b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractSingleRowSet.java
index d8176de..030f95a 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractSingleRowSet.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/AbstractSingleRowSet.java
@@ -184,7 +184,7 @@ public abstract class AbstractSingleRowSet extends AbstractRowSet implements Sin
public ValueVector[] vectors() { return valueVectors; }
@Override
- public int size() {
+ public long size() {
RecordBatchSizer sizer = new RecordBatchSizer(container);
return sizer.actualSize();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/IndirectRowSet.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/IndirectRowSet.java b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/IndirectRowSet.java
index 17a0ac8..1914705 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/IndirectRowSet.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/IndirectRowSet.java
@@ -118,7 +118,7 @@ public class IndirectRowSet extends AbstractSingleRowSet {
public SingleRowSet toIndirect() { return this; }
@Override
- public int size() {
+ public long size() {
RecordBatchSizer sizer = new RecordBatchSizer(container, sv2);
return sizer.actualSize();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSet.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSet.java b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSet.java
index b6bbd4f..6401f75 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSet.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSet.java
@@ -160,7 +160,7 @@ public interface RowSet {
* @return memory size in bytes
*/
- int size();
+ long size();
RowSet merge(RowSet other);
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/vector/src/main/codegen/templates/FixedValueVectors.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/codegen/templates/FixedValueVectors.java b/exec/vector/src/main/codegen/templates/FixedValueVectors.java
index bbda672..51938a3 100644
--- a/exec/vector/src/main/codegen/templates/FixedValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/FixedValueVectors.java
@@ -187,19 +187,34 @@ public final class ${minor.class}Vector extends BaseDataValueVector implements F
}
/**
- * Allocate new buffer with double capacity, and copy data into the new buffer. Replace vector's buffer with new buffer, and release old one
+ * Allocate new buffer with double capacity, and copy data into the new
+ * buffer. Replace vector's buffer with new buffer, and release old one
*
- * @throws org.apache.drill.exec.memory.OutOfMemoryException if it can't allocate the new buffer
+ * @throws org.apache.drill.exec.memory.OutOfMemoryException
+ * if it can't allocate the new buffer
*/
+
public void reAlloc() {
- final long newAllocationSize = allocationSizeInBytes * 2L;
+
+ // Avoid an infinite loop if we try to double the size of
+ // a zero-length buffer. Instead, just allocate a 256 byte
+ // buffer if we start at 0.
+
+ final long newAllocationSize = allocationSizeInBytes == 0
+ ? 256
+ : allocationSizeInBytes * 2L;
+
// TODO: Replace this with MAX_BUFFER_SIZE once all
// code is aware of the maximum vector size.
+
if (newAllocationSize > MAX_ALLOCATION_SIZE) {
throw new OversizedAllocationException("Unable to expand the buffer. Max allowed buffer size is reached.");
}
logger.debug("Reallocating vector [{}]. # of bytes: [{}] -> [{}]", field, allocationSizeInBytes, newAllocationSize);
+ if (newAllocationSize == 0) {
+ throw new IllegalStateException("Attempt to reAlloc a zero-sized vector");
+ }
final DrillBuf newBuf = allocator.buffer((int)newAllocationSize);
newBuf.setBytes(0, data, 0, data.capacity());
final int halfNewCapacity = newBuf.capacity() / 2;
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/vector/src/main/codegen/templates/NullableValueVectors.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/codegen/templates/NullableValueVectors.java b/exec/vector/src/main/codegen/templates/NullableValueVectors.java
index 21959f6..4f3eb17 100644
--- a/exec/vector/src/main/codegen/templates/NullableValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/NullableValueVectors.java
@@ -207,6 +207,7 @@ public final class ${className} extends BaseDataValueVector implements <#if type
accessor.reset();
}
+ @Override
public void reset() {
bits.zeroVector();
mutator.reset();
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java b/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
index 9780b7d..4da3d9e 100644
--- a/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
@@ -183,6 +183,7 @@ public final class Repeated${minor.class}Vector extends BaseRepeatedValueVector
copyFromSafe(fromIndex, toIndex, (Repeated${minor.class}Vector) from);
}
+ @Override
public boolean allocateNewSafe() {
/* boolean to keep track if all the memory allocations were successful.
* Used in the case of composite vectors when we need to allocate multiple
@@ -224,6 +225,7 @@ public final class Repeated${minor.class}Vector extends BaseRepeatedValueVector
.setVarByteLength(values.getVarByteLength());
}
+ @Override
public void allocateNew(int totalBytes, int valueCount, int innerValueCount) {
try {
offsets.allocateNew(valueCount + 1);
@@ -236,6 +238,7 @@ public final class Repeated${minor.class}Vector extends BaseRepeatedValueVector
mutator.reset();
}
+ @Override
public int getByteCapacity(){
return values.getByteCapacity();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/e2a28f7f/exec/vector/src/main/java/org/apache/drill/exec/vector/AllocationHelper.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/AllocationHelper.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/AllocationHelper.java
index 1401373..100997e 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/AllocationHelper.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/AllocationHelper.java
@@ -27,6 +27,12 @@ public class AllocationHelper {
}
public static void allocatePrecomputedChildCount(ValueVector vector, int valueCount, int bytesPerValue, int childValCount) {
+
+ // Don't fail if given a 0-sized array. Assume at least one value.
+ childValCount = Math.max(1, childValCount);
+ valueCount = Math.max(1, valueCount);
+ bytesPerValue = Math.max(1, bytesPerValue);
+
if (vector instanceof FixedWidthVector) {
((FixedWidthVector) vector).allocateNew(valueCount);
} else if (vector instanceof VariableWidthVector) {
@@ -47,6 +53,21 @@ public class AllocationHelper {
}
/**
+ * Allocate an array, but with a fractional value for the number of elements
+ * per array. This form is useful when the number comes from observations and
+ * represents an average.
+ *
+ * @param vector the vector to allocate
+ * @param valueCount the number of top-level values
+ * @param bytesPerValue the width of each value
+ * @param repeatedPerTop the number of array elements per value.
+ */
+ public static void allocate(ValueVector vector, int valueCount, int bytesPerValue, float repeatedPerTop){
+ allocatePrecomputedChildCount(vector, valueCount, bytesPerValue,
+ (int) Math.ceil(repeatedPerTop * valueCount));
+ }
+
+ /**
* Allocates the exact amount if v is fixed width, otherwise falls back to dynamic allocation
* @param vector value vector we are trying to allocate
* @param valueCount size we are trying to allocate