You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2015/04/23 00:51:06 UTC
[16/17] hive git commit: HIVE-9824 : LLAP: Native Vectorization of
Map Join (Matt McCline, reviewed by Sergey Shelukhin, Vikram Dixit K)
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/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 ceb18c6..e2934e3 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
@@ -582,6 +582,7 @@ public class VectorSerializeRow {
}
public void init(List<String> typeNames, int[] columnMap) throws HiveException {
+
writers = new Writer[typeNames.size()];
for (int i = 0; i < typeNames.size(); i++) {
String typeName = typeNames.get(i);
@@ -593,6 +594,7 @@ public class VectorSerializeRow {
}
public void init(List<String> typeNames) throws HiveException {
+
writers = new Writer[typeNames.size()];
for (int i = 0; i < typeNames.size(); i++) {
String typeName = typeNames.get(i);
@@ -602,6 +604,17 @@ public class VectorSerializeRow {
}
}
+ public void init(PrimitiveTypeInfo[] primitiveTypeInfos, List<Integer> columnMap)
+ throws HiveException {
+
+ writers = new Writer[primitiveTypeInfos.length];
+ for (int i = 0; i < primitiveTypeInfos.length; i++) {
+ int columnIndex = columnMap.get(i);
+ Writer writer = createWriter(primitiveTypeInfos[i], columnIndex);
+ writers[i] = writer;
+ }
+ }
+
public int getCount() {
return writers.length;
}
@@ -610,6 +623,10 @@ public class VectorSerializeRow {
serializeWrite.set(output);
}
+ public void setOutputAppend(Output output) {
+ serializeWrite.setAppend(output);
+ }
+
/*
* Note that when serializing a row, the logical mapping using selected in use has already
* been performed. batchIndex is the actual index of the row.
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRowNoNulls.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRowNoNulls.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRowNoNulls.java
index 4a9fdcd..1363004 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRowNoNulls.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRowNoNulls.java
@@ -355,6 +355,7 @@ public class VectorSerializeRowNoNulls {
}
public void init(List<String> typeNames, int[] columnMap) throws HiveException {
+
writers = new Writer[typeNames.size()];
for (int i = 0; i < typeNames.size(); i++) {
String typeName = typeNames.get(i);
@@ -366,6 +367,7 @@ public class VectorSerializeRowNoNulls {
}
public void init(List<String> typeNames) throws HiveException {
+
writers = new Writer[typeNames.size()];
for (int i = 0; i < typeNames.size(); i++) {
String typeName = typeNames.get(i);
@@ -375,6 +377,17 @@ public class VectorSerializeRowNoNulls {
}
}
+ public void init(PrimitiveTypeInfo[] primitiveTypeInfos, List<Integer> columnMap)
+ throws HiveException {
+
+ writers = new Writer[primitiveTypeInfos.length];
+ for (int i = 0; i < primitiveTypeInfos.length; i++) {
+ int columnIndex = columnMap.get(i);
+ Writer writer = createWriter(primitiveTypeInfos[i], columnIndex);
+ writers[i] = writer;
+ }
+ }
+
public int getCount() {
return writers.length;
}
@@ -383,6 +396,10 @@ public class VectorSerializeRowNoNulls {
serializeWrite.set(output);
}
+ public void setOutputAppend(Output output) {
+ serializeWrite.setAppend(output);
+ }
+
/*
* Note that when serializing a row, the logical mapping using selected in use has already
* been performed. batchIndex is the actual index of the row.
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java
index 16b8294..abce606 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java
@@ -286,8 +286,9 @@ public class VectorizationContext {
return projectionColumnMap.get(name);
}
- protected int getInputColumnIndex(ExprNodeColumnDesc colExpr) {
- return projectionColumnMap.get(colExpr.getColumn());
+ protected int getInputColumnIndex(ExprNodeColumnDesc colExpr) throws HiveException {
+ // Call the regular method since it does error checking.
+ return getInputColumnIndex(colExpr.getColumn());
}
private static class OutputColumnManager {
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/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 e9cf6b6..dcea8ae 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
@@ -21,6 +21,7 @@ package org.apache.hadoop.hive.ql.exec.vector;
import java.io.IOException;
import java.sql.Timestamp;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
@@ -581,6 +582,52 @@ public class VectorizedBatchUtil {
return result;
}
+ public static PrimitiveTypeInfo[] primitiveTypeInfosFromTypeNames(
+ String[] typeNames) throws HiveException {
+
+ PrimitiveTypeInfo[] result = new PrimitiveTypeInfo[typeNames.length];
+
+ for(int i = 0; i < typeNames.length; i++) {
+ TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(typeNames[i]);
+ result[i] = (PrimitiveTypeInfo) typeInfo;
+ }
+ return result;
+ }
+
+ /**
+ * Make a new (scratch) batch, which is exactly "like" the batch provided, except that it's empty
+ * @param batch the batch to imitate
+ * @return the new batch
+ * @throws HiveException
+ */
+ public static VectorizedRowBatch makeLike(VectorizedRowBatch batch) throws HiveException {
+ VectorizedRowBatch newBatch = new VectorizedRowBatch(batch.numCols);
+ for (int i = 0; i < batch.numCols; i++) {
+ ColumnVector colVector = batch.cols[i];
+ if (colVector != null) {
+ ColumnVector newColVector;
+ if (colVector instanceof LongColumnVector) {
+ newColVector = new LongColumnVector();
+ } else if (colVector instanceof DoubleColumnVector) {
+ newColVector = new DoubleColumnVector();
+ } else if (colVector instanceof BytesColumnVector) {
+ newColVector = new BytesColumnVector();
+ } else if (colVector instanceof DecimalColumnVector) {
+ DecimalColumnVector decColVector = (DecimalColumnVector) colVector;
+ newColVector = new DecimalColumnVector(decColVector.precision, decColVector.scale);
+ } else {
+ throw new HiveException("Column vector class " + colVector.getClass().getName() +
+ " is not supported!");
+ }
+ newBatch.cols[i] = newColVector;
+ newBatch.cols[i].init();
+ }
+ }
+ newBatch.projectedColumns = Arrays.copyOf(batch.projectedColumns, batch.projectedColumns.length);
+ newBatch.projectionSize = batch.projectionSize;
+ newBatch.reset();
+ return newBatch;
+ }
public static String displayBytes(byte[] bytes, int start, int length) {
StringBuilder sb = new StringBuilder();
@@ -632,7 +679,7 @@ public class VectorizedBatchUtil {
}
sb.append(" ");
}
- System.out.println(sb.toString());
+ LOG.info(sb.toString());
}
public static void debugDisplayBatch(VectorizedRowBatch batch, String prefix) throws HiveException {
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java
index 77c3652..68c4df7 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java
@@ -670,7 +670,7 @@ public class VectorizedRowBatchCtx {
type.equalsIgnoreCase(serdeConstants.INTERVAL_DAY_TIME_TYPE_NAME)) {
return new LongColumnVector(defaultSize);
} else {
- throw new Error("Cannot allocate vector column for " + type);
+ throw new RuntimeException("Cannot allocate vector column for " + type);
}
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/IdentityExpression.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/IdentityExpression.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/IdentityExpression.java
index 56e0d39..402d0f8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/IdentityExpression.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/IdentityExpression.java
@@ -45,6 +45,15 @@ public class IdentityExpression extends VectorExpression {
}
}
+ public static boolean isColumnOnly(VectorExpression ve) {
+ if (ve instanceof IdentityExpression) {
+ VectorExpression identityExpression = (IdentityExpression) ve;
+ return (identityExpression.childExpressions == null);
+ } else {
+ return false;
+ }
+ }
+
@Override
public int getOutputColumn() {
return colNum;
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java
index b2798d2..bbf8862 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java
@@ -453,8 +453,7 @@ public final class VectorExpressionWriterFactory {
case UNION:
case MAP:
case LIST:
- throw new IllegalArgumentException("Unsupported complex type: " +
- fieldObjInspector.getCategory());
+ return genVectorExpressionWritableEmpty();
default:
throw new IllegalArgumentException("Unknown type " +
fieldObjInspector.getCategory());
@@ -1113,6 +1112,32 @@ public final class VectorExpressionWriterFactory {
}.init(fieldObjInspector);
}
+ // For complex types like STRUCT, MAP, etc we do not support, we need a writer that
+ // does nothing. We assume the Vectorizer class has not validated the query to actually
+ // try and use the complex types. They do show up in inputObjInspector[0] and need to be
+ // ignored.
+ private static VectorExpressionWriter genVectorExpressionWritableEmpty() {
+ return new VectorExpressionWriterBase() {
+
+ @Override
+ public Object writeValue(ColumnVector column, int row)
+ throws HiveException {
+ return null;
+ }
+
+ @Override
+ public Object setValue(Object row, ColumnVector column, int columnRow)
+ throws HiveException {
+ return null;
+ }
+
+ @Override
+ public Object initValue(Object ost) throws HiveException {
+ return null;
+ }
+ };
+ }
+
/**
* Helper function to create an array of writers from a list of expression descriptors.
*/
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java
new file mode 100644
index 0000000..f272b6d
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java
@@ -0,0 +1,769 @@
+/**
+ * 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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.HashTableLoaderFactory;
+import org.apache.hadoop.hive.ql.exec.HashTableLoader;
+import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
+import org.apache.hadoop.hive.ql.exec.MapredContext;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecMapperContext;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainer;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainerSerDe;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainer.ReusableGetAdaptor;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorColumnMapping;
+import org.apache.hadoop.hive.ql.exec.vector.VectorColumnOutputMapping;
+import org.apache.hadoop.hive.ql.exec.vector.VectorColumnSourceMapping;
+import org.apache.hadoop.hive.ql.exec.vector.VectorCopyRow;
+import org.apache.hadoop.hive.ql.exec.vector.VectorDeserializeRow;
+import org.apache.hadoop.hive.ql.exec.vector.VectorExpressionDescriptor;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizationContext;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizationContextRegion;
+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.VectorizedRowBatchCtx;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.IdentityExpression;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpression;
+import org.apache.hadoop.hive.ql.exec.vector.mapjoin.optimized.VectorMapJoinOptimizedCreateHashTable;
+import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashTable;
+import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinTableContainer;
+import org.apache.hadoop.hive.ql.exec.vector.mapjoin.fast.VectorMapJoinFastHashTableLoader;
+import org.apache.hadoop.hive.ql.log.PerfLogger;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+import org.apache.hadoop.hive.ql.plan.MapJoinDesc;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+import org.apache.hadoop.hive.ql.plan.VectorMapJoinDesc;
+import org.apache.hadoop.hive.ql.plan.VectorMapJoinDesc.HashTableImplementationType;
+import org.apache.hadoop.hive.ql.plan.api.OperatorType;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.objectinspector.StructField;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+
+/**
+ * This class is common operator class for native vectorized map join.
+ *
+ * It contain common initialization logic.
+ *
+ * It is used by both inner and outer joins.
+ */
+public abstract class VectorMapJoinCommonOperator extends MapJoinOperator implements VectorizationContextRegion {
+ private static final long serialVersionUID = 1L;
+ private static final Log LOG = LogFactory.getLog(VectorMapJoinCommonOperator.class.getName());
+
+ // Whether this operator is an outer join.
+ protected boolean isOuterJoin;
+
+ // Position of the *single* native vector map join small table.
+ protected byte posSingleVectorMapJoinSmallTable;
+
+ // The incoming vectorization context. It describes the input big table vectorized row batch.
+ protected VectorizationContext vContext;
+
+ // This is the vectorized row batch description of the output of the native vectorized map join
+ // operator. It is based on the incoming vectorization context. Its projection may include
+ // a mixture of input big table columns and new scratch columns.
+ protected VectorizationContext vOutContext;
+
+ // The output column projection of the vectorized row batch. And, the type names of the output
+ // columns.
+ protected int[] outputProjection;
+ protected String[] outputTypeNames;
+
+ // These are the vectorized batch expressions for filtering, key expressions, and value
+ // expressions.
+ protected VectorExpression[] bigTableFilterExpressions;
+ protected VectorExpression[] bigTableKeyExpressions;
+ protected VectorExpression[] bigTableValueExpressions;
+
+ // This is map of which vectorized row batch columns are the big table key columns. Since
+ // we may have key expressions that produce new scratch columns, we need a mapping.
+ // And, we have their type names.
+ protected int[] bigTableKeyColumnMap;
+ protected ArrayList<String> bigTableKeyTypeNames;
+
+ // Similarly, this is map of which vectorized row batch columns are the big table value columns.
+ // Since we may have value expressions that produce new scratch columns, we need a mapping.
+ // And, we have their type names.
+ protected int[] bigTableValueColumnMap;
+ protected ArrayList<String> bigTableValueTypeNames;
+
+ // This is a mapping of which big table columns (input and key/value expressions) will be
+ // part of the big table portion of the join output result.
+ protected VectorColumnOutputMapping bigTableRetainedMapping;
+
+ // This is a mapping of which keys will be copied from the big table (input and key expressions)
+ // to the small table result portion of the output for outer join.
+ protected VectorColumnOutputMapping bigTableOuterKeyMapping;
+
+ // This is a mapping of the values in the small table hash table that will be copied to the
+ // small table result portion of the output. That is, a mapping of the LazyBinary field order
+ // to output batch scratch columns for the small table portion.
+ protected VectorColumnSourceMapping smallTableMapping;
+
+ // These are the columns in the big and small table that are ByteColumnVector columns.
+ // We create data buffers for these columns so we can copy strings into those columns by value.
+ protected int[] bigTableByteColumnVectorColumns;
+ protected int[] smallTableByteColumnVectorColumns;
+
+ // The above members are initialized by the constructor and must not be
+ // transient.
+ //---------------------------------------------------------------------------
+
+ // For debug tracing: the name of the map or reduce task.
+ protected transient String taskName;
+
+ // The threshold where we should use a repeating vectorized row batch optimization for
+ // generating join output results.
+ protected transient boolean useOverflowRepeatedThreshold;
+ protected transient int overflowRepeatedThreshold;
+
+ // A helper object that efficiently copies the big table columns that are for the big table
+ // portion of the join output.
+ protected transient VectorCopyRow bigTableRetainedVectorCopy;
+
+ // A helper object that efficiently copies the big table key columns (input or key expressions)
+ // that appear in the small table portion of the join output for outer joins.
+ protected transient VectorCopyRow bigTableVectorCopyOuterKeys;
+
+ // This helper object deserializes LazyBinary format small table values into columns of a row
+ // in a vectorized row batch.
+ protected transient VectorDeserializeRow smallTableVectorDeserializeRow;
+
+ // This a 2nd batch with the same "column schema" as the big table batch that can be used to
+ // build join output results in. If we can create some join output results in the big table
+ // batch, we will for better efficiency (i.e. avoiding copying). Otherwise, we will use the
+ // overflow batch.
+ protected transient VectorizedRowBatch overflowBatch;
+
+ // A scratch batch that will be used to play back big table rows that were spilled
+ // to disk for the Hybrid Grace hash partitioning.
+ protected transient VectorizedRowBatch spillReplayBatch;
+
+ // Whether the native vectorized map join operator has performed its common setup.
+ protected transient boolean needCommonSetup;
+
+ // Whether the native vectorized map join operator has performed its
+ // native vector map join hash table setup.
+ protected transient boolean needHashTableSetup;
+
+ // The small table hash table for the native vectorized map join operator.
+ protected transient VectorMapJoinHashTable vectorMapJoinHashTable;
+
+ public VectorMapJoinCommonOperator() {
+ super();
+ }
+
+ public VectorMapJoinCommonOperator(VectorizationContext vContext, OperatorDesc conf)
+ throws HiveException {
+ super();
+
+ MapJoinDesc desc = (MapJoinDesc) conf;
+ this.conf = desc;
+
+ this.vContext = vContext;
+
+ /*
+ * Create a new vectorization context to create a new projection, but keep
+ * same output column manager must be inherited to track the scratch the columns.
+ */
+ vOutContext = new VectorizationContext(getName(), this.vContext);
+
+ order = desc.getTagOrder();
+ posBigTable = (byte) desc.getPosBigTable();
+ posSingleVectorMapJoinSmallTable = (order[0] == posBigTable ? order[1] : order[0]);
+ isOuterJoin = !desc.getNoOuterJoin();
+
+ Map<Byte, List<ExprNodeDesc>> filterExpressions = desc.getFilters();
+ bigTableFilterExpressions = vContext.getVectorExpressions(filterExpressions.get(posBigTable),
+ VectorExpressionDescriptor.Mode.FILTER);
+
+ List<ExprNodeDesc> keyDesc = desc.getKeys().get(posBigTable);
+ bigTableKeyExpressions = vContext.getVectorExpressions(keyDesc);
+
+ // Since a key expression can be a calculation and the key will go into a scratch column,
+ // we need the mapping and type information.
+ bigTableKeyColumnMap = new int[bigTableKeyExpressions.length];
+ bigTableKeyTypeNames = new ArrayList<String>();
+ boolean onlyColumns = true;
+ for (int i = 0; i < bigTableKeyColumnMap.length; i++) {
+ VectorExpression ve = bigTableKeyExpressions[i];
+ if (!IdentityExpression.isColumnOnly(ve)) {
+ onlyColumns = false;
+ }
+ bigTableKeyTypeNames.add(keyDesc.get(i).getTypeString());
+ bigTableKeyColumnMap[i] = ve.getOutputColumn();
+ }
+ if (onlyColumns) {
+ bigTableKeyExpressions = null;
+ }
+
+ List<ExprNodeDesc> bigTableExprs = desc.getExprs().get(posBigTable);
+ bigTableValueExpressions = vContext.getVectorExpressions(bigTableExprs);
+
+ /*
+ * Similarly, we need a mapping since a value expression can be a calculation and the value
+ * will go into a scratch column.
+ */
+ bigTableValueColumnMap = new int[bigTableValueExpressions.length];
+ bigTableValueTypeNames = new ArrayList<String>();
+ onlyColumns = true;
+ for (int i = 0; i < bigTableValueColumnMap.length; i++) {
+ VectorExpression ve = bigTableValueExpressions[i];
+ if (!IdentityExpression.isColumnOnly(ve)) {
+ onlyColumns = false;
+ }
+ bigTableValueTypeNames.add(bigTableExprs.get(i).getTypeString());
+ bigTableValueColumnMap[i] = ve.getOutputColumn();
+ }
+ if (onlyColumns) {
+ bigTableValueExpressions = null;
+ }
+
+ determineCommonInfo(isOuterJoin);
+ }
+
+ protected void determineCommonInfo(boolean isOuter) {
+
+ bigTableRetainedMapping = new VectorColumnOutputMapping("Big Table Retained Mapping");
+
+ bigTableOuterKeyMapping = new VectorColumnOutputMapping("Big Table Outer Key Mapping");
+
+ // The order of the fields in the LazyBinary small table value must be used, so
+ // we use the source ordering flavor for the mapping.
+ smallTableMapping = new VectorColumnSourceMapping("Small Table Mapping");
+
+ // We use a mapping object here so we can build the projection in any order and
+ // get the ordered by 0 to n-1 output columns at the end.
+ //
+ // Also, to avoid copying a big table key into the small table result area for inner joins,
+ // we reference it with the projection so there can be duplicate output columns
+ // in the projection.
+ VectorColumnSourceMapping projectionMapping = new VectorColumnSourceMapping("Projection Mapping");
+
+ /*
+ * Gather up big and small table output result information from the MapJoinDesc.
+ */
+ List<Integer> bigTableRetainList = conf.getRetainList().get(posBigTable);
+ int bigTableRetainSize = bigTableRetainList.size();
+
+ int[] smallTableIndices;
+ int smallTableIndicesSize;
+ List<ExprNodeDesc> smallTableExprs = conf.getExprs().get(posSingleVectorMapJoinSmallTable);
+ if (conf.getValueIndices() != null && conf.getValueIndices().get(posSingleVectorMapJoinSmallTable) != null) {
+ smallTableIndices = conf.getValueIndices().get(posSingleVectorMapJoinSmallTable);
+ smallTableIndicesSize = smallTableIndices.length;
+ } else {
+ smallTableIndices = null;
+ smallTableIndicesSize = 0;
+ }
+
+ List<Integer> smallTableRetainList = conf.getRetainList().get(posSingleVectorMapJoinSmallTable);
+ int smallTableRetainSize = smallTableRetainList.size();
+
+ int smallTableResultSize = 0;
+ if (smallTableIndicesSize > 0) {
+ smallTableResultSize = smallTableIndicesSize;
+ } else if (smallTableRetainSize > 0) {
+ smallTableResultSize = smallTableRetainSize;
+ }
+
+ /*
+ * Determine the big table retained mapping first so we can optimize out (with
+ * projection) copying inner join big table keys in the subsequent small table results section.
+ */
+ int nextOutputColumn = (order[0] == posBigTable ? 0 : smallTableResultSize);
+ for (int i = 0; i < bigTableRetainSize; i++) {
+
+ // Since bigTableValueExpressions may do a calculation and produce a scratch column, we
+ // need to map to the right batch column.
+
+ int retainColumn = bigTableRetainList.get(i);
+ int batchColumnIndex = bigTableValueColumnMap[retainColumn];
+ String typeName = bigTableValueTypeNames.get(i);
+
+ // With this map we project the big table batch to make it look like an output batch.
+ projectionMapping.add(nextOutputColumn, batchColumnIndex, typeName);
+
+ // Collect columns we copy from the big table batch to the overflow batch.
+ bigTableRetainedMapping.add(batchColumnIndex, batchColumnIndex, typeName);
+
+ nextOutputColumn++;
+ }
+
+ /*
+ * Now determine the small table results.
+ */
+ int firstSmallTableOutputColumn;
+ firstSmallTableOutputColumn = (order[0] == posBigTable ? bigTableRetainSize : 0);
+ int smallTableOutputCount = 0;
+ nextOutputColumn = firstSmallTableOutputColumn;
+
+ // Small table indices has more information (i.e. keys) than retain, so use it if it exists...
+ if (smallTableIndicesSize > 0) {
+ smallTableOutputCount = smallTableIndicesSize;
+
+ for (int i = 0; i < smallTableIndicesSize; i++) {
+ if (smallTableIndices[i] >= 0) {
+
+ // Zero and above numbers indicate a big table key is needed for
+ // small table result "area".
+
+ int keyIndex = smallTableIndices[i];
+
+ // Since bigTableKeyExpressions may do a calculation and produce a scratch column, we
+ // need to map the right column.
+ int batchKeyColumn = bigTableKeyColumnMap[keyIndex];
+ String typeName = bigTableKeyTypeNames.get(keyIndex);
+
+ if (!isOuter) {
+
+ // Optimize inner join keys of small table results.
+
+ // Project the big table key into the small table result "area".
+ projectionMapping.add(nextOutputColumn, batchKeyColumn, typeName);
+
+ if (!bigTableRetainedMapping.containsOutputColumn(batchKeyColumn)) {
+ // If necessary, copy the big table key into the overflow batch's small table
+ // result "area".
+ bigTableRetainedMapping.add(batchKeyColumn, batchKeyColumn, typeName);
+ }
+ } else {
+
+ // For outer joins, since the small table key can be null when there is no match,
+ // we must have a physical (scratch) column for those keys. We cannot use the
+ // projection optimization used by inner joins above.
+
+ int scratchColumn = vOutContext.allocateScratchColumn(typeName);
+ projectionMapping.add(nextOutputColumn, scratchColumn, typeName);
+
+ bigTableRetainedMapping.add(batchKeyColumn, scratchColumn, typeName);
+
+ bigTableOuterKeyMapping.add(batchKeyColumn, scratchColumn, typeName);
+ }
+ } else {
+
+ // Negative numbers indicate a column to be (deserialize) read from the small table's
+ // LazyBinary value row.
+ int smallTableValueIndex = -smallTableIndices[i] - 1;
+
+ String typeName = smallTableExprs.get(i).getTypeString();
+
+ // Make a new big table scratch column for the small table value.
+ int scratchColumn = vOutContext.allocateScratchColumn(typeName);
+ projectionMapping.add(nextOutputColumn, scratchColumn, typeName);
+
+ smallTableMapping.add(smallTableValueIndex, scratchColumn, typeName);
+ }
+ nextOutputColumn++;
+ }
+ } else if (smallTableRetainSize > 0) {
+ smallTableOutputCount = smallTableRetainSize;
+
+ // Only small table values appear in join output result.
+
+ for (int i = 0; i < smallTableRetainSize; i++) {
+ int smallTableValueIndex = smallTableRetainList.get(i);
+
+ // Make a new big table scratch column for the small table value.
+ String typeName = smallTableExprs.get(i).getTypeString();
+ int scratchColumn = vOutContext.allocateScratchColumn(typeName);
+
+ projectionMapping.add(nextOutputColumn, scratchColumn, typeName);
+
+ smallTableMapping.add(smallTableValueIndex, scratchColumn, typeName);
+ nextOutputColumn++;
+ }
+ }
+
+ // Convert dynamic arrays and maps to simple arrays.
+
+ bigTableRetainedMapping.finalize();
+
+ bigTableOuterKeyMapping.finalize();
+
+ smallTableMapping.finalize();
+
+ // Which big table and small table columns are ByteColumnVector and need have their data buffer
+ // to be manually reset for some join result processing?
+
+ bigTableByteColumnVectorColumns = getByteColumnVectorColumns(bigTableOuterKeyMapping);
+
+ smallTableByteColumnVectorColumns = getByteColumnVectorColumns(smallTableMapping);
+
+ projectionMapping.finalize();
+
+ // Verify we added an entry for each output.
+ assert projectionMapping.isSourceSequenceGood();
+
+ outputProjection = projectionMapping.getOutputColumns();
+ outputTypeNames = projectionMapping.getTypeNames();
+
+ if (LOG.isDebugEnabled()) {
+ int[] orderDisplayable = new int[order.length];
+ for (int i = 0; i < order.length; i++) {
+ orderDisplayable[i] = (int) order[i];
+ }
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor order " + Arrays.toString(orderDisplayable));
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor posBigTable " + (int) posBigTable);
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor posSingleVectorMapJoinSmallTable " + (int) posSingleVectorMapJoinSmallTable);
+
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor bigTableKeyColumnMap " + Arrays.toString(bigTableKeyColumnMap));
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor bigTableKeyTypeNames " + bigTableKeyTypeNames);
+
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor bigTableValueColumnMap " + Arrays.toString(bigTableValueColumnMap));
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor bigTableValueTypeNames " + bigTableValueTypeNames);
+
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor smallTableIndices " + Arrays.toString(smallTableIndices));
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor smallTableRetainList " + smallTableRetainList);
+
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor firstSmallTableOutputColumn " + firstSmallTableOutputColumn);
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor smallTableOutputCount " + smallTableOutputCount);
+
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor bigTableRetainedMapping " + bigTableRetainedMapping.toString());
+
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor bigTableOuterKeyMapping " + bigTableOuterKeyMapping.toString());
+
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor smallTableMapping " + smallTableMapping.toString());
+
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor bigTableByteColumnVectorColumns " + Arrays.toString(bigTableByteColumnVectorColumns));
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor smallTableByteColumnVectorColumns " + Arrays.toString(smallTableByteColumnVectorColumns));
+
+ LOG.info(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor outputProjection " + Arrays.toString(outputProjection));
+ LOG.info(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor outputTypeNames " + Arrays.toString(outputTypeNames));
+ }
+
+ setupVOutContext(conf.getOutputColumnNames());
+ }
+
+ /**
+ * Determine from a mapping which columns are BytesColumnVector columns.
+ */
+ private int[] getByteColumnVectorColumns(VectorColumnMapping mapping) {
+ // Search mapping for any strings and return their output columns.
+ ArrayList<Integer> list = new ArrayList<Integer>();
+ int count = mapping.getCount();
+ int[] outputColumns = mapping.getOutputColumns();
+ String[] typeNames = mapping.getTypeNames();
+ for (int i = 0; i < count; i++) {
+ int outputColumn = outputColumns[i];
+ String typeName = typeNames[i];
+ if (VectorizationContext.isStringFamily(typeName)) {
+ list.add(outputColumn);
+ }
+ }
+ return ArrayUtils.toPrimitive(list.toArray(new Integer[0]));
+ }
+
+ /**
+ * Setup the vectorized row batch description of the output of the native vectorized map join
+ * operator. Use the output projection we previously built from a mixture of input big table
+ * columns and new scratch columns.
+ */
+ protected void setupVOutContext(List<String> outputColumnNames) {
+ if (LOG.isDebugEnabled()) {
+ LOG.info(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor outputColumnNames " + outputColumnNames);
+ }
+ if (outputColumnNames.size() != outputProjection.length) {
+ throw new RuntimeException("Output column names " + outputColumnNames + " length and output projection " + Arrays.toString(outputProjection) + " / " + Arrays.toString(outputTypeNames) + " length mismatch");
+ }
+ vOutContext.resetProjectionColumns();
+ for (int i = 0; i < outputColumnNames.size(); ++i) {
+ String columnName = outputColumnNames.get(i);
+ int outputColumn = outputProjection[i];
+ vOutContext.addProjectionColumn(columnName, outputColumn);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator constructor addProjectionColumn " + i + " columnName " + columnName + " outputColumn " + outputColumn);
+ }
+ }
+ }
+
+ /**
+ * This override lets us substitute our own fast vectorized hash table loader.
+ */
+ @Override
+ protected HashTableLoader getHashTableLoader(Configuration hconf) {
+
+ VectorMapJoinDesc vectorDesc = conf.getVectorDesc();
+ HashTableImplementationType hashTableImplementationType = vectorDesc.hashTableImplementationType();
+ HashTableLoader hashTableLoader;
+ switch (vectorDesc.hashTableImplementationType()) {
+ case OPTIMIZED:
+ // Use the Tez hash table loader.
+ hashTableLoader = HashTableLoaderFactory.getLoader(hconf);
+ break;
+ case FAST:
+ // Use our specialized hash table loader.
+ hashTableLoader = new VectorMapJoinFastHashTableLoader();
+ break;
+ default:
+ throw new RuntimeException("Unknown vector map join hash table implementation type " + hashTableImplementationType.name());
+ }
+ return hashTableLoader;
+ }
+
+ @Override
+ protected Collection<Future<?>> initializeOp(Configuration hconf) throws HiveException {
+ Collection<Future<?>> result = super.initializeOp(hconf);
+
+ if (LOG.isDebugEnabled()) {
+ // Determine the name of our map or reduce task for debug tracing.
+ BaseWork work = Utilities.getMapWork(hconf);
+ if (work == null) {
+ work = Utilities.getReduceWork(hconf);
+ }
+ taskName = work.getName();
+ }
+
+ /*
+ * Get configuration parameters.
+ */
+ overflowRepeatedThreshold = HiveConf.getIntVar(hconf,
+ HiveConf.ConfVars.HIVE_VECTORIZATION_MAPJOIN_NATIVE_OVERFLOW_REPEATED_THRESHOLD);
+ useOverflowRepeatedThreshold = (overflowRepeatedThreshold >= 0);
+
+
+ /*
+ * Create our vectorized copy row and deserialize row helper objects.
+ */
+ if (smallTableMapping.getCount() > 0) {
+ smallTableVectorDeserializeRow = new VectorDeserializeRow(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.primitiveTypeInfosFromTypeNames(
+ smallTableMapping.getTypeNames())));
+ smallTableVectorDeserializeRow.init(smallTableMapping.getOutputColumns());
+ }
+
+ if (bigTableRetainedMapping.getCount() > 0) {
+ bigTableRetainedVectorCopy = new VectorCopyRow();
+ bigTableRetainedVectorCopy.init(bigTableRetainedMapping);
+ }
+
+ if (bigTableOuterKeyMapping.getCount() > 0) {
+ bigTableVectorCopyOuterKeys = new VectorCopyRow();
+ bigTableVectorCopyOuterKeys.init(bigTableOuterKeyMapping);
+ }
+
+ /*
+ * Setup the overflow batch.
+ */
+ overflowBatch = setupOverflowBatch();
+
+ needCommonSetup = true;
+ needHashTableSetup = true;
+
+ if (LOG.isDebugEnabled()) {
+ int[] currentScratchColumns = vOutContext.currentScratchColumns();
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator initializeOp currentScratchColumns " + Arrays.toString(currentScratchColumns));
+
+ StructObjectInspector structOutputObjectInspector = (StructObjectInspector) outputObjInspector;
+ List<? extends StructField> fields = structOutputObjectInspector.getAllStructFieldRefs();
+ int i = 0;
+ for (StructField field : fields) {
+ LOG.debug("VectorMapJoinInnerBigOnlyCommonOperator initializeOp " + i + " field " + field.getFieldName() + " type " + field.getFieldObjectInspector().getTypeName());
+ i++;
+ }
+ }
+
+ return result;
+ }
+
+ @Override
+ protected Pair<MapJoinTableContainer[], MapJoinTableContainerSerDe[]> loadHashTable(
+ ExecMapperContext mapContext, MapredContext mrContext) throws HiveException {
+
+ Pair<MapJoinTableContainer[], MapJoinTableContainerSerDe[]> pair;
+
+ VectorMapJoinDesc vectorDesc = conf.getVectorDesc();
+ HashTableImplementationType hashTableImplementationType = vectorDesc.hashTableImplementationType();
+ HashTableLoader hashTableLoader;
+ switch (vectorDesc.hashTableImplementationType()) {
+ case OPTIMIZED:
+ {
+ // Using Tez's HashTableLoader, create either a MapJoinBytesTableContainer or
+ // HybridHashTableContainer.
+ pair = super.loadHashTable(mapContext, mrContext);
+
+ // Create our vector map join optimized hash table variation *above* the
+ // map join table container.
+ MapJoinTableContainer[] mapJoinTableContainers = pair.getLeft();
+ vectorMapJoinHashTable = VectorMapJoinOptimizedCreateHashTable.createHashTable(conf,
+ mapJoinTableContainers[posSingleVectorMapJoinSmallTable]);
+ }
+ break;
+
+ case FAST:
+ {
+ // Use our VectorMapJoinFastHashTableLoader to create a VectorMapJoinTableContainer.
+ pair = super.loadHashTable(mapContext, mrContext);
+
+ // Get our vector map join fast hash table variation from the
+ // vector map join table container.
+ MapJoinTableContainer[] mapJoinTableContainers = pair.getLeft();
+ VectorMapJoinTableContainer vectorMapJoinTableContainer =
+ (VectorMapJoinTableContainer) mapJoinTableContainers[posSingleVectorMapJoinSmallTable];
+ vectorMapJoinHashTable = vectorMapJoinTableContainer.vectorMapJoinHashTable();
+ }
+ break;
+ default:
+ throw new RuntimeException("Unknown vector map join hash table implementation type " + hashTableImplementationType.name());
+ }
+
+ return pair;
+ }
+
+ /*
+ * Setup our 2nd batch with the same "column schema" as the big table batch that can be used to
+ * build join output results in.
+ */
+ protected VectorizedRowBatch setupOverflowBatch() {
+ VectorizedRowBatch overflowBatch;
+
+ Map<Integer, String> scratchColumnTypeMap = vOutContext.getScratchColumnTypeMap();
+ int maxColumn = 0;
+ for (int i = 0; i < outputProjection.length; i++) {
+ int outputColumn = outputProjection[i];
+ if (maxColumn < outputColumn) {
+ maxColumn = outputColumn;
+ }
+ }
+ for (int outputColumn : scratchColumnTypeMap.keySet()) {
+ if (maxColumn < outputColumn) {
+ maxColumn = outputColumn;
+ }
+ }
+ overflowBatch = new VectorizedRowBatch(maxColumn + 1);
+
+ // First, just allocate just the projection columns we will be using.
+ for (int i = 0; i < outputProjection.length; i++) {
+ int outputColumn = outputProjection[i];
+ String typeName = outputTypeNames[i];
+ allocateOverflowBatchColumnVector(overflowBatch, outputColumn, typeName);
+ }
+
+ // Now, add any scratch columns needed for children operators.
+ for (int outputColumn : scratchColumnTypeMap.keySet()) {
+ String typeName = scratchColumnTypeMap.get(outputColumn);
+ allocateOverflowBatchColumnVector(overflowBatch, outputColumn, typeName);
+ }
+
+ overflowBatch.projectedColumns = outputProjection;
+ overflowBatch.projectionSize = outputProjection.length;
+
+ overflowBatch.reset();
+
+ return overflowBatch;
+ }
+
+ /*
+ * Allocate overflow batch columns by hand.
+ */
+ private void allocateOverflowBatchColumnVector(VectorizedRowBatch overflowBatch, int outputColumn,
+ String typeName) {
+
+ if (overflowBatch.cols[outputColumn] == null) {
+ String vectorTypeName;
+ if (VectorizationContext.isIntFamily(typeName) ||
+ VectorizationContext.isDatetimeFamily(typeName)) {
+ vectorTypeName = "long";
+ } else if (VectorizationContext.isFloatFamily(typeName)) {
+ vectorTypeName = "double";
+ } else if (VectorizationContext.isStringFamily(typeName)) {
+ vectorTypeName = "string";
+ } else if (VectorizationContext.decimalTypePattern.matcher(typeName).matches()){
+ vectorTypeName = typeName; // Keep precision and scale.
+ } else {
+ throw new RuntimeException("Cannot determine vector type for " + typeName);
+ }
+ overflowBatch.cols[outputColumn] = VectorizedRowBatchCtx.allocateColumnVector(vectorTypeName, VectorizedRowBatch.DEFAULT_SIZE);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(taskName + ", " + getOperatorId() + " VectorMapJoinCommonOperator initializeOp overflowBatch outputColumn " + outputColumn + " class " + overflowBatch.cols[outputColumn].getClass().getSimpleName());
+ }
+ }
+ }
+
+ /*
+ * Common one time setup by native vectorized map join operator's processOp.
+ */
+ protected void commonSetup(VectorizedRowBatch batch) throws HiveException {
+ LOG.info("VectorMapJoinInnerCommonOperator commonSetup begin...");
+
+ if (LOG.isDebugEnabled()) {
+ displayBatchColumns(batch, "batch");
+ displayBatchColumns(overflowBatch, "overflowBatch");
+ }
+
+ // Make sure big table BytesColumnVectors have room for string values in the overflow batch...
+ for (int column: bigTableByteColumnVectorColumns) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) overflowBatch.cols[column];
+ bytesColumnVector.initBuffer();
+ }
+
+ // Make sure small table BytesColumnVectors have room for string values in the big table and
+ // overflow batchs...
+ for (int column: smallTableByteColumnVectorColumns) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) batch.cols[column];
+ bytesColumnVector.initBuffer();
+ bytesColumnVector = (BytesColumnVector) overflowBatch.cols[column];
+ bytesColumnVector.initBuffer();
+ }
+
+ // Setup a scratch batch that will be used to play back big table rows that were spilled
+ // to disk for the Hybrid Grace hash partitioning.
+ spillReplayBatch = VectorizedBatchUtil.makeLike(batch);
+
+ // TEMPORARY -- Set this up for Hybrid Grace logic in MapJoinOperator.closeOp
+ hashMapRowGetters = new ReusableGetAdaptor[mapJoinTables.length];
+ smallTable = posSingleVectorMapJoinSmallTable;
+ }
+
+ protected void displayBatchColumns(VectorizedRowBatch batch, String batchName) {
+ LOG.debug("commonSetup " + batchName + " column count " + batch.numCols);
+ for (int column = 0; column < batch.numCols; column++) {
+ LOG.debug("commonSetup " + batchName + " column " + column + " type " + (batch.cols[column] == null ? "NULL" : batch.cols[column].getClass().getSimpleName()));
+ }
+ }
+
+ @Override
+ public OperatorType getType() {
+ return OperatorType.MAPJOIN;
+ }
+
+ @Override
+ public VectorizationContext getOuputVectorizationContext() {
+ return vOutContext;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/d47995d9/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
new file mode 100644
index 0000000..743a975
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinGenerateResultOperator.java
@@ -0,0 +1,819 @@
+/**
+ * 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 java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.ql.exec.persistence.HybridHashTableContainer;
+import org.apache.hadoop.hive.ql.exec.persistence.HybridHashTableContainer.HashPartition;
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+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.VectorizationContext;
+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.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.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.WriteBuffers.ByteSegmentRef;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.ByteStream.Output;
+
+/**
+ * This class has methods for generating vectorized join results and forwarding batchs.
+ *
+ * In some cases when can forward the big table batch by setting scratch columns
+ * with small table results and then making use of our output projection to pick out all the
+ * output result columns. This can improve performance by avoiding copying big table values.
+ * So, we will use the big table batch's selected in use to represent those rows.
+ *
+ * At the same time, some output results need to be formed in the overflow batch.
+ * For example, to form N x M cross product output results. In this case, we will copy big
+ * table values into the overflow batch and set scratch columns in it for small table results.
+ * The "schema" of the overflow batch is the same as the big table batch so child operators
+ * only need one definition of their input batch. The overflow batch will be typically be
+ * forwarded when it gets full, which might not be during a process call.
+ *
+ * NOTE: Child operators should not remember a received batch.
+ */
+
+public abstract class VectorMapJoinGenerateResultOperator extends VectorMapJoinCommonOperator {
+
+ private static final long serialVersionUID = 1L;
+ private static final Log LOG = LogFactory.getLog(VectorMapJoinGenerateResultOperator.class.getName());
+ private static final String CLASS_NAME = VectorMapJoinGenerateResultOperator.class.getName();
+
+ private transient PrimitiveTypeInfo[] bigTablePrimitiveTypeInfos;
+
+ private transient VectorSerializeRow bigTableVectorSerializeRow;
+
+ private transient VectorDeserializeRow bigTableVectorDeserializeRow;
+
+ // Debug display.
+ protected transient long batchCounter;
+
+ public VectorMapJoinGenerateResultOperator() {
+ super();
+ }
+
+ public VectorMapJoinGenerateResultOperator(VectorizationContext vContext, OperatorDesc conf)
+ throws HiveException {
+ super(vContext, conf);
+ }
+
+ protected void commonSetup(VectorizedRowBatch batch) throws HiveException {
+ super.commonSetup(batch);
+
+ batchCounter = 0;
+
+ }
+
+ //------------------------------------------------------------------------------------------------
+
+ protected void performValueExpressions(VectorizedRowBatch batch,
+ int[] allMatchs, int allMatchCount) {
+ /*
+ * For the moment, pretend all matched are selected so we can evaluate the value
+ * expressions.
+ *
+ * Since we may use the overflow batch when generating results, we will assign the
+ * selected and real batch size later...
+ */
+ int[] saveSelected = batch.selected;
+ batch.selected = allMatchs;
+ boolean saveSelectedInUse = batch.selectedInUse;
+ batch.selectedInUse = true;
+ batch.size = allMatchCount;
+
+ // Run our value expressions over whole batch.
+ for(VectorExpression ve: bigTableValueExpressions) {
+ ve.evaluate(batch);
+ }
+
+ batch.selected = saveSelected;
+ batch.selectedInUse = saveSelectedInUse;
+ }
+
+ //------------------------------------------------------------------------------------------------
+
+ /*
+ * Common generate join results from hash maps used by Inner and Outer joins.
+ */
+
+ /**
+ * Generate join results for a single small table value match.
+ *
+ * @param batch
+ * The big table batch.
+ * @param hashMapResult
+ * The hash map results for the matching key.
+ * @param allMatchs
+ * The selection array for all matches key.
+ * @param allMatchesIndex
+ * Index into allMatches of the matching key we are generating for.
+ * @param duplicateCount
+ * Number of equal key rows.
+ * @param numSel
+ * Current number of rows that are remaining in the big table for forwarding.
+ * @return
+ * The new count of selected rows.
+ */
+ protected int generateHashMapResultSingleValue(VectorizedRowBatch batch,
+ VectorMapJoinHashMapResult hashMapResult, int[] allMatchs, int allMatchesIndex,
+ int duplicateCount, int numSel) throws HiveException, IOException {
+
+ // Read single value.
+
+ ByteSegmentRef byteSegmentRef = hashMapResult.first();
+
+ // Generate result within big table batch itself.
+
+ for (int i = 0; i < duplicateCount; i++) {
+
+ int batchIndex = allMatchs[allMatchesIndex + i];
+
+ if (bigTableVectorCopyOuterKeys != null) {
+ // Copy within row.
+ bigTableVectorCopyOuterKeys.copyByReference(batch, batchIndex, batch, batchIndex);
+ }
+
+ if (smallTableVectorDeserializeRow != null) {
+
+ byte[] bytes = byteSegmentRef.getBytes();
+ int offset = (int) byteSegmentRef.getOffset();
+ int length = byteSegmentRef.getLength();
+ smallTableVectorDeserializeRow.setBytes(bytes, offset, length);
+
+ smallTableVectorDeserializeRow.deserializeByValue(batch, batchIndex);
+ }
+
+ // VectorizedBatchUtil.debugDisplayOneRow(batch, batchIndex, "generateHashMapResultSingleValue big table");
+
+ // Use the big table row as output.
+ batch.selected[numSel++] = batchIndex;
+ }
+
+ return numSel;
+ }
+
+ /**
+ * Generate results for a N x M cross product.
+ *
+ * @param batch
+ * The big table batch.
+ * @param hashMapResult
+ * The hash map results for the matching key.
+ * @param allMatchs
+ * The all match selected array that contains (physical) batch indices.
+ * @param allMatchesIndex
+ * The index of the match key.
+ * @param duplicateCount
+ * Number of equal key rows.
+ */
+ protected void generateHashMapResultMultiValue(VectorizedRowBatch batch,
+ VectorMapJoinHashMapResult hashMapResult, int[] allMatchs, int allMatchesIndex,
+ int duplicateCount) throws HiveException, IOException {
+
+ if (useOverflowRepeatedThreshold &&
+ hashMapResult.isCappedCountAvailable() &&
+ hashMapResult.cappedCount() > overflowRepeatedThreshold) {
+
+ // Large cross product: generate the vector optimization using repeating vectorized
+ // row batch optimization in the overflow batch.
+
+ generateHashMapResultLargeMultiValue(
+ batch, hashMapResult, allMatchs, allMatchesIndex, duplicateCount);
+ return;
+ }
+
+ // We do the cross product of the N big table equal key row's values against the
+ // small table matching key which has M value rows into overflow batch.
+
+ for (int i = 0; i < duplicateCount; i++) {
+
+ int batchIndex = allMatchs[allMatchesIndex + i];
+
+ ByteSegmentRef byteSegmentRef = hashMapResult.first();
+ while (byteSegmentRef != null) {
+
+ // Copy the BigTable values into the overflow batch. Since the overflow batch may
+ // not get flushed here, we must copy by value.
+ if (bigTableRetainedVectorCopy != null) {
+ bigTableRetainedVectorCopy.copyByValue(batch, batchIndex,
+ overflowBatch, overflowBatch.size);
+ }
+
+ // Reference the keys we just copied above.
+ if (bigTableVectorCopyOuterKeys != null) {
+ bigTableVectorCopyOuterKeys.copyByReference(overflowBatch, overflowBatch.size,
+ overflowBatch, overflowBatch.size);
+ }
+
+ if (smallTableVectorDeserializeRow != null) {
+
+ byte[] bytes = byteSegmentRef.getBytes();
+ int offset = (int) byteSegmentRef.getOffset();
+ int length = byteSegmentRef.getLength();
+ smallTableVectorDeserializeRow.setBytes(bytes, offset, length);
+
+ smallTableVectorDeserializeRow.deserializeByValue(overflowBatch, overflowBatch.size);
+ }
+
+ // VectorizedBatchUtil.debugDisplayOneRow(overflowBatch, overflowBatch.size, "generateHashMapResultMultiValue overflow");
+
+ overflowBatch.size++;
+ if (overflowBatch.size == overflowBatch.DEFAULT_SIZE) {
+ forwardOverflow();
+ }
+ byteSegmentRef = hashMapResult.next();
+ }
+ }
+ }
+
+ /**
+ * Generate optimized results for a large N x M cross product using repeated vectorized row
+ * batch optimization.
+ *
+ * @param batch
+ * The big table batch.
+ * @param hashMapResult
+ * The hash map results for the matching key.
+ * @param allMatchs
+ * The all match selected array that contains (physical) batch indices.
+ * @param allMatchesIndex
+ * The index of the match key.
+ * @param duplicateCount
+ * Number of equal key rows.
+ */
+ private void generateHashMapResultLargeMultiValue(VectorizedRowBatch batch,
+ VectorMapJoinHashMapResult hashMapResult, int[] allMatchs, int allMatchesIndex,
+ int duplicateCount) throws HiveException, IOException {
+
+ // Kick out previous overflow batch results.
+ if (overflowBatch.size > 0) {
+ forwardOverflow();
+ }
+
+ ByteSegmentRef byteSegmentRef = hashMapResult.first();
+ while (true) {
+
+ // Fill up as much of the overflow batch as possible with small table values.
+ while (byteSegmentRef != null) {
+
+ if (smallTableVectorDeserializeRow != null) {
+
+ byte[] bytes = byteSegmentRef.getBytes();
+ int offset = (int) byteSegmentRef.getOffset();
+ int length = byteSegmentRef.getLength();
+ smallTableVectorDeserializeRow.setBytes(bytes, offset, length);
+
+ smallTableVectorDeserializeRow.deserializeByValue(overflowBatch, overflowBatch.DEFAULT_SIZE);
+ }
+
+ overflowBatch.size++;
+ if (overflowBatch.size == overflowBatch.DEFAULT_SIZE) {
+ break;
+ }
+ byteSegmentRef = hashMapResult.next();
+ }
+
+ // Forward the overflow batch over and over:
+ //
+ // Reference a new one big table row's values each time
+ // cross product
+ // Current "batch" of small table values.
+ //
+ // TODO: This could be further optimized to copy big table (equal) keys once
+ // and only copy big table values each time...
+ // And, not set repeating every time...
+ //
+
+ for (int i = 0; i < duplicateCount; i++) {
+
+ int batchIndex = allMatchs[allMatchesIndex + i];
+
+ if (bigTableRetainedVectorCopy != null) {
+ // The one big table row's values repeat.
+ bigTableRetainedVectorCopy.copyByReference(batch, batchIndex, overflowBatch, 0);
+ for (int column : bigTableRetainedMapping.getOutputColumns()) {
+ overflowBatch.cols[column].isRepeating = true;
+ }
+ }
+ if (bigTableVectorCopyOuterKeys != null) {
+ bigTableVectorCopyOuterKeys.copyByReference(batch, batchIndex, overflowBatch, 0);
+ for (int column : bigTableOuterKeyMapping.getOutputColumns()) {
+ overflowBatch.cols[column].isRepeating = true;
+ }
+ }
+
+ // Crucial here that we don't reset the overflow batch, or we will loose the small table
+ // values we put in above.
+ forwardOverflowNoReset();
+
+ // Hand reset the big table columns.
+ for (int column : bigTableRetainedMapping.getOutputColumns()) {
+ ColumnVector colVector = overflowBatch.cols[column];
+ colVector.reset();
+ }
+
+ if (bigTableVectorCopyOuterKeys != null) {
+ for (int column : bigTableOuterKeyMapping.getOutputColumns()) {
+ ColumnVector colVector = overflowBatch.cols[column];
+ colVector.reset();
+ }
+ }
+ }
+
+ if (hashMapResult.isEof()) {
+ break;
+ }
+ byteSegmentRef = hashMapResult.next();
+
+ // Get ready for a another round of small table values.
+ overflowBatch.reset();
+ }
+ // Clear away any residue from our optimizations.
+ overflowBatch.reset();
+ }
+
+ /**
+ * Generate optimized results when entire batch key is repeated and it matched the hash map.
+ *
+ * @param batch
+ * The big table batch.
+ * @param hashMapResult
+ * The hash map results for the repeated key.
+ * @return
+ * The new count of selected rows.
+ */
+ protected int generateHashMapResultRepeatedAll(VectorizedRowBatch batch,
+ VectorMapJoinHashMapResult hashMapResult) throws IOException, HiveException {
+
+ int[] selected = batch.selected;
+
+ if (batch.selectedInUse) {
+ // The selected array is already filled in as we want it.
+ } else {
+ for (int i = 0; i < batch.size; i++) {
+ selected[i] = i;
+ }
+ batch.selectedInUse = true;
+ }
+
+ int numSel = 0;
+ if (hashMapResult.isSingleRow()) {
+ numSel = generateHashMapResultSingleValue(batch, hashMapResult,
+ batch.selected, 0, batch.size, numSel);
+
+ } else {
+ generateHashMapResultMultiValue(batch, hashMapResult,
+ batch.selected, 0, batch.size);
+ }
+
+ return numSel;
+ }
+
+ //-----------------------------------------------------------------------------------------------
+
+ /*
+ * Spill.
+ */
+
+
+ private void setupSpillSerDe(VectorizedRowBatch batch) throws HiveException {
+
+ PrimitiveTypeInfo[] inputObjInspectorsTypeInfos =
+ VectorizedBatchUtil.primitiveTypeInfosFromStructObjectInspector(
+ (StructObjectInspector) inputObjInspectors[posBigTable]);
+
+ List<Integer> projectedColumns = vContext.getProjectedColumns();
+ int projectionSize = vContext.getProjectedColumns().size();
+
+ List<PrimitiveTypeInfo> typeInfoList = new ArrayList<PrimitiveTypeInfo>();
+ List<Integer> noNullsProjectionList = new ArrayList<Integer>();
+ for (int i = 0; i < projectionSize; i++) {
+ int projectedColumn = projectedColumns.get(i);
+ if (batch.cols[projectedColumn] != null) {
+ typeInfoList.add(inputObjInspectorsTypeInfos[i]);
+ noNullsProjectionList.add(projectedColumn);
+ }
+ }
+
+ int[] noNullsProjection = ArrayUtils.toPrimitive(noNullsProjectionList.toArray(new Integer[0]));
+ int noNullsProjectionSize = noNullsProjection.length;
+ bigTablePrimitiveTypeInfos = typeInfoList.toArray(new PrimitiveTypeInfo[0]);
+
+ bigTableVectorSerializeRow =
+ new VectorSerializeRow(new LazyBinarySerializeWrite(noNullsProjectionSize));
+
+ bigTableVectorSerializeRow.init(
+ bigTablePrimitiveTypeInfos,
+ noNullsProjectionList);
+
+ bigTableVectorDeserializeRow = new VectorDeserializeRow(
+ new LazyBinaryDeserializeRead(bigTablePrimitiveTypeInfos));
+
+ bigTableVectorDeserializeRow.init(noNullsProjection);
+ }
+
+ private void spillSerializeRow(VectorizedRowBatch batch, int batchIndex,
+ VectorMapJoinHashTableResult hashTableResult) throws IOException {
+
+ int partitionId = hashTableResult.spillPartitionId();
+
+ HybridHashTableContainer ht = (HybridHashTableContainer) mapJoinTables[smallTable];
+ HashPartition hp = ht.getHashPartitions()[partitionId];
+
+ VectorMapJoinRowBytesContainer rowBytesContainer = hp.getMatchfileRowBytesContainer();
+ Output output = rowBytesContainer.getOuputForRowBytes();
+// int offset = output.getLength();
+ bigTableVectorSerializeRow.setOutputAppend(output);
+ bigTableVectorSerializeRow.serializeWrite(batch, batchIndex);
+// int length = output.getLength() - offset;
+ rowBytesContainer.finishRow();
+
+// LOG.info("spillSerializeRow spilled batchIndex " + batchIndex + ", length " + length);
+ }
+
+ protected void spillHashMapBatch(VectorizedRowBatch batch,
+ VectorMapJoinHashTableResult[] hashTableResults,
+ int[] spills, int[] spillHashTableResultIndices, int spillCount)
+ throws HiveException, IOException {
+
+ if (bigTableVectorSerializeRow == null) {
+ setupSpillSerDe(batch);
+ }
+
+ for (int i = 0; i < spillCount; i++) {
+ int batchIndex = spills[i];
+
+ int hashTableResultIndex = spillHashTableResultIndices[i];
+ VectorMapJoinHashTableResult hashTableResult = hashTableResults[hashTableResultIndex];
+
+ spillSerializeRow(batch, batchIndex, hashTableResult);
+ }
+ }
+
+ protected void spillBatchRepeated(VectorizedRowBatch batch,
+ VectorMapJoinHashTableResult hashTableResult) throws HiveException, IOException {
+
+ if (bigTableVectorSerializeRow == null) {
+ setupSpillSerDe(batch);
+ }
+
+ int[] selected = batch.selected;
+ boolean selectedInUse = batch.selectedInUse;
+
+ for (int logical = 0; logical < batch.size; logical++) {
+ int batchIndex = (selectedInUse ? selected[logical] : logical);
+ spillSerializeRow(batch, batchIndex, hashTableResult);
+ }
+ }
+
+ @Override
+ protected void reloadHashTable(HashPartition partition,
+ HybridHashTableContainer hybridHtContainer)
+ throws IOException, ClassNotFoundException, HiveException, SerDeException {
+
+ // The super method will reload a hash table partition and
+ // put a single MapJoinBytesTableContainer into the currentSmallTable member.
+ super.reloadHashTable(partition, hybridHtContainer);
+
+ vectorMapJoinHashTable = VectorMapJoinOptimizedCreateHashTable.createHashTable(conf,
+ currentSmallTable);
+ needHashTableSetup = true;
+
+ LOG.info(CLASS_NAME + " reloadHashTable!");
+ }
+
+ @Override
+ protected void reProcessBigTable(HybridHashTableContainer.HashPartition partition)
+ throws HiveException, IOException {
+
+ LOG.info(CLASS_NAME + " reProcessBigTable enter...");
+
+ int rowCount = 0;
+ int batchCount = 0;
+
+ try {
+ VectorMapJoinRowBytesContainer bigTable = partition.getMatchfileRowBytesContainer();
+ bigTable.prepareForReading();
+
+ while (bigTable.readNext()) {
+ rowCount++;
+
+ byte[] bytes = bigTable.currentBytes();
+ int offset = bigTable.currentOffset();
+ int length = bigTable.currentLength();
+
+// LOG.info(CLASS_NAME + " reProcessBigTable serialized row #" + rowCount + ", offset " + offset + ", length " + length);
+
+ bigTableVectorDeserializeRow.setBytes(bytes, offset, length);
+ bigTableVectorDeserializeRow.deserializeByValue(spillReplayBatch, spillReplayBatch.size);
+ spillReplayBatch.size++;
+
+ if (spillReplayBatch.size == VectorizedRowBatch.DEFAULT_SIZE) {
+ LOG.info("reProcessBigTable going to call process with spillReplayBatch.size " + spillReplayBatch.size + " rows");
+ process(spillReplayBatch, posBigTable); // call process once we have a full batch
+ spillReplayBatch.reset();
+ batchCount++;
+ }
+ }
+ // Process the row batch that has less than DEFAULT_SIZE rows
+ if (spillReplayBatch.size > 0) {
+ LOG.info("reProcessBigTable going to call process with spillReplayBatch.size " + spillReplayBatch.size + " rows");
+ process(spillReplayBatch, posBigTable);
+ spillReplayBatch.reset();
+ batchCount++;
+ }
+ bigTable.clear();
+ } catch (Exception e) {
+ LOG.info(CLASS_NAME + " reProcessBigTable exception! " + e);
+ throw new HiveException(e);
+ }
+
+ LOG.info(CLASS_NAME + " reProcessBigTable exit! " + rowCount + " row processed and " + batchCount + " batches processed");
+ }
+
+
+ //-----------------------------------------------------------------------------------------------
+
+ /*
+ * Forwarding.
+ */
+
+ /**
+ * Forward the big table batch to the children.
+ *
+ * @param batch
+ * The big table batch.
+ */
+ public void forwardBigTableBatch(VectorizedRowBatch batch) throws HiveException {
+
+ // Save original projection.
+ int[] originalProjections = batch.projectedColumns;
+ int originalProjectionSize = batch.projectionSize;
+
+ // Project with the output of our operator.
+ batch.projectionSize = outputProjection.length;
+ batch.projectedColumns = outputProjection;
+
+ forward(batch, null);
+
+ // Revert the projected columns back, because batch can be re-used by our parent operators.
+ batch.projectionSize = originalProjectionSize;
+ batch.projectedColumns = originalProjections;
+ }
+
+
+ /**
+ * Forward the overflow batch and reset the batch.
+ */
+ protected void forwardOverflow() throws HiveException {
+ forward(overflowBatch, null);
+ overflowBatch.reset();
+ }
+
+ /**
+ * Forward the overflow batch, but do not reset the batch.
+ */
+ private void forwardOverflowNoReset() throws HiveException {
+ forward(overflowBatch, null);
+ }
+
+ /*
+ * Close.
+ */
+
+ /**
+ * On close, make sure a partially filled overflow batch gets forwarded.
+ */
+ @Override
+ public void closeOp(boolean aborted) throws HiveException {
+ super.closeOp(aborted);
+ if (!aborted && overflowBatch.size > 0) {
+ forwardOverflow();
+ }
+ LOG.info("VectorMapJoinInnerLongOperator closeOp " + batchCounter + " batches processed");
+ }
+
+ //-----------------------------------------------------------------------------------------------
+
+ /*
+ * Debug.
+ */
+
+ public static String intArrayToRangesString(int selection[], int size) {
+ if (size == 0) {
+ return "[]";
+ }
+
+ StringBuilder sb = new StringBuilder();
+
+ // Use ranges and duplicate multipliers to reduce the size of the display.
+ sb.append("[");
+ int firstIndex = 0;
+ int firstValue = selection[0];
+
+ boolean duplicates = false;
+
+ int i = 1;
+ for ( ; i < size; i++) {
+ int newValue = selection[i];
+ if (newValue == selection[i - 1]) {
+
+ // Duplicate.
+ duplicates = true;
+
+ if (newValue == firstValue) {
+ continue;
+ } else {
+ // Prior none, singleton, or range?
+ int priorRangeLength = i - 1 - firstIndex;
+
+ if (priorRangeLength == 0) {
+ continue;
+ }
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(firstValue);
+ if (priorRangeLength > 1) {
+ sb.append(".." + selection[i - 2]);
+ }
+ firstIndex = i - 1;
+ firstValue = newValue;
+ continue;
+ }
+ } else {
+ if (duplicates) {
+ int numDuplicates = i - firstIndex;
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(numDuplicates + "*" + firstValue);
+ duplicates = false;
+ firstIndex = i;
+ firstValue = newValue;
+ continue;
+ } if (newValue == selection[i - 1] + 1) {
+ // Continue range..
+ continue;
+ } else {
+ // Prior singleton or range?
+ int priorRangeLength = i - firstIndex;
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(firstValue);
+ if (priorRangeLength > 1) {
+ sb.append(".." + selection[i - 1]);
+ }
+ firstIndex = i;
+ firstValue = newValue;
+ continue;
+ }
+ }
+ }
+ if (duplicates) {
+ int numDuplicates = i - firstIndex;
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(numDuplicates + "*" + firstValue);
+ } else {
+ // Last singleton or range?
+ int priorRangeLength = i - firstIndex;
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(firstValue);
+ if (priorRangeLength > 1) {
+ sb.append(".." + selection[i - 1]);
+ }
+ }
+ sb.append("]");
+ return sb.toString();
+ }
+
+ public static String longArrayToRangesString(long selection[], int size) {
+ if (size == 0) {
+ return "[]";
+ }
+
+ StringBuilder sb = new StringBuilder();
+
+ // Use ranges and duplicate multipliers to reduce the size of the display.
+ sb.append("[");
+ int firstIndex = 0;
+ long firstValue = selection[0];
+
+ boolean duplicates = false;
+
+ int i = 1;
+ for ( ; i < size; i++) {
+ long newValue = selection[i];
+ if (newValue == selection[i - 1]) {
+
+ // Duplicate.
+ duplicates = true;
+
+ if (newValue == firstValue) {
+ continue;
+ } else {
+ // Prior none, singleton, or range?
+ int priorRangeLength = i - 1 - firstIndex;
+
+ if (priorRangeLength == 0) {
+ continue;
+ }
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(firstValue);
+ if (priorRangeLength > 1) {
+ sb.append(".." + selection[i - 2]);
+ }
+ firstIndex = i - 1;
+ firstValue = newValue;
+ continue;
+ }
+ } else {
+ if (duplicates) {
+ int numDuplicates = i - firstIndex;
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(numDuplicates + "*" + firstValue);
+ duplicates = false;
+ firstIndex = i;
+ firstValue = newValue;
+ continue;
+ } if (newValue == selection[i - 1] + 1) {
+ // Continue range..
+ continue;
+ } else {
+ // Prior singleton or range?
+ int priorRangeLength = i - firstIndex;
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(firstValue);
+ if (priorRangeLength > 1) {
+ sb.append(".." + selection[i - 1]);
+ }
+ firstIndex = i;
+ firstValue = newValue;
+ continue;
+ }
+ }
+ }
+ if (duplicates) {
+ int numDuplicates = i - firstIndex;
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(numDuplicates + "*" + firstValue);
+ } else {
+ // Last singleton or range?
+ int priorRangeLength = i - firstIndex;
+ if (firstIndex > 0) {
+ sb.append(",");
+ }
+ sb.append(firstValue);
+ if (priorRangeLength > 1) {
+ sb.append(".." + selection[i - 1]);
+ }
+ }
+ sb.append("]");
+ return sb.toString();
+ }
+}
\ No newline at end of file