You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by mm...@apache.org on 2017/10/18 19:20:41 UTC
[4/4] hive git commit: HIVE-17164: Vectorization: Support PTF (Part
2: Unbounded Support-- Turn ON by default) (Matt McCline,
reviewed by Teddy Choi)
HIVE-17164: Vectorization: Support PTF (Part 2: Unbounded Support-- Turn ON by default) (Matt McCline, reviewed by Teddy Choi)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/c129bb95
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/c129bb95
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/c129bb95
Branch: refs/heads/master
Commit: c129bb95d1f864a004f21a051bbcddfb61464528
Parents: acb3ba2
Author: Matt McCline <mm...@hortonworks.com>
Authored: Wed Oct 18 14:19:41 2017 -0500
Committer: Matt McCline <mm...@hortonworks.com>
Committed: Wed Oct 18 14:19:41 2017 -0500
----------------------------------------------------------------------
.../org/apache/hadoop/hive/conf/HiveConf.java | 12 +-
.../persistence/HybridHashTableContainer.java | 8 +-
.../hive/ql/exec/tez/ReduceRecordProcessor.java | 3 +-
.../hive/ql/exec/tez/ReduceRecordSource.java | 16 +-
.../hive/ql/exec/vector/VectorSerializeRow.java | 16 +
.../ql/exec/vector/VectorizedBatchUtil.java | 232 +-
.../VectorMapJoinGenerateResultOperator.java | 5 +-
.../mapjoin/VectorMapJoinRowBytesContainer.java | 321 ---
.../exec/vector/ptf/VectorPTFGroupBatches.java | 273 ++-
.../ql/exec/vector/ptf/VectorPTFOperator.java | 39 +-
.../VectorRowBytesContainer.java | 331 +++
.../hive/ql/optimizer/physical/Vectorizer.java | 69 +-
.../apache/hadoop/hive/ql/plan/BaseWork.java | 9 +
.../hadoop/hive/ql/plan/VectorPTFDesc.java | 33 +-
.../vector/TestVectorRowBytesContainer.java | 76 +
.../TestVectorMapJoinRowBytesContainer.java | 75 -
.../clientpositive/vector_ptf_part_simple.q | 47 +
.../vector_windowing_expressions.q | 35 +
.../llap/vector_ptf_part_simple.q.out | 342 +++
.../llap/vector_windowing_expressions.q.out | 2245 ++++++++++++++++++
.../vector_windowing_expressions.q.out | 223 ++
.../hive/ql/exec/vector/BytesColumnVector.java | 2 +-
.../hive/ql/exec/vector/ColumnVector.java | 5 +-
.../ql/exec/vector/DecimalColumnVector.java | 2 +-
.../hive/ql/exec/vector/DoubleColumnVector.java | 2 +-
.../vector/IntervalDayTimeColumnVector.java | 2 +-
.../hive/ql/exec/vector/ListColumnVector.java | 2 +-
.../hive/ql/exec/vector/LongColumnVector.java | 2 +-
.../hive/ql/exec/vector/MapColumnVector.java | 2 +-
.../ql/exec/vector/MultiValuedColumnVector.java | 4 +-
.../hive/ql/exec/vector/StructColumnVector.java | 2 +-
.../ql/exec/vector/TimestampColumnVector.java | 2 +-
.../hive/ql/exec/vector/UnionColumnVector.java | 2 +-
33 files changed, 3915 insertions(+), 524 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index 1934126..a6ecb37 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -2872,9 +2872,17 @@ public class HiveConf extends Configuration {
"1. chosen : use VectorUDFAdaptor for a small set of UDFs that were choosen for good performance\n" +
"2. all : use VectorUDFAdaptor for all UDFs"
),
- HIVE_VECTORIZATION_PTF_ENABLED("hive.vectorized.execution.ptf.enabled", false,
+ HIVE_VECTORIZATION_PTF_ENABLED("hive.vectorized.execution.ptf.enabled", true,
"This flag should be set to true to enable vectorized mode of the PTF of query execution.\n" +
- "The default value is false."),
+ "The default value is true."),
+
+ HIVE_VECTORIZATION_PTF_MAX_MEMORY_BUFFERING_BATCH_COUNT("hive.vectorized.ptf.max.memory.buffering.batch.count", 25,
+ "Maximum number of vectorized row batches to buffer in memory for PTF\n" +
+ "The default value is 25"),
+ HIVE_VECTORIZATION_TESTING_REDUCER_BATCH_SIZE("hive.vectorized.testing.reducer.batch.size", -1,
+ "internal use only, used for creating small group key vectorized row batches to exercise more logic\n" +
+ "The default value is -1 which means don't restrict for testing",
+ true),
HIVE_VECTORIZATION_COMPLEX_TYPES_ENABLED("hive.vectorized.complex.types.enabled", true,
"This flag should be set to true to enable vectorization\n" +
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
index 6523f00..ac4f5ab 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
@@ -42,7 +42,7 @@ import org.apache.hadoop.hive.ql.exec.persistence.MapJoinBytesTableContainer.Key
import org.apache.hadoop.hive.ql.exec.vector.VectorHashKeyWrapper;
import org.apache.hadoop.hive.ql.exec.vector.VectorHashKeyWrapperBatch;
import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpressionWriter;
-import org.apache.hadoop.hive.ql.exec.vector.mapjoin.VectorMapJoinRowBytesContainer;
+import org.apache.hadoop.hive.ql.exec.vector.rowbytescontainer.VectorRowBytesContainer;
import org.apache.hadoop.hive.ql.io.HiveKey;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.HiveUtils;
@@ -131,7 +131,7 @@ public class HybridHashTableContainer
BytesBytesMultiHashMap hashMap; // In memory hashMap
KeyValueContainer sidefileKVContainer; // Stores small table key/value pairs
ObjectContainer matchfileObjContainer; // Stores big table rows
- VectorMapJoinRowBytesContainer matchfileRowBytesContainer;
+ VectorRowBytesContainer matchfileRowBytesContainer;
// Stores big table rows as bytes for native vector map join.
Path hashMapLocalPath; // Local file system path for spilled hashMap
boolean hashMapOnDisk; // Status of hashMap. true: on disk, false: in memory
@@ -216,9 +216,9 @@ public class HybridHashTableContainer
}
/* Get the big table row bytes container for native vector map join */
- public VectorMapJoinRowBytesContainer getMatchfileRowBytesContainer() {
+ public VectorRowBytesContainer getMatchfileRowBytesContainer() {
if (matchfileRowBytesContainer == null) {
- matchfileRowBytesContainer = new VectorMapJoinRowBytesContainer(spillLocalDirs);
+ matchfileRowBytesContainer = new VectorRowBytesContainer(spillLocalDirs);
}
return matchfileRowBytesContainer;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
index f6f2dd0..9de3850 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
@@ -297,7 +297,8 @@ public class ReduceRecordProcessor extends RecordProcessor {
boolean vectorizedRecordSource = (tag == bigTablePosition) && redWork.getVectorMode();
sources[tag].init(jconf, redWork.getReducer(), vectorizedRecordSource, keyTableDesc,
valueTableDesc, reader, tag == bigTablePosition, (byte) tag,
- redWork.getVectorizedRowBatchCtx(), redWork.getVectorizedVertexNum());
+ redWork.getVectorizedRowBatchCtx(), redWork.getVectorizedVertexNum(),
+ redWork.getVectorizedTestingReducerBatchSize());
ois[tag] = sources[tag].getObjectInspector();
}
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java
index 872c223..95dbdba 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java
@@ -121,13 +121,22 @@ public class ReduceRecordSource implements RecordSource {
private final GroupIterator groupIterator = new GroupIterator();
private long vectorizedVertexNum;
+ private int vectorizedTestingReducerBatchSize;
void init(JobConf jconf, Operator<?> reducer, boolean vectorized, TableDesc keyTableDesc,
TableDesc valueTableDesc, Reader reader, boolean handleGroupKey, byte tag,
- VectorizedRowBatchCtx batchContext, long vectorizedVertexNum)
+ VectorizedRowBatchCtx batchContext, long vectorizedVertexNum,
+ int vectorizedTestingReducerBatchSize)
throws Exception {
this.vectorizedVertexNum = vectorizedVertexNum;
+ if (vectorizedTestingReducerBatchSize > VectorizedRowBatch.DEFAULT_SIZE) {
+
+ // For now, we don't go higher than the default batch size unless we do more work
+ // to verify every vectorized operator downstream can handle a larger batch size.
+ vectorizedTestingReducerBatchSize = VectorizedRowBatch.DEFAULT_SIZE;
+ }
+ this.vectorizedTestingReducerBatchSize = vectorizedTestingReducerBatchSize;
ObjectInspector keyObjectInspector;
this.reducer = reducer;
@@ -417,7 +426,10 @@ public class ReduceRecordSource implements RecordSource {
VectorizedBatchUtil.setRepeatingColumn(batch, i);
}
- final int maxSize = batch.getMaxSize();
+ final int maxSize =
+ (vectorizedTestingReducerBatchSize > 0 ?
+ Math.min(vectorizedTestingReducerBatchSize, batch.getMaxSize()) :
+ batch.getMaxSize());
Preconditions.checkState(maxSize > 0);
int rowIdx = 0;
int batchBytes = keyBytes.length;
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRow.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRow.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRow.java
index be471c6..211622d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRow.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRow.java
@@ -111,6 +111,22 @@ public final class VectorSerializeRow<T extends SerializeWrite> {
vectorExtractRow.init(typeInfos);
}
+ public void init(TypeInfo[] typeInfos)
+ throws HiveException {
+
+ final int size = typeInfos.length;
+ this.typeInfos = Arrays.copyOf(typeInfos, size);
+ outputColumnNums = new int[size];
+ objectInspectors = new ObjectInspector[size];
+ for (int i = 0; i < size; i++) {
+ objectInspectors[i] =
+ TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(typeInfos[i]);
+ outputColumnNums[i] = i;
+ }
+
+ vectorExtractRow.init(this.typeInfos, outputColumnNums);
+ }
+
public void init(TypeInfo[] typeInfos, int[] columnMap)
throws HiveException {
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedBatchUtil.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedBatchUtil.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedBatchUtil.java
index 03c09e7..4945d74 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedBatchUtil.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedBatchUtil.java
@@ -625,12 +625,168 @@ public class VectorizedBatchUtil {
" is not supported!");
}
- public static void swapColumnVector(
- VectorizedRowBatch batch1, int batch1ColumnNum,
- VectorizedRowBatch batch2, int batch2ColumnNum) {
- ColumnVector colVector1 = batch1.cols[batch1ColumnNum];
- batch1.cols[batch1ColumnNum] = batch2.cols[batch2ColumnNum];
- batch2.cols[batch2ColumnNum] = colVector1;
+ private static final byte[] EMPTY_BYTES = new byte[0];
+ private static final HiveIntervalDayTime emptyIntervalDayTime = new HiveIntervalDayTime(0, 0);
+
+ public static void copyNonSelectedColumnVector(
+ VectorizedRowBatch sourceBatch, int sourceColumnNum,
+ VectorizedRowBatch targetBatch, int targetColumnNum,
+ int size) {
+
+ ColumnVector sourceColVector = sourceBatch.cols[sourceColumnNum];
+ ColumnVector targetColVector = targetBatch.cols[targetColumnNum];
+ if (sourceColVector.noNulls && targetColVector.noNulls) {
+ // No isNull copying necessary.
+ } else if (sourceColVector.noNulls) {
+
+ // Clear out isNull array.
+ targetColVector.reset();
+ } else {
+ System.arraycopy(sourceColVector.isNull, 0, targetColVector.isNull, 0, size);
+ targetColVector.noNulls = false;
+ }
+ if (sourceColVector.isRepeating) {
+ size = 1;
+ targetColVector.isRepeating = true;
+ } else {
+ targetColVector.isRepeating = false;
+ }
+
+ // Primitive column types ignore nulls and just copy all values.
+ switch (sourceColVector.type) {
+ case LONG:
+ {
+ long[] sourceVector = ((LongColumnVector) sourceColVector).vector;
+ long[] targetVector = ((LongColumnVector) targetColVector).vector;
+ System.arraycopy(sourceVector, 0, targetVector, 0, size);
+ }
+ break;
+ case DOUBLE:
+ {
+ double[] sourceVector = ((DoubleColumnVector) sourceColVector).vector;
+ double[] targetVector = ((DoubleColumnVector) targetColVector).vector;
+ System.arraycopy(sourceVector, 0, targetVector, 0, size);
+ }
+ break;
+ case BYTES:
+ {
+ BytesColumnVector sourceBytesColVector = ((BytesColumnVector) sourceColVector);
+ byte[][] sourceVector = sourceBytesColVector.vector;
+ int[] sourceStart = sourceBytesColVector.start;
+ int[] sourceLength = sourceBytesColVector.length;
+
+ BytesColumnVector targetBytesColVector = ((BytesColumnVector) targetColVector);
+
+ if (sourceColVector.noNulls) {
+ for (int i = 0; i < size; i++) {
+ targetBytesColVector.setVal(i, sourceVector[i], sourceStart[i], sourceLength[i]);
+ }
+ } else {
+ boolean[] sourceIsNull = sourceColVector.isNull;
+
+ // Target isNull was copied at beginning of method.
+ for (int i = 0; i < size; i++) {
+ if (!sourceIsNull[i]) {
+ targetBytesColVector.setVal(i, sourceVector[i], sourceStart[i], sourceLength[i]);
+ } else {
+ targetBytesColVector.setRef(i, EMPTY_BYTES, 0, 0);
+ }
+ }
+ }
+ }
+ break;
+ case DECIMAL:
+ {
+ DecimalColumnVector sourceDecimalColVector = ((DecimalColumnVector) sourceColVector);
+ HiveDecimalWritable[] sourceVector = sourceDecimalColVector.vector;
+
+ DecimalColumnVector targetDecimalColVector = ((DecimalColumnVector) targetColVector);
+
+ if (sourceColVector.noNulls) {
+ for (int i = 0; i < size; i++) {
+ targetDecimalColVector.set(i, sourceVector[i]);
+ }
+ } else {
+ boolean[] sourceIsNull = sourceColVector.isNull;
+
+ // Target isNull was copied at beginning of method.
+ for (int i = 0; i < size; i++) {
+ if (!sourceIsNull[i]) {
+ targetDecimalColVector.set(i, sourceVector[i]);
+ } else {
+ targetDecimalColVector.vector[i].setFromLong(0);
+ }
+ }
+ }
+ }
+ break;
+ case TIMESTAMP:
+ {
+ TimestampColumnVector sourceTimestampColVector = ((TimestampColumnVector) sourceColVector);
+ long[] sourceTime = sourceTimestampColVector.time;
+ int[] sourceNanos = sourceTimestampColVector.nanos;
+
+ TimestampColumnVector targetTimestampColVector = ((TimestampColumnVector) targetColVector);
+ long[] targetTime = targetTimestampColVector.time;
+ int[] targetNanos = targetTimestampColVector.nanos;
+
+ if (sourceColVector.noNulls) {
+ for (int i = 0; i < size; i++) {
+ targetTime[i] = sourceTime[i];
+ targetNanos[i] = targetNanos[i];
+ }
+ } else {
+ boolean[] sourceIsNull = sourceColVector.isNull;
+
+ // Target isNull was copied at beginning of method.
+ for (int i = 0; i < size; i++) {
+ if (!sourceIsNull[i]) {
+ targetTime[i] = sourceTime[i];
+ targetNanos[i] = targetNanos[i];
+ } else {
+ targetTime[i] = 0;
+ targetNanos[i] = 0;
+ }
+ }
+ }
+ }
+ break;
+ case INTERVAL_DAY_TIME:
+ {
+ IntervalDayTimeColumnVector sourceIntervalDayTimeColVector = ((IntervalDayTimeColumnVector) sourceColVector);
+
+ IntervalDayTimeColumnVector targetIntervalDayTimeColVector = ((IntervalDayTimeColumnVector) targetColVector);
+
+ if (sourceColVector.noNulls) {
+ for (int i = 0; i < size; i++) {
+ targetIntervalDayTimeColVector.set(
+ i, targetIntervalDayTimeColVector.asScratchIntervalDayTime(i));
+ }
+ } else {
+ boolean[] sourceIsNull = sourceColVector.isNull;
+
+ // Target isNull was copied at beginning of method.
+ for (int i = 0; i < size; i++) {
+ if (!sourceIsNull[i]) {
+ targetIntervalDayTimeColVector.set(
+ i, targetIntervalDayTimeColVector.asScratchIntervalDayTime(i));
+ } else {
+ targetIntervalDayTimeColVector.set(
+ i, emptyIntervalDayTime);
+ }
+ }
+ }
+ }
+ break;
+ case STRUCT:
+ case LIST:
+ case MAP:
+ case UNION:
+ // No complex type support for now.
+ default:
+ throw new RuntimeException("Unexpected column vector type " + sourceColVector.type);
+ }
+
}
public static void copyRepeatingColumn(VectorizedRowBatch sourceBatch, int sourceColumnNum,
@@ -646,29 +802,47 @@ public class VectorizedBatchUtil {
return;
}
- if (sourceColVector instanceof LongColumnVector) {
- ((LongColumnVector) targetColVector).vector[0] = ((LongColumnVector) sourceColVector).vector[0];
- } else if (sourceColVector instanceof DoubleColumnVector) {
- ((DoubleColumnVector) targetColVector).vector[0] = ((DoubleColumnVector) sourceColVector).vector[0];
- } else if (sourceColVector instanceof BytesColumnVector) {
- BytesColumnVector bytesColVector = (BytesColumnVector) sourceColVector;
- byte[] bytes = bytesColVector.vector[0];
- final int start = bytesColVector.start[0];
- final int length = bytesColVector.length[0];
- if (setByValue) {
- ((BytesColumnVector) targetColVector).setVal(0, bytes, start, length);
- } else {
- ((BytesColumnVector) targetColVector).setRef(0, bytes, start, length);
- }
- } else if (sourceColVector instanceof DecimalColumnVector) {
- ((DecimalColumnVector) targetColVector).set(0, ((DecimalColumnVector) sourceColVector).vector[0]);
- } else if (sourceColVector instanceof TimestampColumnVector) {
- ((TimestampColumnVector) targetColVector).set(0, ((TimestampColumnVector) sourceColVector).asScratchTimestamp(0));
- } else if (sourceColVector instanceof IntervalDayTimeColumnVector) {
- ((IntervalDayTimeColumnVector) targetColVector).set(0, ((IntervalDayTimeColumnVector) sourceColVector).asScratchIntervalDayTime(0));
- } else {
- throw new RuntimeException("Column vector class " + sourceColVector.getClass().getName() +
- " is not supported!");
+ switch (sourceColVector.type) {
+ case LONG:
+ ((LongColumnVector) targetColVector).vector[0] =
+ ((LongColumnVector) sourceColVector).vector[0];
+ break;
+ case DOUBLE:
+ ((DoubleColumnVector) targetColVector).vector[0] =
+ ((DoubleColumnVector) sourceColVector).vector[0];
+ break;
+ case BYTES:
+ {
+ BytesColumnVector bytesColVector = (BytesColumnVector) sourceColVector;
+ byte[] bytes = bytesColVector.vector[0];
+ final int start = bytesColVector.start[0];
+ final int length = bytesColVector.length[0];
+ if (setByValue) {
+ ((BytesColumnVector) targetColVector).setVal(0, bytes, start, length);
+ } else {
+ ((BytesColumnVector) targetColVector).setRef(0, bytes, start, length);
+ }
+ }
+ break;
+ case DECIMAL:
+ ((DecimalColumnVector) targetColVector).set(
+ 0, ((DecimalColumnVector) sourceColVector).vector[0]);
+ break;
+ case TIMESTAMP:
+ ((TimestampColumnVector) targetColVector).set(
+ 0, ((TimestampColumnVector) sourceColVector).asScratchTimestamp(0));
+ break;
+ case INTERVAL_DAY_TIME:
+ ((IntervalDayTimeColumnVector) targetColVector).set(
+ 0, ((IntervalDayTimeColumnVector) sourceColVector).asScratchIntervalDayTime(0));
+ break;
+ case STRUCT:
+ case LIST:
+ case MAP:
+ case UNION:
+ // No complex type support for now.
+ default:
+ throw new RuntimeException("Unexpected column vector type " + sourceColVector.type);
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinGenerateResultOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinGenerateResultOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinGenerateResultOperator.java
index bab5ee4..ecf4b9a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinGenerateResultOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinGenerateResultOperator.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpression;
import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashTableResult;
import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashMapResult;
import org.apache.hadoop.hive.ql.exec.vector.mapjoin.optimized.VectorMapJoinOptimizedCreateHashTable;
+import org.apache.hadoop.hive.ql.exec.vector.rowbytescontainer.VectorRowBytesContainer;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
import org.apache.hadoop.hive.serde2.SerDeException;
@@ -481,7 +482,7 @@ public abstract class VectorMapJoinGenerateResultOperator extends VectorMapJoinC
HybridHashTableContainer ht = (HybridHashTableContainer) mapJoinTables[posSingleVectorMapJoinSmallTable];
HashPartition hp = ht.getHashPartitions()[partitionId];
- VectorMapJoinRowBytesContainer rowBytesContainer = hp.getMatchfileRowBytesContainer();
+ VectorRowBytesContainer rowBytesContainer = hp.getMatchfileRowBytesContainer();
Output output = rowBytesContainer.getOuputForRowBytes();
// int offset = output.getLength();
bigTableVectorSerializeRow.setOutputAppend(output);
@@ -568,7 +569,7 @@ public abstract class VectorMapJoinGenerateResultOperator extends VectorMapJoinC
int batchCount = 0;
try {
- VectorMapJoinRowBytesContainer bigTable = partition.getMatchfileRowBytesContainer();
+ VectorRowBytesContainer bigTable = partition.getMatchfileRowBytesContainer();
bigTable.prepareForReading();
while (bigTable.readNext()) {
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinRowBytesContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinRowBytesContainer.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinRowBytesContainer.java
deleted file mode 100644
index fa96ae9..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinRowBytesContainer.java
+++ /dev/null
@@ -1,321 +0,0 @@
-/**
- * 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.hadoop.hive.ql.exec.vector.mapjoin;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.hadoop.fs.FileUtil;
-import org.apache.hadoop.hive.common.FileUtils;
-import org.apache.hadoop.hive.serde2.ByteStream.Output;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-
-/**
- * An eager bytes container that puts row bytes to an output stream.
- */
-public class VectorMapJoinRowBytesContainer {
-
- private static final Logger LOG = LoggerFactory.getLogger(VectorMapJoinRowBytesContainer.class);
-
- private File parentDir;
- private File tmpFile;
-
- // We buffer in a org.apache.hadoop.hive.serde2.ByteStream.Output since that is what
- // is used by VectorSerializeRow / SerializeWrite. Periodically, we flush this buffer
- // to disk.
- private Output output;
- private int rowBeginPos;
- private static final int OUTPUT_SIZE = 4096;
- private static final int THRESHOLD = 8 * (OUTPUT_SIZE / 10);
- private static final int INPUT_SIZE = 4096;
-
- private FileOutputStream fileOutputStream;
-
- private boolean isOpen;
-
- private byte[] readBuffer;
- private byte[] largeRowBuffer;
- private int readOffset;
- private int readLength;
-
- private int readNextCount;
- private int readNextIndex;
-
- private static final int MAX_READS = 256;
- private byte[][] readNextBytes;
- private int readNextOffsets[];
- private int readNextLengths[];
-
- private byte[] currentBytes;
- private int currentOffset;
- private int currentLength;
-
- private long totalWriteLength;
- private long totalReadLength;
-
- private FileInputStream fileInputStream;
-
- private final String spillLocalDirs;
-
- public VectorMapJoinRowBytesContainer(String spillLocalDirs) {
- output = new Output();
- readBuffer = new byte[INPUT_SIZE];
- readNextBytes = new byte[MAX_READS][];
- readNextOffsets = new int[MAX_READS];
- readNextLengths = new int[MAX_READS];
- isOpen = false;
- totalWriteLength = 0;
- totalReadLength = 0;
- this.spillLocalDirs = spillLocalDirs;
- }
-
- private void setupOutputFileStreams() throws IOException {
- parentDir = FileUtils.createLocalDirsTempFile(spillLocalDirs, "bytes-container", "", true);
- parentDir.deleteOnExit();
- tmpFile = File.createTempFile("BytesContainer", ".tmp", parentDir);
- LOG.debug("BytesContainer created temp file " + tmpFile.getAbsolutePath());
- tmpFile.deleteOnExit();
-
- fileOutputStream = new FileOutputStream(tmpFile);
- }
-
- private void initFile() {
- try {
- setupOutputFileStreams();
- } catch (IOException e) {
- throw new RuntimeException("Failed to create temporary output file on disk", e);
- }
- }
-
- public Output getOuputForRowBytes() {
- if (!isOpen) {
- initFile();
- isOpen = true;
- }
- // Reserve space for the int length.
- output.reserve(4);
- rowBeginPos = output.getLength();
- return output;
- }
-
- public void finishRow() throws IOException {
- int length = output.getLength() - rowBeginPos;
- output.writeInt(rowBeginPos - 4, length);
- if (output.getLength() > THRESHOLD) {
- fileOutputStream.write(output.getData(), 0, output.getLength());
- totalWriteLength += output.getLength();
- output.reset();
- }
- }
-
- public void prepareForReading() throws IOException {
- if (!isOpen) {
- return;
- }
- if (output.getLength() > 0) {
- fileOutputStream.write(output.getData(), 0, output.getLength());
- totalWriteLength += output.getLength();
- fileOutputStream.flush();
- output.reset();
- }
- if (fileInputStream != null) {
- fileInputStream.close();
- }
- fileInputStream = new FileInputStream(tmpFile);
- readNextIndex = 0;
- readNextCount = 0;
- }
-
- private int readInt() {
- int value = (((readBuffer[readOffset] & 0xFF) << 24) |
- ((readBuffer[readOffset + 1] & 0xFF) << 16) |
- ((readBuffer[readOffset + 2] & 0xFF) << 8) |
- ((readBuffer[readOffset + 3] & 0xFF)));
- readOffset += 4;
- return value;
- }
-
- // Call when nextReadIndex == nextReadCount.
- private void bufferedRead() throws IOException {
-
- // Reset for reading.
- readNextIndex = 0;
-
- // Reset for filling.
- readNextCount = 0;
-
- if (readOffset < readLength) {
- // Move unprocessed remainder to beginning of buffer.
- int unprocessLength = readLength - readOffset;
- System.arraycopy(readBuffer, readOffset, readBuffer, 0, unprocessLength);
-
- int maxReadLength = readBuffer.length - unprocessLength;
- int partialReadLength = fileInputStream.read(readBuffer, unprocessLength, maxReadLength);
- if (partialReadLength == -1) {
- partialReadLength = 0;
- }
- totalReadLength += partialReadLength;
- readLength = unprocessLength + partialReadLength;
- readOffset = 0;
- } else {
- readOffset = 0;
- readLength = fileInputStream.read(readBuffer, 0, readBuffer.length);
- if (readLength == -1) {
- readLength = 0;
- }
- totalReadLength += readLength;
- }
- if (readLength == 0) {
- return;
- }
- if (readLength < 0) {
- throw new IOException("Negative read length");
- }
-
- // Get length word.
- if (readLength < 4) {
- throw new IOException("Expecting 4 byte length");
- }
-
- while (true) {
- // Use Input class to read length.
- int saveReadOffset = readOffset;
- int rowLength = readInt();
- if (rowLength < 0) {
- throw new IOException("Negative row length");
- }
- int remainingLength = readLength - readOffset;
- if (remainingLength < rowLength) {
- if (readNextCount > 0) {
- // Leave this one for the next round.
- readOffset = saveReadOffset;
- break;
- }
-
- // Buffer needed to bridge.
- if (largeRowBuffer == null || largeRowBuffer.length < rowLength) {
- int newLargeBufferLength = Math.max(Integer.highestOneBit(rowLength) << 1, INPUT_SIZE);
- largeRowBuffer = new byte[newLargeBufferLength];
- }
- System.arraycopy(readBuffer, readOffset, largeRowBuffer, 0, remainingLength);
- int expectedPartialLength = rowLength - remainingLength;
- int partialReadLength = fileInputStream.read(largeRowBuffer, remainingLength, expectedPartialLength);
- if (partialReadLength == -1) {
- throw new IOException("Unexpected EOF (total write length " + totalWriteLength +
- ", total read length " + totalReadLength + ", read length " +
- expectedPartialLength + ")");
- }
-
- if (expectedPartialLength != partialReadLength) {
- throw new IOException("Unable to read a complete row of length " + rowLength +
- " (total write length " + totalWriteLength +
- ", total read length " + totalReadLength + ", read length " +
- expectedPartialLength + ", actual length " + partialReadLength + ")");
- }
- totalReadLength += partialReadLength;
-
- readNextBytes[readNextCount] = largeRowBuffer;
- readNextOffsets[readNextCount] = 0;
- readNextLengths[readNextCount] = rowLength;
-
- // Indicate we used the last row's bytes for large buffer.
- readOffset = readLength;
- readNextCount++;
- break;
- }
-
- readNextBytes[readNextCount] = readBuffer;
- readNextOffsets[readNextCount] = readOffset;
- readNextLengths[readNextCount] = rowLength;
- readOffset += rowLength;
- readNextCount++;
-
- if (readNextCount >= readNextBytes.length){
- break;
- }
- if (readLength - readOffset < 4) {
- // Handle in next round.
- break;
- }
- }
- }
-
- public boolean readNext() throws IOException {
- if (!isOpen) {
- return false;
- }
- if (readNextIndex >= readNextCount) {
- bufferedRead();
- // Any more left?
- if (readNextIndex >= readNextCount) {
- return false;
- }
- }
-
- currentBytes = readNextBytes[readNextIndex];
- currentOffset = readNextOffsets[readNextIndex];
- currentLength = readNextLengths[readNextIndex];
-
- readNextIndex++;
- return true;
- }
-
- public byte[] currentBytes() {
- return currentBytes;
- }
-
- public int currentOffset() {
- return currentOffset;
- }
-
- public int currentLength() {
- return currentLength;
- }
-
- public void clear() {
- if (fileInputStream != null) {
- try {
- fileInputStream.close();
- } catch (Throwable ignored) {
- }
- fileInputStream = null;
- }
- if (fileOutputStream != null) {
- try {
- fileOutputStream.close();
- } catch (Throwable ignored) {
- }
- fileOutputStream = null;
- }
-
- if (parentDir != null) {
- try {
- FileUtil.fullyDelete(parentDir);
- } catch (Throwable ignored) {
- }
- }
- parentDir = null;
- tmpFile = null;
- isOpen = false;
- totalWriteLength = 0;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java
index a843f48..0c611a8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java
@@ -18,17 +18,26 @@
package org.apache.hadoop.hive.ql.exec.vector.ptf;
+import java.io.IOException;
import java.util.ArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorDeserializeRow;
+import org.apache.hadoop.hive.ql.exec.vector.VectorSerializeRow;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedBatchUtil;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.ql.exec.vector.rowbytescontainer.VectorRowBytesContainer;
import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.HiveUtils;
+import org.apache.hadoop.hive.serde2.ByteStream.Output;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import com.google.common.base.Preconditions;
@@ -42,10 +51,15 @@ public class VectorPTFGroupBatches {
private static final String CLASS_NAME = VectorPTFGroupBatches.class.getName();
private static final Log LOG = LogFactory.getLog(CLASS_NAME);
+ private Configuration hconf;
+
private VectorPTFEvaluatorBase[] evaluators;
- private int[] outputColumnMap;
+ private int[] outputProjectionColumnMap;
private int[] keyInputColumnMap;
+
+ private int bufferedColumnCount;
private int[] bufferedColumnMap;
+ private TypeInfo[] bufferedTypeInfos;
private ArrayList<VectorizedRowBatch> bufferedBatches;
@@ -54,31 +68,101 @@ public class VectorPTFGroupBatches {
private int allocatedBufferedBatchCount;
private int currentBufferedBatchCount;
- public VectorPTFGroupBatches() {
+ private int spillLimitBufferedBatchCount;
+ private boolean didSpillToDisk;
+ private String spillLocalDirs;
+ private long spillRowCount;
+ private VectorRowBytesContainer spillRowBytesContainer;
+
+ private transient VectorSerializeRow bufferedBatchVectorSerializeRow;
+ private transient VectorDeserializeRow bufferedBatchVectorDeserializeRow;
+
+ public VectorPTFGroupBatches(Configuration hconf, int vectorizedPTFMaxMemoryBufferingBatchCount) {
+ this.hconf = hconf;
allocatedBufferedBatchCount = 0;
currentBufferedBatchCount = 0;
+
+ spillLocalDirs = HiveUtils.getLocalDirList(hconf);
+
+ // Cannot be 0.
+ spillLimitBufferedBatchCount = Math.max(1, vectorizedPTFMaxMemoryBufferingBatchCount);
+
+ didSpillToDisk = false;
+ spillLocalDirs = null;
+ spillRowBytesContainer = null;
+ bufferedBatchVectorSerializeRow = null;
+ bufferedBatchVectorDeserializeRow = null;
}
- public void init(VectorPTFEvaluatorBase[] evaluators, int[] outputColumnMap,
- int[] keyInputColumnMap, int[] nonKeyInputColumnMap, int[] streamingColumnMap,
+ public void init(
+ TypeInfo[] reducerBatchTypeInfos,
+ VectorPTFEvaluatorBase[] evaluators,
+ int[] outputProjectionColumnMap, TypeInfo[] outputTypeInfos,
+ int[] keyInputColumnMap, int[] nonKeyInputColumnMap, int[] streamingEvaluatorNums,
VectorizedRowBatch overflowBatch) {
this.evaluators = evaluators;
- this.outputColumnMap = outputColumnMap;
+ this.outputProjectionColumnMap = outputProjectionColumnMap;
this.keyInputColumnMap = keyInputColumnMap;
+
+ /*
+ * If we have more than one group key batch, we will buffer their contents.
+ * We don't buffer the key columns since they are a constant for the group key.
+ *
+ * We buffer the non-key input columns. And, we buffer any streaming columns that will already
+ * have their output values.
+ */
final int nonKeyInputColumnCount = nonKeyInputColumnMap.length;
- final int streamingColumnCount = streamingColumnMap.length;
- final int bufferedColumnCount = nonKeyInputColumnCount + streamingColumnCount;
+ final int streamingEvaluatorCount = streamingEvaluatorNums.length;
+ bufferedColumnCount = nonKeyInputColumnCount + streamingEvaluatorCount;
bufferedColumnMap = new int[bufferedColumnCount];
+ bufferedTypeInfos = new TypeInfo[bufferedColumnCount];
for (int i = 0; i < nonKeyInputColumnCount; i++) {
- bufferedColumnMap[i] = nonKeyInputColumnMap[i];
+ final int columnNum = nonKeyInputColumnMap[i];
+ bufferedColumnMap[i] = columnNum;
+ bufferedTypeInfos[i] = reducerBatchTypeInfos[columnNum];
}
- for (int i = nonKeyInputColumnCount; i < bufferedColumnCount; i++) {
- bufferedColumnMap[i] = streamingColumnMap[i - nonKeyInputColumnCount];
+
+ for (int i = 0; i < streamingEvaluatorCount; i++) {
+ final int streamingEvaluatorNum = streamingEvaluatorNums[i];
+ final int bufferedMapIndex = nonKeyInputColumnCount + i;
+ bufferedColumnMap[bufferedMapIndex] = outputProjectionColumnMap[streamingEvaluatorNum];
+ bufferedTypeInfos[bufferedMapIndex] = outputTypeInfos[streamingEvaluatorNum];
}
this.overflowBatch = overflowBatch;
bufferedBatches = new ArrayList<VectorizedRowBatch>(0);
}
+ private VectorRowBytesContainer getSpillRowBytesContainer() throws HiveException {
+ if (spillRowBytesContainer == null) {
+ spillRowBytesContainer = new VectorRowBytesContainer(spillLocalDirs);
+
+ if (bufferedBatchVectorSerializeRow == null) {
+ bufferedBatchVectorSerializeRow =
+ new VectorSerializeRow<LazyBinarySerializeWrite>(
+ new LazyBinarySerializeWrite(bufferedColumnMap.length));
+
+ // Deserialize just the columns we a buffered batch, which has only the non-key inputs and
+ // streamed column outputs.
+ bufferedBatchVectorSerializeRow.init(bufferedTypeInfos);
+
+ bufferedBatchVectorDeserializeRow =
+ new VectorDeserializeRow<LazyBinaryDeserializeRead>(
+ new LazyBinaryDeserializeRead(
+ bufferedTypeInfos,
+ /* useExternalBuffer */ true));
+
+ // Deserialize the fields into the *overflow* batch using the buffered batch column map.
+ bufferedBatchVectorDeserializeRow.init(bufferedColumnMap);
+ }
+ }
+ return spillRowBytesContainer;
+ }
+
+ private void releaseSpillRowBytesContainer() {
+ spillRowBytesContainer.clear();
+ spillRowBytesContainer = null;
+ }
+
public void evaluateStreamingGroupBatch(VectorizedRowBatch batch, boolean isLastGroupBatch) {
// Streaming evaluators fill in their results during the evaluate call.
@@ -128,32 +212,101 @@ public class VectorPTFGroupBatches {
throws HiveException {
VectorizedRowBatch bufferedBatch = bufferedBatches.get(index);
- final int size = bufferedColumnMap.length;
- for (int i = 0; i < size; i++) {
+ final int size = bufferedBatch.size;
+ final int bufferedColumnCount = bufferedColumnMap.length;
+ for (int i = 0; i < bufferedColumnCount; i++) {
- // Swap ColumnVectors with overflowBatch. We remember buffered columns compactly in the
+ // Copy ColumnVectors to overflowBatch. We remember buffered columns compactly in the
// buffered VRBs without other columns or scratch columns.
- VectorizedBatchUtil.swapColumnVector(
- bufferedBatch, i, overflowBatch, bufferedColumnMap[i]);
+ VectorizedBatchUtil.copyNonSelectedColumnVector(
+ bufferedBatch, i, overflowBatch, bufferedColumnMap[i], size);
+ }
+ overflowBatch.size = size;
+ fillGroupResults(overflowBatch);
+ vecPTFOperator.forward(overflowBatch, null);
+ }
+
+ private void forwardSpilledBatches(VectorPTFOperator vecPTFOperator, VectorizedRowBatch lastBatch)
+ throws HiveException {
+
+ overflowBatch.reset();
+ copyPartitionAndOrderColumnsToOverflow(lastBatch);
+
+ long spillRowsRead = 0;
+ try {
+ VectorRowBytesContainer rowBytesContainer = getSpillRowBytesContainer();
+ rowBytesContainer.prepareForReading();
+
+ while (rowBytesContainer.readNext()) {
+
+ byte[] bytes = rowBytesContainer.currentBytes();
+ int offset = rowBytesContainer.currentOffset();
+ int length = rowBytesContainer.currentLength();
+
+ bufferedBatchVectorDeserializeRow.setBytes(bytes, offset, length);
+ try {
+ bufferedBatchVectorDeserializeRow.deserialize(overflowBatch, overflowBatch.size);
+ } catch (Exception e) {
+ throw new HiveException(
+ "\nDeserializeRead detail: " +
+ bufferedBatchVectorDeserializeRow.getDetailedReadPositionString(),
+ e);
+ }
+ overflowBatch.size++;
+ spillRowsRead++;
+
+ if (overflowBatch.size == VectorizedRowBatch.DEFAULT_SIZE) {
+
+ fillGroupResults(overflowBatch);
+ vecPTFOperator.forward(overflowBatch, null);
- overflowBatch.size = bufferedBatch.size;
- fillGroupResults(overflowBatch);
- vecPTFOperator.forward(overflowBatch, null);
+ overflowBatch.reset();
+ copyPartitionAndOrderColumnsToOverflow(lastBatch);
+ }
+ }
+ // Process the row batch that has less than DEFAULT_SIZE rows
+ if (overflowBatch.size > 0) {
+
+ fillGroupResults(overflowBatch);
+ vecPTFOperator.forward(overflowBatch, null);
+
+ overflowBatch.reset();
+ copyPartitionAndOrderColumnsToOverflow(lastBatch);
+ }
+ Preconditions.checkState(spillRowsRead == spillRowCount);
+
+ // For now, throw away file.
+ releaseSpillRowBytesContainer();
+
+ } catch (Exception e) {
+ throw new HiveException(e);
+ }
+ }
+
+ private void copyPartitionAndOrderColumnsToOverflow(VectorizedRowBatch lastBatch) {
+
+ // Set partition and order columns in overflowBatch.
+ // We can set by ref since our last batch is held by us.
+ final int keyInputColumnCount = keyInputColumnMap.length;
+ for (int i = 0; i < keyInputColumnCount; i++) {
+ final int keyColumnNum = keyInputColumnMap[i];
+ Preconditions.checkState(overflowBatch.cols[keyColumnNum] != null);
+ VectorizedBatchUtil.copyRepeatingColumn(
+ lastBatch, keyColumnNum, overflowBatch, keyColumnNum, /* setByValue */ false);
}
}
public void fillGroupResultsAndForward(VectorPTFOperator vecPTFOperator,
VectorizedRowBatch lastBatch) throws HiveException {
- if (currentBufferedBatchCount > 0) {
-
- // Set partition and order columns in overflowBatch.
- // We can set by ref since our last batch is held by us.
- final int keyInputColumnCount = keyInputColumnMap.length;
- for (int i = 0; i < keyInputColumnCount; i++) {
- VectorizedBatchUtil.copyRepeatingColumn(lastBatch, i, overflowBatch, i, /* setByValue */ false);
- }
+ if (didSpillToDisk) {
+ forwardSpilledBatches(vecPTFOperator, lastBatch);
+ didSpillToDisk = false;
+ }
+ if (currentBufferedBatchCount > 0) {
+ overflowBatch.reset();
+ copyPartitionAndOrderColumnsToOverflow(lastBatch);
for (int i = 0; i < currentBufferedBatchCount; i++) {
forwardBufferedBatches(vecPTFOperator, i);
}
@@ -167,8 +320,8 @@ public class VectorPTFGroupBatches {
int originalProjectionSize = lastBatch.projectionSize;
// Project with the output of our operator.
- lastBatch.projectionSize = outputColumnMap.length;
- lastBatch.projectedColumns = outputColumnMap;
+ lastBatch.projectionSize = outputProjectionColumnMap.length;
+ lastBatch.projectedColumns = outputProjectionColumnMap;
vecPTFOperator.forward(lastBatch, null);
@@ -197,20 +350,60 @@ public class VectorPTFGroupBatches {
public void bufferGroupBatch(VectorizedRowBatch batch) throws HiveException {
- final int bufferedColumnCount = bufferedColumnMap.length;
- if (allocatedBufferedBatchCount <= currentBufferedBatchCount) {
- VectorizedRowBatch newBatch = newBufferedBatch(batch);
- bufferedBatches.add(newBatch);
- allocatedBufferedBatchCount++;
- }
+ try {
+ // When we've buffered the max allowed, spill the oldest one to make space.
+ if (currentBufferedBatchCount >= spillLimitBufferedBatchCount) {
- VectorizedRowBatch bufferedBatch = bufferedBatches.get(currentBufferedBatchCount++);
+ VectorRowBytesContainer rowBytesContainer = getSpillRowBytesContainer();
- for (int i = 0; i < bufferedColumnCount; i++) {
- VectorizedBatchUtil.swapColumnVector(
- batch, bufferedColumnMap[i], bufferedBatch, i);
- }
+ if (!didSpillToDisk) {
+ // UNDONE: Don't reuse for now.
+ // rowBytesContainer.resetWrite();
+ didSpillToDisk = true;
+ spillRowCount = 0;
+ }
+
+ // Grab the oldest in-memory buffered batch and dump it to disk.
+ VectorizedRowBatch oldestBufferedBatch = bufferedBatches.remove(0);
+
+ final boolean selectedInUse = oldestBufferedBatch.selectedInUse;
+ int[] selected = oldestBufferedBatch.selected;
+ final int size = oldestBufferedBatch.size;
+ for (int logicalIndex = 0; logicalIndex < size; logicalIndex++) {
+ final int batchIndex = (selectedInUse ? selected[logicalIndex] : logicalIndex);
+
+ Output output = rowBytesContainer.getOuputForRowBytes();
+ bufferedBatchVectorSerializeRow.setOutputAppend(output);
+ bufferedBatchVectorSerializeRow.serializeWrite(oldestBufferedBatch, batchIndex);
+ rowBytesContainer.finishRow();
+ spillRowCount++;
+ }
- bufferedBatch.size = batch.size;
+ // Put now available buffered batch at end.
+ oldestBufferedBatch.reset();
+ bufferedBatches.add(oldestBufferedBatch);
+ currentBufferedBatchCount--;
+ }
+
+ final int bufferedColumnCount = bufferedColumnMap.length;
+ if (allocatedBufferedBatchCount <= currentBufferedBatchCount) {
+ VectorizedRowBatch newBatch = newBufferedBatch(batch);
+ bufferedBatches.add(newBatch);
+ allocatedBufferedBatchCount++;
+ }
+
+ VectorizedRowBatch bufferedBatch = bufferedBatches.get(currentBufferedBatchCount++);
+
+ // Copy critical columns.
+ final int size = batch.size;
+ for (int i = 0; i < bufferedColumnCount; i++) {
+ VectorizedBatchUtil.copyNonSelectedColumnVector(
+ batch, bufferedColumnMap[i], bufferedBatch, i, size);
+ }
+
+ bufferedBatch.size = size;
+ } catch (IOException e) {
+ throw new HiveException(e);
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java
index 7522624..0d72ba8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java
@@ -97,10 +97,9 @@ public class VectorPTFOperator extends Operator<PTFDesc>
* PTF vector expressions.
*/
- // This is map of which vectorized row batch columns are the input columns and the group value
- // (aggregation) output columns.
- // And, their types.
- private int[] outputColumnMap;
+ private TypeInfo[] reducerBatchTypeInfos;
+
+ private int[] outputProjectionColumnMap;
private String[] outputColumnNames;
private TypeInfo[] outputTypeInfos;
@@ -135,7 +134,7 @@ public class VectorPTFOperator extends Operator<PTFDesc>
private transient VectorPTFEvaluatorBase[] evaluators;
- private transient int[] streamingColumnMap;
+ private transient int[] streamingEvaluatorNums;
private transient boolean allEvaluatorsAreStreaming;
@@ -179,11 +178,13 @@ public class VectorPTFOperator extends Operator<PTFDesc>
vectorPTFInfo = vectorDesc.getVectorPTFInfo();
this.vContext = vContext;
+ reducerBatchTypeInfos = vectorDesc.getReducerBatchTypeInfos();
+
isPartitionOrderBy = vectorDesc.getIsPartitionOrderBy();
outputColumnNames = vectorDesc.getOutputColumnNames();
outputTypeInfos = vectorDesc.getOutputTypeInfos();
- outputColumnMap = vectorPTFInfo.getOutputColumnMap();
+ outputProjectionColumnMap = vectorPTFInfo.getOutputColumnMap();
/*
* Create a new vectorization context to create a new projection, but keep
@@ -222,7 +223,7 @@ public class VectorPTFOperator extends Operator<PTFDesc>
final int count = outputColumnNames.length;
for (int i = 0; i < count; ++i) {
String columnName = outputColumnNames[i];
- int outputColumn = outputColumnMap[i];
+ int outputColumn = outputProjectionColumnMap[i];
vOutContext.addProjectionColumn(columnName, outputColumn);
}
}
@@ -255,8 +256,8 @@ public class VectorPTFOperator extends Operator<PTFDesc>
overflowBatch = new VectorizedRowBatch(totalNumColumns);
// First, just allocate just the output columns we will be using.
- for (int i = 0; i < outputColumnMap.length; i++) {
- int outputColumn = outputColumnMap[i];
+ for (int i = 0; i < outputProjectionColumnMap.length; i++) {
+ int outputColumn = outputProjectionColumnMap[i];
String typeName = outputTypeInfos[i].getTypeName();
allocateOverflowBatchColumnVector(overflowBatch, outputColumn, typeName);
}
@@ -267,8 +268,8 @@ public class VectorPTFOperator extends Operator<PTFDesc>
allocateOverflowBatchColumnVector(overflowBatch, outputColumn++, typeName);
}
- overflowBatch.projectedColumns = outputColumnMap;
- overflowBatch.projectionSize = outputColumnMap.length;
+ overflowBatch.projectedColumns = outputProjectionColumnMap;
+ overflowBatch.projectionSize = outputProjectionColumnMap.length;
overflowBatch.reset();
@@ -311,18 +312,26 @@ public class VectorPTFOperator extends Operator<PTFDesc>
evaluators = VectorPTFDesc.getEvaluators(vectorDesc, vectorPTFInfo);
- streamingColumnMap = VectorPTFDesc.getStreamingColumnMap(evaluators);
+ streamingEvaluatorNums = VectorPTFDesc.getStreamingEvaluatorNums(evaluators);
- allEvaluatorsAreStreaming = (streamingColumnMap.length == evaluatorCount);
+ allEvaluatorsAreStreaming = (streamingEvaluatorNums.length == evaluatorCount);
/*
* Setup the overflow batch.
*/
overflowBatch = setupOverflowBatch();
- groupBatches = new VectorPTFGroupBatches();
+ groupBatches = new VectorPTFGroupBatches(
+ hconf, vectorDesc.getVectorizedPTFMaxMemoryBufferingBatchCount());
groupBatches.init(
- evaluators, outputColumnMap, keyInputColumnMap, nonKeyInputColumnMap, streamingColumnMap, overflowBatch);
+ reducerBatchTypeInfos,
+ evaluators,
+ outputProjectionColumnMap,
+ outputTypeInfos,
+ keyInputColumnMap,
+ nonKeyInputColumnMap,
+ streamingEvaluatorNums,
+ overflowBatch);
isFirstPartition = true;
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/rowbytescontainer/VectorRowBytesContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/rowbytescontainer/VectorRowBytesContainer.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/rowbytescontainer/VectorRowBytesContainer.java
new file mode 100644
index 0000000..4b801cc
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/rowbytescontainer/VectorRowBytesContainer.java
@@ -0,0 +1,331 @@
+/**
+ * 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.hadoop.hive.ql.exec.vector.rowbytescontainer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.hive.common.FileUtils;
+import org.apache.hadoop.hive.serde2.ByteStream.Output;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+
+/**
+ * An eager bytes container that puts row bytes to an output stream.
+ */
+public class VectorRowBytesContainer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(VectorRowBytesContainer.class);
+
+ private File parentDir;
+ private File tmpFile;
+
+ // We buffer in a org.apache.hadoop.hive.serde2.ByteStream.Output since that is what
+ // is used by VectorSerializeRow / SerializeWrite. Periodically, we flush this buffer
+ // to disk.
+ private Output output;
+ private int rowBeginPos;
+ private static final int OUTPUT_SIZE = 4096;
+ private static final int THRESHOLD = 8 * (OUTPUT_SIZE / 10);
+ private static final int INPUT_SIZE = 4096;
+
+ private FileOutputStream fileOutputStream;
+
+ private boolean isOpen;
+
+ private byte[] readBuffer;
+ private byte[] largeRowBuffer;
+ private int readOffset;
+ private int readLength;
+
+ private int readNextCount;
+ private int readNextIndex;
+
+ private static final int MAX_READS = 256;
+ private byte[][] readNextBytes;
+ private int readNextOffsets[];
+ private int readNextLengths[];
+
+ private byte[] currentBytes;
+ private int currentOffset;
+ private int currentLength;
+
+ private long totalWriteLength;
+ private long totalReadLength;
+
+ private FileInputStream fileInputStream;
+
+ private final String spillLocalDirs;
+
+ public VectorRowBytesContainer(String spillLocalDirs) {
+ output = new Output();
+ readBuffer = new byte[INPUT_SIZE];
+ readNextBytes = new byte[MAX_READS][];
+ readNextOffsets = new int[MAX_READS];
+ readNextLengths = new int[MAX_READS];
+ isOpen = false;
+ totalWriteLength = 0;
+ totalReadLength = 0;
+ this.spillLocalDirs = spillLocalDirs;
+ }
+
+ private void setupOutputFileStreams() throws IOException {
+ parentDir = FileUtils.createLocalDirsTempFile(spillLocalDirs, "bytes-container", "", true);
+ parentDir.deleteOnExit();
+ tmpFile = File.createTempFile("BytesContainer", ".tmp", parentDir);
+ LOG.debug("BytesContainer created temp file " + tmpFile.getAbsolutePath());
+ tmpFile.deleteOnExit();
+
+ fileOutputStream = new FileOutputStream(tmpFile);
+ }
+
+ private void initFile() {
+ try {
+ setupOutputFileStreams();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to create temporary output file on disk", e);
+ }
+ }
+
+ public Output getOuputForRowBytes() {
+ if (!isOpen) {
+ initFile();
+ isOpen = true;
+ }
+ // Reserve space for the int length.
+ output.reserve(4);
+ rowBeginPos = output.getLength();
+ return output;
+ }
+
+ public void finishRow() throws IOException {
+ int length = output.getLength() - rowBeginPos;
+ output.writeInt(rowBeginPos - 4, length);
+ if (output.getLength() > THRESHOLD) {
+ fileOutputStream.write(output.getData(), 0, output.getLength());
+ totalWriteLength += output.getLength();
+ output.reset();
+ }
+ }
+
+ public void prepareForReading() throws IOException {
+ if (!isOpen) {
+ return;
+ }
+ if (output.getLength() > 0) {
+ fileOutputStream.write(output.getData(), 0, output.getLength());
+ totalWriteLength += output.getLength();
+ fileOutputStream.flush();
+ output.reset();
+ }
+ if (fileInputStream != null) {
+ fileInputStream.close();
+ }
+ fileInputStream = new FileInputStream(tmpFile);
+ readNextIndex = 0;
+ readNextCount = 0;
+ }
+
+ private int readInt() {
+ int value = (((readBuffer[readOffset] & 0xFF) << 24) |
+ ((readBuffer[readOffset + 1] & 0xFF) << 16) |
+ ((readBuffer[readOffset + 2] & 0xFF) << 8) |
+ ((readBuffer[readOffset + 3] & 0xFF)));
+ readOffset += 4;
+ return value;
+ }
+
+ // Call when nextReadIndex == nextReadCount.
+ private void bufferedRead() throws IOException {
+
+ // Reset for reading.
+ readNextIndex = 0;
+
+ // Reset for filling.
+ readNextCount = 0;
+
+ if (readOffset < readLength) {
+ // Move unprocessed remainder to beginning of buffer.
+ int unprocessLength = readLength - readOffset;
+ System.arraycopy(readBuffer, readOffset, readBuffer, 0, unprocessLength);
+
+ int maxReadLength = readBuffer.length - unprocessLength;
+ int partialReadLength = fileInputStream.read(readBuffer, unprocessLength, maxReadLength);
+ if (partialReadLength == -1) {
+ partialReadLength = 0;
+ }
+ totalReadLength += partialReadLength;
+ readLength = unprocessLength + partialReadLength;
+ readOffset = 0;
+ } else {
+ readOffset = 0;
+ readLength = fileInputStream.read(readBuffer, 0, readBuffer.length);
+ if (readLength == -1) {
+ readLength = 0;
+ }
+ totalReadLength += readLength;
+ }
+ if (readLength == 0) {
+ return;
+ }
+ if (readLength < 0) {
+ throw new IOException("Negative read length");
+ }
+
+ // Get length word.
+ if (readLength < 4) {
+ throw new IOException("Expecting 4 byte length");
+ }
+
+ while (true) {
+ // Use Input class to read length.
+ int saveReadOffset = readOffset;
+ int rowLength = readInt();
+ if (rowLength < 0) {
+ throw new IOException("Negative row length");
+ }
+ int remainingLength = readLength - readOffset;
+ if (remainingLength < rowLength) {
+ if (readNextCount > 0) {
+ // Leave this one for the next round.
+ readOffset = saveReadOffset;
+ break;
+ }
+
+ // Buffer needed to bridge.
+ if (largeRowBuffer == null || largeRowBuffer.length < rowLength) {
+ int newLargeBufferLength = Math.max(Integer.highestOneBit(rowLength) << 1, INPUT_SIZE);
+ largeRowBuffer = new byte[newLargeBufferLength];
+ }
+ System.arraycopy(readBuffer, readOffset, largeRowBuffer, 0, remainingLength);
+ int expectedPartialLength = rowLength - remainingLength;
+ int partialReadLength = fileInputStream.read(largeRowBuffer, remainingLength, expectedPartialLength);
+ if (partialReadLength == -1) {
+ throw new IOException("Unexpected EOF (total write length " + totalWriteLength +
+ ", total read length " + totalReadLength + ", read length " +
+ expectedPartialLength + ")");
+ }
+
+ if (expectedPartialLength != partialReadLength) {
+ throw new IOException("Unable to read a complete row of length " + rowLength +
+ " (total write length " + totalWriteLength +
+ ", total read length " + totalReadLength + ", read length " +
+ expectedPartialLength + ", actual length " + partialReadLength + ")");
+ }
+ totalReadLength += partialReadLength;
+
+ readNextBytes[readNextCount] = largeRowBuffer;
+ readNextOffsets[readNextCount] = 0;
+ readNextLengths[readNextCount] = rowLength;
+
+ // Indicate we used the last row's bytes for large buffer.
+ readOffset = readLength;
+ readNextCount++;
+ break;
+ }
+
+ readNextBytes[readNextCount] = readBuffer;
+ readNextOffsets[readNextCount] = readOffset;
+ readNextLengths[readNextCount] = rowLength;
+ readOffset += rowLength;
+ readNextCount++;
+
+ if (readNextCount >= readNextBytes.length){
+ break;
+ }
+ if (readLength - readOffset < 4) {
+ // Handle in next round.
+ break;
+ }
+ }
+ }
+
+ public boolean readNext() throws IOException {
+ if (!isOpen) {
+ return false;
+ }
+ if (readNextIndex >= readNextCount) {
+ bufferedRead();
+ // Any more left?
+ if (readNextIndex >= readNextCount) {
+ return false;
+ }
+ }
+
+ currentBytes = readNextBytes[readNextIndex];
+ currentOffset = readNextOffsets[readNextIndex];
+ currentLength = readNextLengths[readNextIndex];
+
+ readNextIndex++;
+ return true;
+ }
+
+ public byte[] currentBytes() {
+ return currentBytes;
+ }
+
+ public int currentOffset() {
+ return currentOffset;
+ }
+
+ public int currentLength() {
+ return currentLength;
+ }
+
+ public void resetWrite() throws IOException {
+ if (!isOpen) {
+ return;
+ }
+
+ // Truncate by re-opening FileOutputStream.
+ fileOutputStream.close();
+ fileOutputStream = new FileOutputStream(tmpFile);
+ }
+
+ public void clear() {
+ if (fileInputStream != null) {
+ try {
+ fileInputStream.close();
+ } catch (Throwable ignored) {
+ }
+ fileInputStream = null;
+ }
+ if (fileOutputStream != null) {
+ try {
+ fileOutputStream.close();
+ } catch (Throwable ignored) {
+ }
+ fileOutputStream = null;
+ }
+
+ if (parentDir != null) {
+ try {
+ FileUtil.fullyDelete(parentDir);
+ } catch (Throwable ignored) {
+ }
+ }
+ parentDir = null;
+ tmpFile = null;
+ isOpen = false;
+ totalWriteLength = 0;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java
index a02863b..28400c7 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java
@@ -294,6 +294,8 @@ public class Vectorizer implements PhysicalPlanResolver {
private boolean isVectorizationGroupByComplexTypesEnabled;
private boolean isVectorizedRowIdentifierEnabled;
private Collection<Class<?>> rowDeserializeInputFormatExcludes;
+ private int vectorizedPTFMaxMemoryBufferingBatchCount;
+ private int vectorizedTestingReducerBatchSize;
private boolean isSchemaEvolution;
@@ -1278,6 +1280,7 @@ public class Vectorizer implements PhysicalPlanResolver {
vectorTaskColumnInfo.assume();
reduceWork.setVectorizedVertexNum(++vectorizedVertexNum);
+ reduceWork.setVectorizedTestingReducerBatchSize(vectorizedTestingReducerBatchSize);
boolean ret;
try {
@@ -1392,7 +1395,8 @@ public class Vectorizer implements PhysicalPlanResolver {
}
// Now check the reduce operator tree.
Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
- ReduceWorkValidationNodeProcessor vnp = new ReduceWorkValidationNodeProcessor();
+ ReduceWorkValidationNodeProcessor vnp =
+ new ReduceWorkValidationNodeProcessor(vectorTaskColumnInfo);
addReduceWorkRules(opRules, vnp);
Dispatcher disp = new DefaultRuleDispatcher(vnp, opRules, null);
GraphWalker ogw = new DefaultGraphWalker(disp);
@@ -1542,6 +1546,14 @@ public class Vectorizer implements PhysicalPlanResolver {
class ReduceWorkValidationNodeProcessor implements NodeProcessor {
+ private final VectorTaskColumnInfo vectorTaskColumnInfo;
+ private final TypeInfo[] reducerBatchTypeInfos;
+
+ public ReduceWorkValidationNodeProcessor(VectorTaskColumnInfo vectorTaskColumnInfo) {
+ this.vectorTaskColumnInfo = vectorTaskColumnInfo;
+ reducerBatchTypeInfos = vectorTaskColumnInfo.allTypeInfos.toArray(new TypeInfo[0]);
+ }
+
// Children of Vectorized GROUPBY that outputs rows instead of vectorized row batchs.
protected final Set<Operator<? extends OperatorDesc>> nonVectorizedOps =
new HashSet<Operator<? extends OperatorDesc>>();
@@ -1563,7 +1575,7 @@ public class Vectorizer implements PhysicalPlanResolver {
return new Boolean(true);
}
currentOperator = op;
- boolean ret = validateReduceWorkOperator(op);
+ boolean ret = validateReduceWorkOperator(op, reducerBatchTypeInfos);
if (!ret) {
return new Boolean(false);
}
@@ -1865,6 +1877,13 @@ public class Vectorizer implements PhysicalPlanResolver {
HiveConf.getBoolVar(hiveConf,
HiveConf.ConfVars.HIVE_VECTORIZATION_ROW_IDENTIFIER_ENABLED);
+ vectorizedPTFMaxMemoryBufferingBatchCount =
+ HiveConf.getIntVar(hiveConf,
+ HiveConf.ConfVars.HIVE_VECTORIZATION_PTF_MAX_MEMORY_BUFFERING_BATCH_COUNT);
+ vectorizedTestingReducerBatchSize =
+ HiveConf.getIntVar(hiveConf,
+ HiveConf.ConfVars.HIVE_VECTORIZATION_TESTING_REDUCER_BATCH_SIZE);
+
isSchemaEvolution =
HiveConf.getBoolVar(hiveConf,
HiveConf.ConfVars.HIVE_SCHEMA_EVOLUTION);
@@ -1951,7 +1970,8 @@ public class Vectorizer implements PhysicalPlanResolver {
return ret;
}
- boolean validateReduceWorkOperator(Operator<? extends OperatorDesc> op) {
+ boolean validateReduceWorkOperator(Operator<? extends OperatorDesc> op,
+ TypeInfo[] reducerBatchTypeInfos) {
boolean ret;
switch (op.getType()) {
case MAPJOIN:
@@ -1996,7 +2016,8 @@ public class Vectorizer implements PhysicalPlanResolver {
validateSparkHashTableSinkOperator((SparkHashTableSinkOperator) op);
break;
case PTF:
- ret = validatePTFOperator((PTFOperator) op);
+ // PTF needs the TypeInfo of the reducer batch.
+ ret = validatePTFOperator((PTFOperator) op, reducerBatchTypeInfos);
break;
default:
setOperatorNotSupported(op);
@@ -2287,7 +2308,7 @@ public class Vectorizer implements PhysicalPlanResolver {
return false;
}
- private boolean validatePTFOperator(PTFOperator op) {
+ private boolean validatePTFOperator(PTFOperator op, TypeInfo[] reducerBatchTypeInfos) {
if (!isPtfVectorizationEnabled) {
setNodeIssue("Vectorization of PTF is not enabled (" +
@@ -2323,7 +2344,8 @@ public class Vectorizer implements PhysicalPlanResolver {
VectorPTFDesc vectorPTFDesc = null;
try {
- vectorPTFDesc = createVectorPTFDesc(op, ptfDesc);
+ vectorPTFDesc = createVectorPTFDesc(
+ op, ptfDesc, reducerBatchTypeInfos, vectorizedPTFMaxMemoryBufferingBatchCount);
} catch (HiveException e) {
setOperatorIssue("exception: " + VectorizationContext.getStackTraceAsSingleLine(e));
return false;
@@ -3835,7 +3857,8 @@ public class Vectorizer implements PhysicalPlanResolver {
* VectorizationContext to lookup column names, etc.
*/
private static VectorPTFDesc createVectorPTFDesc(Operator<? extends OperatorDesc> ptfOp,
- PTFDesc ptfDesc) throws HiveException {
+ PTFDesc ptfDesc, TypeInfo[] reducerBatchTypeInfos,
+ int vectorizedPTFMaxMemoryBufferingBatchCount) throws HiveException {
PartitionedTableFunctionDef funcDef = ptfDesc.getFuncDef();
@@ -3849,6 +3872,8 @@ public class Vectorizer implements PhysicalPlanResolver {
/*
* Output columns.
*/
+
+ // Evaluator results are first.
String[] outputColumnNames = new String[outputSize];
TypeInfo[] outputTypeInfos = new TypeInfo[outputSize];
for (int i = 0; i < functionCount; i++) {
@@ -3857,6 +3882,8 @@ public class Vectorizer implements PhysicalPlanResolver {
outputColumnNames[i] = colInfo.getInternalName();
outputTypeInfos[i] = typeInfo;
}
+
+ // Followed by key and non-key input columns (some may be missing).
for (int i = functionCount; i < outputSize; i++) {
ColumnInfo colInfo = outputSignature.get(i);
outputColumnNames[i] = colInfo.getInternalName();
@@ -3904,6 +3931,8 @@ public class Vectorizer implements PhysicalPlanResolver {
VectorPTFDesc vectorPTFDesc = new VectorPTFDesc();
+ vectorPTFDesc.setReducerBatchTypeInfos(reducerBatchTypeInfos);
+
vectorPTFDesc.setIsPartitionOrderBy(isPartitionOrderBy);
vectorPTFDesc.setOrderExprNodeDescs(orderExprNodeDescs);
@@ -3916,19 +3945,22 @@ public class Vectorizer implements PhysicalPlanResolver {
vectorPTFDesc.setOutputColumnNames(outputColumnNames);
vectorPTFDesc.setOutputTypeInfos(outputTypeInfos);
+ vectorPTFDesc.setVectorizedPTFMaxMemoryBufferingBatchCount(
+ vectorizedPTFMaxMemoryBufferingBatchCount);
+
return vectorPTFDesc;
}
- private static void determineKeyAndNonKeyInputColumnMap(int[] outputColumnMap,
+ private static void determineKeyAndNonKeyInputColumnMap(int[] outputColumnProjectionMap,
boolean isPartitionOrderBy, int[] orderColumnMap, int[] partitionColumnMap,
int evaluatorCount, ArrayList<Integer> keyInputColumns,
ArrayList<Integer> nonKeyInputColumns) {
- final int outputSize = outputColumnMap.length;
+ final int outputSize = outputColumnProjectionMap.length;
final int orderKeyCount = orderColumnMap.length;
final int partitionKeyCount = (isPartitionOrderBy ? partitionColumnMap.length : 0);
for (int i = evaluatorCount; i < outputSize; i++) {
- final int nonEvalColumnNum = outputColumnMap[i];
+ final int nonEvalColumnNum = outputColumnProjectionMap[i];
boolean isKey = false;
for (int o = 0; o < orderKeyCount; o++) {
if (nonEvalColumnNum == orderColumnMap[o]) {
@@ -3957,7 +3989,8 @@ public class Vectorizer implements PhysicalPlanResolver {
* execution.
*/
private static VectorPTFInfo createVectorPTFInfo(Operator<? extends OperatorDesc> ptfOp,
- PTFDesc ptfDesc, VectorizationContext vContext) throws HiveException {
+ PTFDesc ptfDesc, VectorizationContext vContext)
+ throws HiveException {
PartitionedTableFunctionDef funcDef = ptfDesc.getFuncDef();
@@ -3978,17 +4011,21 @@ public class Vectorizer implements PhysicalPlanResolver {
/*
* Output columns.
*/
- int[] outputColumnMap = new int[outputSize];
+ int[] outputColumnProjectionMap = new int[outputSize];
+
+ // Evaluator results are first.
for (int i = 0; i < evaluatorCount; i++) {
ColumnInfo colInfo = outputSignature.get(i);
TypeInfo typeInfo = colInfo.getType();
final int outputColumnNum;
outputColumnNum = vContext.allocateScratchColumn(typeInfo);
- outputColumnMap[i] = outputColumnNum;
+ outputColumnProjectionMap[i] = outputColumnNum;
}
+
+ // Followed by key and non-key input columns (some may be missing).
for (int i = evaluatorCount; i < outputSize; i++) {
ColumnInfo colInfo = outputSignature.get(i);
- outputColumnMap[i] = vContext.getInputColumnIndex(colInfo.getInternalName());
+ outputColumnProjectionMap[i] = vContext.getInputColumnIndex(colInfo.getInternalName());
}
/*
@@ -4038,7 +4075,7 @@ public class Vectorizer implements PhysicalPlanResolver {
ArrayList<Integer> keyInputColumns = new ArrayList<Integer>();
ArrayList<Integer> nonKeyInputColumns = new ArrayList<Integer>();
- determineKeyAndNonKeyInputColumnMap(outputColumnMap, isPartitionOrderBy, orderColumnMap,
+ determineKeyAndNonKeyInputColumnMap(outputColumnProjectionMap, isPartitionOrderBy, orderColumnMap,
partitionColumnMap, evaluatorCount, keyInputColumns, nonKeyInputColumns);
int[] keyInputColumnMap = ArrayUtils.toPrimitive(keyInputColumns.toArray(new Integer[0]));
int[] nonKeyInputColumnMap = ArrayUtils.toPrimitive(nonKeyInputColumns.toArray(new Integer[0]));
@@ -4075,7 +4112,7 @@ public class Vectorizer implements PhysicalPlanResolver {
VectorPTFInfo vectorPTFInfo = new VectorPTFInfo();
- vectorPTFInfo.setOutputColumnMap(outputColumnMap);
+ vectorPTFInfo.setOutputColumnMap(outputColumnProjectionMap);
vectorPTFInfo.setPartitionColumnMap(partitionColumnMap);
vectorPTFInfo.setPartitionColumnVectorTypes(partitionColumnVectorTypes);
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java
index 0984df7..ce21add 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java
@@ -90,6 +90,7 @@ public abstract class BaseWork extends AbstractOperatorDesc {
private boolean usesVectorUDFAdaptor;
protected long vectorizedVertexNum;
+ protected int vectorizedTestingReducerBatchSize;
protected boolean llapMode = false;
protected boolean uberMode = false;
@@ -193,6 +194,14 @@ public abstract class BaseWork extends AbstractOperatorDesc {
return vectorizedVertexNum;
}
+ public void setVectorizedTestingReducerBatchSize(int vectorizedTestingReducerBatchSize) {
+ this.vectorizedTestingReducerBatchSize = vectorizedTestingReducerBatchSize;
+ }
+
+ public int getVectorizedTestingReducerBatchSize() {
+ return vectorizedTestingReducerBatchSize;
+ }
+
// -----------------------------------------------------------------------------------------------
public void setVectorizationExamined(boolean vectorizationExamined) {
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/java/org/apache/hadoop/hive/ql/plan/VectorPTFDesc.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/VectorPTFDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/VectorPTFDesc.java
index a2f4cbc..2021da1 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/VectorPTFDesc.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/VectorPTFDesc.java
@@ -99,6 +99,8 @@ public class VectorPTFDesc extends AbstractVectorDesc {
supportedFunctionNames.addAll(treeSet);
}
+ private TypeInfo[] reducerBatchTypeInfos;
+
private boolean isPartitionOrderBy;
private String[] evaluatorFunctionNames;
@@ -113,6 +115,8 @@ public class VectorPTFDesc extends AbstractVectorDesc {
private VectorPTFInfo vectorPTFInfo;
+ private int vectorizedPTFMaxMemoryBufferingBatchCount;
+
public VectorPTFDesc() {
isPartitionOrderBy = false;
@@ -124,6 +128,9 @@ public class VectorPTFDesc extends AbstractVectorDesc {
outputColumnNames = null;
outputTypeInfos = null;
+
+ vectorizedPTFMaxMemoryBufferingBatchCount = -1;
+
}
// We provide this public method to help EXPLAIN VECTORIZATION show the evaluator classes.
@@ -274,16 +281,24 @@ public class VectorPTFDesc extends AbstractVectorDesc {
return evaluators;
}
- public static int[] getStreamingColumnMap(VectorPTFEvaluatorBase[] evaluators) {
+ public static int[] getStreamingEvaluatorNums(VectorPTFEvaluatorBase[] evaluators) {
final int evaluatorCount = evaluators.length;
- ArrayList<Integer> streamingColumns = new ArrayList<Integer>();
+ ArrayList<Integer> streamingEvaluatorNums = new ArrayList<Integer>();
for (int i = 0; i < evaluatorCount; i++) {
final VectorPTFEvaluatorBase evaluator = evaluators[i];
if (evaluator.streamsResult()) {
- streamingColumns.add(evaluator.getOutputColumnNum());
+ streamingEvaluatorNums.add(i);
}
}
- return ArrayUtils.toPrimitive(streamingColumns.toArray(new Integer[0]));
+ return ArrayUtils.toPrimitive(streamingEvaluatorNums.toArray(new Integer[0]));
+ }
+
+ public TypeInfo[] getReducerBatchTypeInfos() {
+ return reducerBatchTypeInfos;
+ }
+
+ public void setReducerBatchTypeInfos(TypeInfo[] reducerBatchTypeInfos) {
+ this.reducerBatchTypeInfos = reducerBatchTypeInfos;
}
public boolean getIsPartitionOrderBy() {
@@ -357,4 +372,14 @@ public class VectorPTFDesc extends AbstractVectorDesc {
public VectorPTFInfo getVectorPTFInfo() {
return vectorPTFInfo;
}
+
+ public void setVectorizedPTFMaxMemoryBufferingBatchCount(
+ int vectorizedPTFMaxMemoryBufferingBatchCount) {
+ this.vectorizedPTFMaxMemoryBufferingBatchCount = vectorizedPTFMaxMemoryBufferingBatchCount;
+ }
+
+ public int getVectorizedPTFMaxMemoryBufferingBatchCount() {
+ return vectorizedPTFMaxMemoryBufferingBatchCount;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/TestVectorRowBytesContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/TestVectorRowBytesContainer.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/TestVectorRowBytesContainer.java
new file mode 100644
index 0000000..d9e2dbd
--- /dev/null
+++ b/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/TestVectorRowBytesContainer.java
@@ -0,0 +1,76 @@
+/**
+ * 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.hadoop.hive.ql.exec.vector;
+
+import java.util.Random;
+
+import org.apache.hadoop.hive.ql.exec.vector.mapjoin.fast.RandomByteArrayStream;
+import org.apache.hadoop.hive.ql.exec.vector.rowbytescontainer.VectorRowBytesContainer;
+import org.apache.hadoop.hive.serde2.ByteStream.Output;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class TestVectorRowBytesContainer {
+
+ public void doFillReplay(Random random, int maxCount) throws Exception {
+
+ RandomByteArrayStream randomByteArrayStream = new RandomByteArrayStream(random);
+ VectorRowBytesContainer vectorMapJoinRowBytesContainer =
+ new VectorRowBytesContainer(null);
+
+ int count = Math.min(maxCount, random.nextInt(500));
+ for (int i = 0; i < count; i++) {
+ byte[] bytes = randomByteArrayStream.next();
+ Output output = vectorMapJoinRowBytesContainer.getOuputForRowBytes();
+ output.write(bytes);
+ vectorMapJoinRowBytesContainer.finishRow();
+ }
+ vectorMapJoinRowBytesContainer.prepareForReading();
+
+ for (int i = 0; i < count; i++) {
+ if (!vectorMapJoinRowBytesContainer.readNext()) {
+ assertTrue(false);
+ }
+ byte[] readBytes = vectorMapJoinRowBytesContainer.currentBytes();
+ int readOffset = vectorMapJoinRowBytesContainer.currentOffset();
+ int readLength = vectorMapJoinRowBytesContainer.currentLength();
+ byte[] expectedBytes = randomByteArrayStream.get(i);
+ if (readLength != expectedBytes.length) {
+ assertTrue(false);
+ }
+ for (int j = 0; j < readLength; j++) {
+ byte readByte = readBytes[readOffset + j];
+ byte expectedByte = expectedBytes[j];
+ if (readByte != expectedByte) {
+ assertTrue(false);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testFillReplay() throws Exception {
+ Random random = new Random(47496);
+
+ for (int i = 0; i < 10; i++) {
+ doFillReplay(random, 1 << i);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/c129bb95/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/mapjoin/fast/TestVectorMapJoinRowBytesContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/mapjoin/fast/TestVectorMapJoinRowBytesContainer.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/mapjoin/fast/TestVectorMapJoinRowBytesContainer.java
deleted file mode 100644
index afe4e70..0000000
--- a/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/mapjoin/fast/TestVectorMapJoinRowBytesContainer.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/**
- * 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.hadoop.hive.ql.exec.vector.mapjoin.fast;
-
-import java.util.Random;
-
-import org.apache.hadoop.hive.ql.exec.vector.mapjoin.VectorMapJoinRowBytesContainer;
-import org.apache.hadoop.hive.serde2.ByteStream.Output;
-import org.junit.Test;
-
-import static org.junit.Assert.*;
-
-public class TestVectorMapJoinRowBytesContainer {
-
- public void doFillReplay(Random random, int maxCount) throws Exception {
-
- RandomByteArrayStream randomByteArrayStream = new RandomByteArrayStream(random);
- VectorMapJoinRowBytesContainer vectorMapJoinRowBytesContainer =
- new VectorMapJoinRowBytesContainer(null);
-
- int count = Math.min(maxCount, random.nextInt(500));
- for (int i = 0; i < count; i++) {
- byte[] bytes = randomByteArrayStream.next();
- Output output = vectorMapJoinRowBytesContainer.getOuputForRowBytes();
- output.write(bytes);
- vectorMapJoinRowBytesContainer.finishRow();
- }
- vectorMapJoinRowBytesContainer.prepareForReading();
-
- for (int i = 0; i < count; i++) {
- if (!vectorMapJoinRowBytesContainer.readNext()) {
- assertTrue(false);
- }
- byte[] readBytes = vectorMapJoinRowBytesContainer.currentBytes();
- int readOffset = vectorMapJoinRowBytesContainer.currentOffset();
- int readLength = vectorMapJoinRowBytesContainer.currentLength();
- byte[] expectedBytes = randomByteArrayStream.get(i);
- if (readLength != expectedBytes.length) {
- assertTrue(false);
- }
- for (int j = 0; j < readLength; j++) {
- byte readByte = readBytes[readOffset + j];
- byte expectedByte = expectedBytes[j];
- if (readByte != expectedByte) {
- assertTrue(false);
- }
- }
- }
- }
-
- @Test
- public void testFillReplay() throws Exception {
- Random random = new Random(47496);
-
- for (int i = 0; i < 10; i++) {
- doFillReplay(random, 1 << i);
- }
- }
-}
\ No newline at end of file