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 2015/11/01 05:32:53 UTC
[08/10] hive git commit: HIVE-12290 Native Vector ReduceSink (Matt
McCline, reviewed by Gopal V)
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/common/src/java/org/apache/hive/common/util/HashCodeUtil.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hive/common/util/HashCodeUtil.java b/common/src/java/org/apache/hive/common/util/HashCodeUtil.java
new file mode 100644
index 0000000..fa30273
--- /dev/null
+++ b/common/src/java/org/apache/hive/common/util/HashCodeUtil.java
@@ -0,0 +1,132 @@
+/**
+ * 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.hive.common.util;
+
+/*
+ * Common hash code routines.
+ */
+public class HashCodeUtil {
+
+ public static int calculateIntHashCode(int key) {
+ key = ~key + (key << 15); // key = (key << 15) - key - 1;
+ key = key ^ (key >>> 12);
+ key = key + (key << 2);
+ key = key ^ (key >>> 4);
+ key = key * 2057; // key = (key + (key << 3)) + (key << 11);
+ key = key ^ (key >>> 16);
+ return key;
+ }
+
+ public static int calculateLongHashCode(long key) {
+
+ key = (~key) + (key << 21); // key = (key << 21) - key - 1;
+ key = key ^ (key >>> 24);
+ key = (key + (key << 3)) + (key << 8); // key * 265
+ key = key ^ (key >>> 14);
+ key = (key + (key << 2)) + (key << 4); // key * 21
+ key = key ^ (key >>> 28);
+ key = key + (key << 31);
+
+ return (int) key;
+ }
+
+ public static void calculateLongArrayHashCodes(long[] longs, int[] hashCodes, final int count) {
+ long key;
+ for (int v = 0; v < count; v++) {
+
+ key = longs[v];
+
+ // Hash code logic from calculateLongHashCode.
+ key = (~key) + (key << 21); // key = (key << 21) - key - 1;
+ key = key ^ (key >>> 24);
+ key = (key + (key << 3)) + (key << 8); // key * 265
+ key = key ^ (key >>> 14);
+ key = (key + (key << 2)) + (key << 4); // key * 21
+ key = key ^ (key >>> 28);
+ key = key + (key << 31);
+ hashCodes[v] = (int) key;
+ }
+ }
+
+ public static int calculateBytesHashCode(byte[] keyBytes, int keyStart, int keyLength) {
+ return murmurHash(keyBytes, keyStart, keyLength);
+ }
+
+ public static void calculateBytesArrayHashCodes(byte[][] bytesArrays,
+ int[] starts, int[] lengths, int[] valueSelected, int[] hashCodes, final int count) {
+
+ for (int i = 0; i < count; i++) {
+ int batchIndex = valueSelected[i];
+ hashCodes[i] = murmurHash(bytesArrays[batchIndex], starts[batchIndex],
+ lengths[batchIndex]);
+ }
+ }
+
+ // Lifted from org.apache.hadoop.util.hash.MurmurHash... but supports offset.
+ // Must produce the same result as MurmurHash.hash with seed = 0.
+ public static int murmurHash(byte[] data, int offset, int length) {
+ int m = 0x5bd1e995;
+ int r = 24;
+
+ int h = length;
+
+ int len_4 = length >> 2;
+
+ for (int i = 0; i < len_4; i++) {
+ int i_4 = offset + (i << 2);
+ int k = data[i_4 + 3];
+ k = k << 8;
+ k = k | (data[i_4 + 2] & 0xff);
+ k = k << 8;
+ k = k | (data[i_4 + 1] & 0xff);
+ k = k << 8;
+ k = k | (data[i_4 + 0] & 0xff);
+ k *= m;
+ k ^= k >>> r;
+ k *= m;
+ h *= m;
+ h ^= k;
+ }
+
+ // avoid calculating modulo
+ int len_m = len_4 << 2;
+ int left = length - len_m;
+
+ if (left != 0) {
+ length += offset;
+ if (left >= 3) {
+ h ^= (int) data[length - 3] << 16;
+ }
+ if (left >= 2) {
+ h ^= (int) data[length - 2] << 8;
+ }
+ if (left >= 1) {
+ h ^= (int) data[length - 1];
+ }
+
+ h *= m;
+ }
+
+ h ^= h >>> 13;
+ h *= m;
+ h ^= h >>> 15;
+
+ return h;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/itests/src/test/resources/testconfiguration.properties
----------------------------------------------------------------------
diff --git a/itests/src/test/resources/testconfiguration.properties b/itests/src/test/resources/testconfiguration.properties
index 8a4c1fa..7416d32 100644
--- a/itests/src/test/resources/testconfiguration.properties
+++ b/itests/src/test/resources/testconfiguration.properties
@@ -202,9 +202,11 @@ minitez.query.files.shared=alter_merge_2_orc.q,\
update_two_cols.q,\
vector_acid3.q,\
vector_aggregate_9.q,\
+ vector_aggregate_without_gby.q,\
vector_auto_smb_mapjoin_14.q,\
vector_between_in.q,\
vector_binary_join_groupby.q,\
+ vector_bround.q,\
vector_bucket.q,\
vector_char_cast.q,\
vector_cast_constant.q,\
@@ -257,6 +259,7 @@ minitez.query.files.shared=alter_merge_2_orc.q,\
vector_non_string_partition.q,\
vector_nullsafe_join.q,\
vector_null_projection.q,\
+ vector_nvl.q,\
vector_orderby_5.q,\
vector_outer_join0.q,\
vector_outer_join1.q,\
@@ -267,7 +270,11 @@ minitez.query.files.shared=alter_merge_2_orc.q,\
vector_partition_diff_num_cols.q,\
vector_partitioned_date_time.q,\
vector_reduce_groupby_decimal.q,\
+ vector_reduce1.q,\
+ vector_reduce2.q,\
+ vector_reduce3.q,\
vector_string_concat.q,\
+ vector_struct_in.q,\
vector_varchar_4.q,\
vector_varchar_mapjoin1.q,\
vector_varchar_simple.q,\
@@ -296,6 +303,7 @@ minitez.query.files.shared=alter_merge_2_orc.q,\
vectorization_not.q,\
vectorization_part.q,\
vectorization_part_project.q,\
+ vectorization_part_varchar.q,\
vectorization_pushdown.q,\
vectorization_short_regress.q,\
vectorized_bucketmapjoin1.q,\
@@ -308,11 +316,13 @@ minitez.query.files.shared=alter_merge_2_orc.q,\
vectorized_math_funcs.q,\
vectorized_nested_mapjoin.q,\
vectorized_parquet.q,\
+ vectorized_parquet_types.q,\
vectorized_ptf.q,\
vectorized_rcfile_columnar.q,\
vectorized_shufflejoin.q,\
vectorized_string_funcs.q,\
vectorized_timestamp_funcs.q,\
+ vectorized_timestamp_ints_casts.q,\
auto_sortmerge_join_1.q,\
auto_sortmerge_join_10.q,\
auto_sortmerge_join_11.q,\
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
index b6fec61..7282228 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
@@ -1374,4 +1374,12 @@ public abstract class Operator<T extends OperatorDesc> implements Serializable,C
removeParent(parent);
}
}
+
+ public boolean getIsReduceSink() {
+ return false;
+ }
+
+ public String getReduceOutputName() {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorUtils.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorUtils.java
index 67e5c2a..4828d70 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorUtils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorUtils.java
@@ -119,12 +119,11 @@ public class OperatorUtils {
return;
}
for (Operator<? extends OperatorDesc> op : childOperators) {
- if(op.getName().equals(ReduceSinkOperator.getOperatorName())) {
- ReduceSinkOperator rs = ((ReduceSinkOperator)op);
- if (outMap.containsKey(rs.getConf().getOutputName())) {
- LOG.info("Setting output collector: " + rs + " --> "
- + rs.getConf().getOutputName());
- rs.setOutputCollector(outMap.get(rs.getConf().getOutputName()));
+ if (op.getIsReduceSink()) {
+ String outputName = op.getReduceOutputName();
+ if (outMap.containsKey(outputName)) {
+ LOG.info("Setting output collector: " + op + " --> " + outputName);
+ op.setOutputCollector(outMap.get(outputName));
}
} else {
setChildrenCollector(op.getChildOperators(), outMap);
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java
index e33c1d4..ef5ee95 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java
@@ -614,6 +614,16 @@ public class ReduceSinkOperator extends TerminalOperator<ReduceSinkDesc>
}
@Override
+ public boolean getIsReduceSink() {
+ return true;
+ }
+
+ @Override
+ public String getReduceOutputName() {
+ return conf.getOutputName();
+ }
+
+ @Override
public void setOutputCollector(OutputCollector _out) {
this.out = _out;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/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 72b2cb0..632ba4f 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
@@ -32,6 +32,7 @@ import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hive.common.util.HashCodeUtil;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator;
import org.apache.hadoop.hive.ql.exec.JoinUtil;
@@ -828,7 +829,7 @@ public class HybridHashTableContainer
* the evaluation for this big table row will be postponed.
*/
public JoinUtil.JoinResult setFromOutput(Output output) throws HiveException {
- int keyHash = WriteBuffers.murmurHash(output.getData(), 0, output.getLength());
+ int keyHash = HashCodeUtil.murmurHash(output.getData(), 0, output.getLength());
if (!bloom1.testLong(keyHash)) {
/*
@@ -981,7 +982,7 @@ public class HybridHashTableContainer
public JoinUtil.JoinResult setDirect(byte[] bytes, int offset, int length,
BytesBytesMultiHashMap.Result hashMapResult) {
- int keyHash = WriteBuffers.murmurHash(bytes, offset, length);
+ int keyHash = HashCodeUtil.murmurHash(bytes, offset, length);
partitionId = keyHash & (hashPartitions.length - 1);
// If the target hash table is on disk, spill this row to disk as well to be processed later
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java
index 58d6a9b..3dddee7 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java
@@ -60,6 +60,7 @@ import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import org.apache.hadoop.io.BinaryComparable;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Writable;
+import org.apache.hive.common.util.HashCodeUtil;
/**
* Table container that serializes keys and values using LazyBinarySerDe into
@@ -245,7 +246,7 @@ public class MapJoinBytesTableContainer
}
sanityCheckKeyForTag();
BinaryComparable b = (BinaryComparable)key;
- return WriteBuffers.murmurHash(b.getBytes(), 0, b.getLength() - (hasTag ? 1 : 0));
+ return HashCodeUtil.murmurHash(b.getBytes(), 0, b.getLength() - (hasTag ? 1 : 0));
}
/**
@@ -340,7 +341,7 @@ public class MapJoinBytesTableContainer
public int getHashFromKey() throws SerDeException {
byte[] keyBytes = key.getBytes();
int keyLength = key.getLength();
- return WriteBuffers.murmurHash(keyBytes, 0, keyLength);
+ return HashCodeUtil.murmurHash(keyBytes, 0, keyLength);
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/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 41cf953..aff5765 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
@@ -94,9 +94,9 @@ public class ReduceRecordSource implements RecordSource {
private boolean vectorized = false;
- private VectorDeserializeRow keyBinarySortableDeserializeToRow;
+ private VectorDeserializeRow<BinarySortableDeserializeRead> keyBinarySortableDeserializeToRow;
- private VectorDeserializeRow valueLazyBinaryDeserializeToRow;
+ private VectorDeserializeRow<LazyBinaryDeserializeRead> valueLazyBinaryDeserializeToRow;
private VectorizedRowBatch batch;
@@ -184,7 +184,7 @@ public class ReduceRecordSource implements RecordSource {
BinarySortableSerDe binarySortableSerDe = (BinarySortableSerDe) inputKeyDeserializer;
keyBinarySortableDeserializeToRow =
- new VectorDeserializeRow(
+ new VectorDeserializeRow<BinarySortableDeserializeRead>(
new BinarySortableDeserializeRead(
VectorizedBatchUtil.primitiveTypeInfosFromStructObjectInspector(
keyStructInspector),
@@ -194,7 +194,7 @@ public class ReduceRecordSource implements RecordSource {
final int valuesSize = valueStructInspectors.getAllStructFieldRefs().size();
if (valuesSize > 0) {
valueLazyBinaryDeserializeToRow =
- new VectorDeserializeRow(
+ new VectorDeserializeRow<LazyBinaryDeserializeRead>(
new LazyBinaryDeserializeRead(
VectorizedBatchUtil.primitiveTypeInfosFromStructObjectInspector(
valueStructInspectors)));
@@ -412,6 +412,10 @@ public class ReduceRecordSource implements RecordSource {
// a data buffer.
byte[] keyBytes = keyWritable.getBytes();
int keyLength = keyWritable.getLength();
+
+ // l4j.info("ReduceRecordSource processVectorGroup keyBytes " + keyLength + " " +
+ // VectorizedBatchUtil.displayBytes(keyBytes, 0, keyLength));
+
keyBinarySortableDeserializeToRow.setBytes(keyBytes, 0, keyLength);
keyBinarySortableDeserializeToRow.deserializeByValue(batch, 0);
for(int i = 0; i < firstValueColumnOffset; i++) {
@@ -426,6 +430,10 @@ public class ReduceRecordSource implements RecordSource {
BytesWritable valueWritable = (BytesWritable) value;
byte[] valueBytes = valueWritable.getBytes();
int valueLength = valueWritable.getLength();
+
+ // l4j.info("ReduceRecordSource processVectorGroup valueBytes " + valueLength + " " +
+ // VectorizedBatchUtil.displayBytes(valueBytes, 0, valueLength));
+
valueLazyBinaryDeserializeToRow.setBytes(valueBytes, 0, valueLength);
valueLazyBinaryDeserializeToRow.deserializeByValue(batch, rowIdx);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorDeserializeRow.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorDeserializeRow.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorDeserializeRow.java
index e621745..4d86db6 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorDeserializeRow.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorDeserializeRow.java
@@ -31,15 +31,10 @@ import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
import org.apache.hadoop.hive.ql.exec.vector.expressions.StringExpr;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.serde2.fast.DeserializeRead;
-import org.apache.hadoop.hive.serde2.objectinspector.StructField;
-import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
import org.apache.hadoop.hive.serde2.typeinfo.CharTypeInfo;
-import org.apache.hadoop.hive.serde2.typeinfo.DecimalTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
-import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import org.apache.hadoop.hive.serde2.typeinfo.VarcharTypeInfo;
import org.apache.hive.common.util.DateUtils;
@@ -53,20 +48,21 @@ import org.apache.hive.common.util.DateUtils;
* field-by-field from a serialization format into the primitive values of the VectorizedRowBatch.
*/
-public class VectorDeserializeRow {
+public final class VectorDeserializeRow<T extends DeserializeRead> {
+
private static final long serialVersionUID = 1L;
private static final Logger LOG = LoggerFactory.getLogger(VectorDeserializeRow.class);
- private DeserializeRead deserializeRead;
+ private T deserializeRead;
- private Reader[] readersByValue;
- private Reader[] readersByReference;
- private PrimitiveTypeInfo[] primitiveTypeInfos;
+ private Reader<T>[] readersByValue;
+ private Reader<T>[] readersByReference;
+ private TypeInfo[] typeInfos;
- public VectorDeserializeRow(DeserializeRead deserializeRead) {
+ public VectorDeserializeRow(T deserializeRead) {
this();
this.deserializeRead = deserializeRead;
- primitiveTypeInfos = deserializeRead.primitiveTypeInfos();
+ typeInfos = deserializeRead.typeInfos();
}
@@ -74,7 +70,7 @@ public class VectorDeserializeRow {
private VectorDeserializeRow() {
}
- private abstract class Reader {
+ private abstract class Reader<R extends DeserializeRead> {
protected int columnIndex;
Reader(int columnIndex) {
@@ -84,7 +80,7 @@ public class VectorDeserializeRow {
abstract void apply(VectorizedRowBatch batch, int batchIndex) throws IOException;
}
- private abstract class AbstractLongReader extends Reader {
+ private abstract class AbstractLongReader extends Reader<T> {
AbstractLongReader(int columnIndex) {
super(columnIndex);
@@ -277,7 +273,7 @@ public class VectorDeserializeRow {
}
}
- private abstract class AbstractDoubleReader extends Reader {
+ private abstract class AbstractDoubleReader extends Reader<T> {
AbstractDoubleReader(int columnIndex) {
super(columnIndex);
@@ -322,7 +318,7 @@ public class VectorDeserializeRow {
}
}
- private abstract class AbstractBytesReader extends Reader {
+ private abstract class AbstractBytesReader extends Reader<T> {
AbstractBytesReader(int columnIndex) {
super(columnIndex);
@@ -537,7 +533,7 @@ public class VectorDeserializeRow {
}
}
- private class HiveDecimalReader extends Reader {
+ private class HiveDecimalReader extends Reader<T> {
private DeserializeRead.ReadDecimalResults readDecimalResults;
@@ -561,10 +557,10 @@ public class VectorDeserializeRow {
}
private void addReader(int index, int outputColumn) throws HiveException {
- Reader readerByValue = null;
- Reader readerByReference = null;
+ Reader<T> readerByValue = null;
+ Reader<T> readerByReference = null;
- PrimitiveTypeInfo primitiveTypeInfo = primitiveTypeInfos[index];
+ PrimitiveTypeInfo primitiveTypeInfo = (PrimitiveTypeInfo) typeInfos[index];
PrimitiveCategory primitiveCategory = primitiveTypeInfo.getPrimitiveCategory();
switch (primitiveCategory) {
// case VOID:
@@ -642,10 +638,10 @@ public class VectorDeserializeRow {
public void init(int[] outputColumns) throws HiveException {
- readersByValue = new Reader[primitiveTypeInfos.length];
- readersByReference = new Reader[primitiveTypeInfos.length];
+ readersByValue = new Reader[typeInfos.length];
+ readersByReference = new Reader[typeInfos.length];
- for (int i = 0; i < primitiveTypeInfos.length; i++) {
+ for (int i = 0; i < typeInfos.length; i++) {
int outputColumn = outputColumns[i];
addReader(i, outputColumn);
}
@@ -653,10 +649,10 @@ public class VectorDeserializeRow {
public void init(List<Integer> outputColumns) throws HiveException {
- readersByValue = new Reader[primitiveTypeInfos.length];
- readersByReference = new Reader[primitiveTypeInfos.length];
+ readersByValue = new Reader[typeInfos.length];
+ readersByReference = new Reader[typeInfos.length];
- for (int i = 0; i < primitiveTypeInfos.length; i++) {
+ for (int i = 0; i < typeInfos.length; i++) {
int outputColumn = outputColumns.get(i);
addReader(i, outputColumn);
}
@@ -664,10 +660,10 @@ public class VectorDeserializeRow {
public void init(int startColumn) throws HiveException {
- readersByValue = new Reader[primitiveTypeInfos.length];
- readersByReference = new Reader[primitiveTypeInfos.length];
+ readersByValue = new Reader[typeInfos.length];
+ readersByReference = new Reader[typeInfos.length];
- for (int i = 0; i < primitiveTypeInfos.length; i++) {
+ for (int i = 0; i < typeInfos.length; i++) {
int outputColumn = startColumn + i;
addReader(i, outputColumn);
}
@@ -709,12 +705,12 @@ public class VectorDeserializeRow {
private void throwMoreDetailedException(IOException e, int index) throws EOFException {
StringBuilder sb = new StringBuilder();
- sb.append("Detail: \"" + e.toString() + "\" occured for field " + index + " of " + primitiveTypeInfos.length + " fields (");
- for (int i = 0; i < primitiveTypeInfos.length; i++) {
+ sb.append("Detail: \"" + e.toString() + "\" occured for field " + index + " of " + typeInfos.length + " fields (");
+ for (int i = 0; i < typeInfos.length; i++) {
if (i > 0) {
sb.append(", ");
}
- sb.append(primitiveTypeInfos[i].getPrimitiveCategory().name());
+ sb.append(((PrimitiveTypeInfo) typeInfos[i]).getPrimitiveCategory().name());
}
sb.append(")");
throw new EOFException(sb.toString());
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/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 e2934e3..fe889b5 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
@@ -22,8 +22,6 @@ import java.io.IOException;
import java.sql.Timestamp;
import java.util.List;
-import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
-import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.serde2.ByteStream.Output;
import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
@@ -46,11 +44,11 @@ import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
* Note that when serializing a row, the logical mapping using selected in use has already
* been performed.
*/
-public class VectorSerializeRow {
+public final class VectorSerializeRow<T extends SerializeWrite> {
- private SerializeWrite serializeWrite;
+ private T serializeWrite;
- public VectorSerializeRow(SerializeWrite serializeWrite) {
+ public VectorSerializeRow(T serializeWrite) {
this();
this.serializeWrite = serializeWrite;
}
@@ -59,7 +57,7 @@ public class VectorSerializeRow {
private VectorSerializeRow() {
}
- private abstract class Writer {
+ private abstract class Writer<W extends SerializeWrite> {
protected int columnIndex;
Writer(int columnIndex) {
@@ -69,7 +67,7 @@ public class VectorSerializeRow {
abstract boolean apply(VectorizedRowBatch batch, int batchIndex) throws IOException;
}
- private abstract class AbstractLongWriter extends Writer {
+ private abstract class AbstractLongWriter extends Writer<T> {
AbstractLongWriter(int columnIndex) {
super(columnIndex);
@@ -351,7 +349,7 @@ public class VectorSerializeRow {
}
}
- private abstract class AbstractDoubleWriter extends Writer {
+ private abstract class AbstractDoubleWriter extends Writer<T> {
AbstractDoubleWriter(int columnIndex) {
super(columnIndex);
@@ -418,7 +416,7 @@ public class VectorSerializeRow {
}
}
- private class StringWriter extends Writer {
+ private class StringWriter extends Writer<T> {
StringWriter(int columnIndex) {
super(columnIndex);
@@ -449,7 +447,7 @@ public class VectorSerializeRow {
}
}
- private class BinaryWriter extends Writer {
+ private class BinaryWriter extends Writer<T> {
BinaryWriter(int columnIndex) {
super(columnIndex);
@@ -480,7 +478,7 @@ public class VectorSerializeRow {
}
}
- private class HiveDecimalWriter extends Writer {
+ private class HiveDecimalWriter extends Writer<T> {
protected HiveDecimalWritable[] vector;
HiveDecimalWriter(int columnIndex) {
@@ -511,10 +509,10 @@ public class VectorSerializeRow {
}
}
- private Writer[] writers;
+ private Writer<T>[] writers;
- private Writer createWriter(TypeInfo typeInfo, int columnIndex) throws HiveException {
- Writer writer;
+ private Writer<T> createWriter(TypeInfo typeInfo, int columnIndex) throws HiveException {
+ Writer<T> writer;
Category category = typeInfo.getCategory();
switch (category) {
case PRIMITIVE:
@@ -588,7 +586,7 @@ public class VectorSerializeRow {
String typeName = typeNames.get(i);
TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(typeName);
int columnIndex = columnMap[i];
- Writer writer = createWriter(typeInfo, columnIndex);
+ Writer<T> writer = createWriter(typeInfo, columnIndex);
writers[i] = writer;
}
}
@@ -599,18 +597,18 @@ public class VectorSerializeRow {
for (int i = 0; i < typeNames.size(); i++) {
String typeName = typeNames.get(i);
TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(typeName);
- Writer writer = createWriter(typeInfo, i);
+ Writer<T> writer = createWriter(typeInfo, i);
writers[i] = writer;
}
}
- public void init(PrimitiveTypeInfo[] primitiveTypeInfos, List<Integer> columnMap)
+ public void init(TypeInfo[] typeInfos, int[] 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 = new Writer[typeInfos.length];
+ for (int i = 0; i < typeInfos.length; i++) {
+ int columnIndex = columnMap[i];
+ Writer<T> writer = createWriter(typeInfos[i], columnIndex);
writers[i] = writer;
}
}
@@ -627,17 +625,31 @@ public class VectorSerializeRow {
serializeWrite.setAppend(output);
}
+ private boolean hasAnyNulls;
+ private boolean isAllNulls;
+
/*
* 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.
*/
- public boolean serializeWrite(VectorizedRowBatch batch, int batchIndex) throws IOException {
- boolean anyNulls = false;
- for (Writer writer : writers) {
+ public void serializeWrite(VectorizedRowBatch batch, int batchIndex) throws IOException {
+
+ hasAnyNulls = false;
+ isAllNulls = true;
+ for (Writer<T> writer : writers) {
if (!writer.apply(batch, batchIndex)) {
- anyNulls = true;
+ hasAnyNulls = true;
+ } else {
+ isAllNulls = false;
}
}
- return anyNulls;
+ }
+
+ public boolean getHasAnyNulls() {
+ return hasAnyNulls;
+ }
+
+ public boolean getIsAllNulls() {
+ return isAllNulls;
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/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
deleted file mode 100644
index c67945b..0000000
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorSerializeRowNoNulls.java
+++ /dev/null
@@ -1,412 +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;
-
-import java.io.IOException;
-import java.sql.Timestamp;
-import java.util.List;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
-import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.serde2.ByteStream.Output;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
-import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
-import org.apache.hadoop.hive.serde2.fast.SerializeWrite;
-import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
-import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
-import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
-
-/**
- * This class serializes columns from a row in a VectorizedRowBatch into a serialization format.
- *
- * The caller provides the hive type names and column numbers in the order desired to
- * serialize.
- *
- * This class uses an provided SerializeWrite object to directly serialize by writing
- * field-by-field into a serialization format from the primitive values of the VectorizedRowBatch.
- *
- * Note that when serializing a row, the logical mapping using selected in use has already
- * been performed.
- *
- * NOTE: This class is a variation of VectorSerializeRow for serialization of columns that
- * have no nulls.
- */
-public class VectorSerializeRowNoNulls {
- private static final Logger LOG = LoggerFactory.getLogger(VectorSerializeRowNoNulls.class.getName());
-
- private SerializeWrite serializeWrite;
-
- public VectorSerializeRowNoNulls(SerializeWrite serializeWrite) {
- this();
- this.serializeWrite = serializeWrite;
- }
-
- // Not public since we must have the serialize write object.
- private VectorSerializeRowNoNulls() {
- }
-
- private abstract class Writer {
- protected int columnIndex;
-
- Writer(int columnIndex) {
- this.columnIndex = columnIndex;
- }
-
- abstract void apply(VectorizedRowBatch batch, int batchIndex) throws IOException;
- }
-
- private abstract class AbstractLongWriter extends Writer {
-
- AbstractLongWriter(int columnIndex) {
- super(columnIndex);
- }
- }
-
- private class BooleanWriter extends AbstractLongWriter {
-
- BooleanWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- serializeWrite.writeBoolean(colVector.vector[colVector.isRepeating ? 0 : batchIndex] != 0);
- }
- }
-
- private class ByteWriter extends AbstractLongWriter {
-
- ByteWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- serializeWrite.writeByte((byte) colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private class ShortWriter extends AbstractLongWriter {
-
- ShortWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- serializeWrite.writeShort((short) colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private class IntWriter extends AbstractLongWriter {
-
- IntWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- serializeWrite.writeInt((int) colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private class LongWriter extends AbstractLongWriter {
-
- LongWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- serializeWrite.writeLong(colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private class DateWriter extends AbstractLongWriter {
-
- DateWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- serializeWrite.writeDate((int) colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private class TimestampWriter extends AbstractLongWriter {
-
- Timestamp scratchTimestamp;
-
- TimestampWriter(int columnIndex) {
- super(columnIndex);
- scratchTimestamp = new Timestamp(0);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- TimestampUtils.assignTimeInNanoSec(colVector.vector[colVector.isRepeating ? 0 : batchIndex], scratchTimestamp);
- serializeWrite.writeTimestamp(scratchTimestamp);
- }
- }
-
- private class IntervalYearMonthWriter extends AbstractLongWriter {
-
- IntervalYearMonthWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- serializeWrite.writeHiveIntervalYearMonth((int) colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private class IntervalDayTimeWriter extends AbstractLongWriter {
-
- IntervalDayTimeWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- LongColumnVector colVector = (LongColumnVector) batch.cols[columnIndex];
- serializeWrite.writeHiveIntervalDayTime(colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private abstract class AbstractDoubleWriter extends Writer {
-
- AbstractDoubleWriter(int columnIndex) {
- super(columnIndex);
- }
- }
-
- private class FloatWriter extends AbstractDoubleWriter {
-
- FloatWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- DoubleColumnVector colVector = (DoubleColumnVector) batch.cols[columnIndex];
- serializeWrite.writeFloat((float) colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private class DoubleWriter extends AbstractDoubleWriter {
-
- DoubleWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- DoubleColumnVector colVector = (DoubleColumnVector) batch.cols[columnIndex];
- serializeWrite.writeDouble(colVector.vector[colVector.isRepeating ? 0 : batchIndex]);
- }
- }
-
- private class StringWriter extends Writer {
-
- StringWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- BytesColumnVector colVector = (BytesColumnVector) batch.cols[columnIndex];
-
- if (colVector.isRepeating) {
- serializeWrite.writeString(colVector.vector[0], colVector.start[0], colVector.length[0]);
- } else {
- serializeWrite.writeString(colVector.vector[batchIndex], colVector.start[batchIndex], colVector.length[batchIndex]);
- }
- }
- }
-
- private class BinaryWriter extends Writer {
-
- BinaryWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- BytesColumnVector colVector = (BytesColumnVector) batch.cols[columnIndex];
-
- if (colVector.isRepeating) {
- serializeWrite.writeBinary(colVector.vector[0], colVector.start[0], colVector.length[0]);
- } else {
- serializeWrite.writeBinary(colVector.vector[batchIndex], colVector.start[batchIndex], colVector.length[batchIndex]);
- }
- }
- }
-
- private class HiveDecimalWriter extends Writer {
-
- HiveDecimalWriter(int columnIndex) {
- super(columnIndex);
- }
-
- @Override
- void apply(VectorizedRowBatch batch, int batchIndex) throws IOException {
- DecimalColumnVector colVector = (DecimalColumnVector) batch.cols[columnIndex];
- serializeWrite.writeHiveDecimal(colVector.vector[colVector.isRepeating ? 0 : batchIndex].getHiveDecimal());
- }
- }
-
- private Writer[] writers;
-
- private Writer createWriter(TypeInfo typeInfo, int columnIndex) throws HiveException {
- Writer writer;
- Category category = typeInfo.getCategory();
- switch (category) {
- case PRIMITIVE:
- {
- PrimitiveTypeInfo primitiveTypeInfo = (PrimitiveTypeInfo) typeInfo;
- PrimitiveCategory primitiveCategory = primitiveTypeInfo.getPrimitiveCategory();
- switch (primitiveCategory) {
- // case VOID:
- // UNDONE:
- // break;
- case BOOLEAN:
- writer = new BooleanWriter(columnIndex);
- break;
- case BYTE:
- writer = new ByteWriter(columnIndex);
- break;
- case SHORT:
- writer = new ShortWriter(columnIndex);
- break;
- case INT:
- writer = new IntWriter(columnIndex);
- break;
- case LONG:
- writer = new LongWriter(columnIndex);
- break;
- case DATE:
- writer = new DateWriter(columnIndex);
- break;
- case TIMESTAMP:
- writer = new TimestampWriter(columnIndex);
- break;
- case FLOAT:
- writer = new FloatWriter(columnIndex);
- break;
- case DOUBLE:
- writer = new DoubleWriter(columnIndex);
- break;
- case STRING:
- case CHAR:
- case VARCHAR:
- // We store CHAR and VARCHAR without pads, so use STRING writer class.
- writer = new StringWriter(columnIndex);
- break;
- case BINARY:
- writer = new BinaryWriter(columnIndex);
- break;
- case DECIMAL:
- writer = new HiveDecimalWriter(columnIndex);
- break;
- case INTERVAL_YEAR_MONTH:
- writer = new IntervalYearMonthWriter(columnIndex);
- break;
- case INTERVAL_DAY_TIME:
- writer = new IntervalDayTimeWriter(columnIndex);
- break;
- default:
- throw new HiveException("Unexpected primitive type category " + primitiveCategory);
- }
- }
- break;
- default:
- throw new HiveException("Unexpected type category " + category);
- }
- return writer;
- }
-
- 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);
- TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(typeName);
- int columnIndex = columnMap[i];
- Writer writer = createWriter(typeInfo, columnIndex);
- writers[i] = writer;
- }
- }
-
- 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);
- TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(typeName);
- Writer writer = createWriter(typeInfo, i);
- writers[i] = writer;
- }
- }
-
- 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;
- }
-
- public void setOutput(Output output) {
- 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.
- */
- public void serializeWriteNoNulls(VectorizedRowBatch batch, int batchIndex) throws IOException {
- for (Writer writer : writers) {
- writer.apply(batch, batchIndex);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeries.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeries.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeries.java
new file mode 100644
index 0000000..ac6c4b8
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeries.java
@@ -0,0 +1,98 @@
+/**
+ * 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.keyseries;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+
+/**
+ * An abstraction of keys within a VectorizedRowBatch.
+ *
+ * A key is one or more columns.
+ *
+ * When there is a sequential "run" of equal keys, they are collapsed and represented by a
+ * duplicate count.
+ *
+ * The batch of keys (with sequential duplicates collapsed) is called a series.
+ *
+ * A key can be all null, or a key with no or some nulls.
+ *
+ * All keys have a duplicate count.
+ *
+ * A key with no or some nulls has:
+ * 1) A hash code.
+ * 2) Key values and other value(s) defined by other interfaces.
+ *
+ * The key series is logically indexed. That is, if batch.selectedInUse is true, the indices
+ * will be logical and need to be mapped through batch.selected to get the physical batch
+ * indices. Otherwise, the indices are physical batch indices.
+ */
+public interface VectorKeySeries {
+
+ /**
+ * Process a non-empty batch of rows and compute a key series.
+ *
+ * The key series will be positioned to the beginning.
+ *
+ * @param batch
+ * @throws IOException
+ */
+ void processBatch(VectorizedRowBatch batch) throws IOException;
+
+ /**
+ * Position to the beginning of the key series.
+ */
+ void positionToFirst();
+
+ /**
+ * @return the current logical index of the first row of the current key.
+ * The next duplicate count rows have the same key.
+ */
+ int getCurrentLogical();
+
+ /**
+ * @return true when the current key is all nulls.
+ */
+ boolean getCurrentIsAllNull();
+
+ /**
+ * @return the number of duplicate keys of the current key.
+ */
+ int getCurrentDuplicateCount();
+
+
+ /**
+ * @return true when there is at least one null in the current key.
+ * Only valid when getCurrentIsAllNull is false. Otherwise, undefined.
+ */
+ boolean getCurrentHasAnyNulls();
+
+ /**
+ * @return the hash code of the current key.
+ * Only valid when getCurrentIsAllNull is false. Otherwise, undefined.
+ */
+ int getCurrentHashCode();
+
+ /**
+ * Move to the next key.
+ * @return true when there is another key. Otherwise, the key series is complete.
+ */
+ boolean next();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesBytesSerialized.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesBytesSerialized.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesBytesSerialized.java
new file mode 100644
index 0000000..81a8455
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesBytesSerialized.java
@@ -0,0 +1,271 @@
+/**
+ * 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.keyseries;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.StringExpr;
+import org.apache.hadoop.hive.serde2.fast.SerializeWrite;
+import com.google.common.base.Preconditions;
+
+/**
+ * A key series of a single column of byte array keys where the keys get serialized.
+ */
+public class VectorKeySeriesBytesSerialized<T extends SerializeWrite>
+ extends VectorKeySeriesSerializedImpl<T> implements VectorKeySeriesSerialized {
+
+ private final int columnNum;
+
+ private int outputStartPosition;
+
+ public VectorKeySeriesBytesSerialized(int columnNum, T serializeWrite) {
+ super(serializeWrite);
+ this.columnNum = columnNum;
+ }
+
+ @Override
+ public void processBatch(VectorizedRowBatch batch) throws IOException {
+
+ currentBatchSize = batch.size;
+ Preconditions.checkState(currentBatchSize > 0);
+
+ BytesColumnVector bytesColVector = (BytesColumnVector) batch.cols[columnNum];
+
+ byte[][] vectorBytesArrays = bytesColVector.vector;
+ int[] vectorStarts = bytesColVector.start;
+ int[] vectorLengths = bytesColVector.length;
+
+ // The serialize routine uses this to build serializedKeyLengths.
+ outputStartPosition = 0;
+ output.reset();
+
+ if (bytesColVector.isRepeating){
+ duplicateCounts[0] = currentBatchSize;
+ if (bytesColVector.noNulls || !bytesColVector.isNull[0]) {
+ seriesIsAllNull[0] = false;
+ serialize(0, vectorBytesArrays[0], vectorStarts[0], vectorLengths[0]);
+ nonNullKeyCount = 1;
+ } else {
+ seriesIsAllNull[0] = true;
+ nonNullKeyCount = 0;
+ }
+ seriesCount = 1;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ } else {
+ seriesCount = 0;
+ nonNullKeyCount = 0;
+ if (batch.selectedInUse) {
+ int[] selected = batch.selected;
+ if (bytesColVector.noNulls) {
+
+ duplicateCounts[0] = 1;
+ int index;
+ index = selected[0];
+ byte[] prevKeyBytes = vectorBytesArrays[index];
+ int prevKeyStart = vectorStarts[index];
+ int prevKeyLength = vectorLengths[index];
+ serialize(0, prevKeyBytes, prevKeyStart, prevKeyLength);
+
+ int currentKeyStart;
+ int currentKeyLength;
+ byte[] currentKeyBytes;
+ for (int logical = 1; logical < currentBatchSize; logical++) {
+ index = selected[logical];
+ currentKeyBytes = vectorBytesArrays[index];
+ currentKeyStart = vectorStarts[index];
+ currentKeyLength = vectorLengths[index];
+ if (StringExpr.equal(prevKeyBytes, prevKeyStart, prevKeyLength,
+ currentKeyBytes, currentKeyStart, currentKeyLength)) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ serialize(seriesCount, currentKeyBytes, currentKeyStart, currentKeyLength);
+ prevKeyBytes = currentKeyBytes;
+ prevKeyStart = currentKeyStart;
+ prevKeyLength = currentKeyLength;
+ }
+ }
+ Arrays.fill(seriesIsAllNull, 0, ++seriesCount, false);
+ nonNullKeyCount = seriesCount;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ } else {
+ boolean[] isNull = bytesColVector.isNull;
+
+ boolean prevKeyIsNull;
+ byte[] prevKeyBytes = null;
+ int prevKeyStart = 0;
+ int prevKeyLength = 0;
+ duplicateCounts[0] = 1;
+ int index = selected[0];
+ if (isNull[index]) {
+ seriesIsAllNull[0] = true;
+ prevKeyIsNull = true;
+ } else {
+ seriesIsAllNull[0] = false;
+ prevKeyIsNull = false;
+ prevKeyBytes = vectorBytesArrays[index];
+ prevKeyStart = vectorStarts[index];
+ prevKeyLength = vectorLengths[index];
+ serialize(0, prevKeyBytes, prevKeyStart, prevKeyLength);
+ nonNullKeyCount = 1;
+ }
+
+ int currentKeyStart;
+ int currentKeyLength;
+ byte[] currentKeyBytes;
+ for (int logical = 1; logical < currentBatchSize; logical++) {
+ index = selected[logical];
+ if (isNull[index]) {
+ if (prevKeyIsNull) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = true;
+ prevKeyIsNull = true;
+ }
+ } else {
+ currentKeyBytes = vectorBytesArrays[index];
+ currentKeyStart = vectorStarts[index];
+ currentKeyLength = vectorLengths[index];
+ if (!prevKeyIsNull &&
+ StringExpr.equal(prevKeyBytes, prevKeyStart, prevKeyLength,
+ currentKeyBytes, currentKeyStart, currentKeyLength)) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = false;
+ serialize(nonNullKeyCount++, currentKeyBytes, currentKeyStart, currentKeyLength);
+ prevKeyIsNull = false;
+ prevKeyBytes = currentKeyBytes;
+ prevKeyStart = currentKeyStart;
+ prevKeyLength = currentKeyLength;
+ }
+ }
+ }
+ seriesCount++;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ }
+ } else {
+
+ // NOT selectedInUse
+
+ if (bytesColVector.noNulls) {
+
+ duplicateCounts[0] = 1;
+ byte[] prevKeyBytes = vectorBytesArrays[0];
+ int prevKeyStart = vectorStarts[0];
+ int prevKeyLength = vectorLengths[0];
+ serialize(0, prevKeyBytes, prevKeyStart, prevKeyLength);
+
+ int currentKeyStart;
+ int currentKeyLength;
+ byte[] currentKeyBytes;
+ for (int index = 1; index < currentBatchSize; index++) {
+ currentKeyBytes = vectorBytesArrays[index];
+ currentKeyStart = vectorStarts[index];
+ currentKeyLength = vectorLengths[index];
+ if (StringExpr.equal(prevKeyBytes, prevKeyStart, prevKeyLength,
+ currentKeyBytes, currentKeyStart, currentKeyLength)) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ serialize(seriesCount, currentKeyBytes, currentKeyStart, currentKeyLength);
+ prevKeyBytes = currentKeyBytes;
+ prevKeyStart = currentKeyStart;
+ prevKeyLength = currentKeyLength;
+ }
+ }
+ Arrays.fill(seriesIsAllNull, 0, ++seriesCount, false);
+ nonNullKeyCount = seriesCount;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ } else {
+ boolean[] isNull = bytesColVector.isNull;
+
+ boolean prevKeyIsNull;
+ byte[] prevKeyBytes = null;
+ int prevKeyStart = 0;
+ int prevKeyLength = 0;
+ duplicateCounts[0] = 1;
+ if (isNull[0]) {
+ seriesIsAllNull[0] = true;
+ prevKeyIsNull = true;
+ } else {
+ seriesIsAllNull[0] = false;
+ prevKeyIsNull = false;
+ prevKeyBytes = vectorBytesArrays[0];
+ prevKeyStart = vectorStarts[0];
+ prevKeyLength = vectorLengths[0];
+ serialize(0, prevKeyBytes, prevKeyStart, prevKeyLength);
+ nonNullKeyCount = 1;
+ }
+
+ byte[] currentKeyBytes;
+ int currentKeyStart;
+ int currentKeyLength;
+ for (int index = 1; index < currentBatchSize; index++) {
+ if (isNull[index]) {
+ if (prevKeyIsNull) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = true;
+ prevKeyIsNull = true;
+ }
+ } else {
+ currentKeyBytes = vectorBytesArrays[index];
+ currentKeyStart = vectorStarts[index];
+ currentKeyLength = vectorLengths[index];
+ if (!prevKeyIsNull &&
+ StringExpr.equal(prevKeyBytes, prevKeyStart, prevKeyLength,
+ currentKeyBytes, currentKeyStart, currentKeyLength)) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = false;
+ serialize(nonNullKeyCount++, currentKeyBytes, currentKeyStart, currentKeyLength);
+ prevKeyIsNull = false;
+ prevKeyBytes = currentKeyBytes;
+ prevKeyStart = currentKeyStart;
+ prevKeyLength = currentKeyLength;
+ }
+ }
+ }
+ seriesCount++;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ }
+ }
+ }
+
+ // Finally.
+ computeSerializedHashCodes();
+ positionToFirst();
+ Preconditions.checkState(validate());
+ }
+
+ private void serialize(int pos, byte[] bytes, int start, int length) throws IOException {
+ serializeWrite.setAppend(output);
+ serializeWrite.writeString(bytes, start, length);
+ int outputNewPosition = output.getLength();
+ serializedKeyLengths[pos] = outputNewPosition - outputStartPosition;
+ outputStartPosition = outputNewPosition;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesImpl.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesImpl.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesImpl.java
new file mode 100644
index 0000000..55e923e
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesImpl.java
@@ -0,0 +1,68 @@
+/**
+ * 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.keyseries;
+
+/**
+ * A base implementation of VectorKeySeries.
+ *
+ */
+public abstract class VectorKeySeriesImpl implements VectorKeySeries {
+
+ protected int currentLogical;
+ protected boolean currentIsAllNull;
+ protected boolean currentHasAnyNulls;
+ protected int currentDuplicateCount;
+ protected int currentHashCode;
+
+ VectorKeySeriesImpl() {
+ currentLogical = 0;
+ currentIsAllNull = false;
+
+ // Set to true by default. Only actively set in the multiple key case to support Outer Join.
+ currentHasAnyNulls = true;
+
+ currentDuplicateCount = 0;
+ currentHashCode = 0;
+ }
+
+ @Override
+ public int getCurrentLogical() {
+ return currentLogical;
+ }
+
+ @Override
+ public boolean getCurrentIsAllNull() {
+ return currentIsAllNull;
+ }
+
+ @Override
+ public boolean getCurrentHasAnyNulls() {
+ return currentHasAnyNulls;
+ }
+
+ @Override
+ public int getCurrentDuplicateCount() {
+ return currentDuplicateCount;
+ }
+
+ @Override
+ public int getCurrentHashCode() {
+ return currentHashCode;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesLongSerialized.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesLongSerialized.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesLongSerialized.java
new file mode 100644
index 0000000..a0134fd
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesLongSerialized.java
@@ -0,0 +1,249 @@
+/**
+ * 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.keyseries;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.fast.SerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * A key series of a single column of long keys where the keys get serialized.
+ */
+public class VectorKeySeriesLongSerialized<T extends SerializeWrite>
+ extends VectorKeySeriesSerializedImpl<T> implements VectorKeySeriesSerialized {
+
+ private final int columnNum;
+ private PrimitiveCategory primitiveCategory;
+
+ private int currentKeyStart;
+
+ public VectorKeySeriesLongSerialized(int columnNum, PrimitiveTypeInfo primitiveTypeInfo,
+ T serializeWrite) {
+ super(serializeWrite);
+ this.columnNum = columnNum;
+ primitiveCategory = primitiveTypeInfo.getPrimitiveCategory();
+ }
+
+ @Override
+ public void processBatch(VectorizedRowBatch batch) throws IOException {
+
+ currentBatchSize = batch.size;
+ Preconditions.checkState(currentBatchSize > 0);
+
+ LongColumnVector longColVector = (LongColumnVector) batch.cols[columnNum];
+
+ long[] vector = longColVector.vector;
+
+ // The serialize routine uses this to build serializedKeyLengths.
+ currentKeyStart = 0;
+ output.reset();
+
+ if (longColVector.isRepeating){
+ duplicateCounts[0] = currentBatchSize;
+ if (longColVector.noNulls || !longColVector.isNull[0]) {
+ seriesIsAllNull[0] = false;
+ serialize(0, vector[0]);
+ nonNullKeyCount = 1;
+ } else {
+ seriesIsAllNull[0] = true;
+ nonNullKeyCount = 0;
+ }
+ seriesCount = 1;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ } else {
+ seriesCount = 0;
+ nonNullKeyCount = 0;
+ if (batch.selectedInUse) {
+ int[] selected = batch.selected;
+ if (longColVector.noNulls) {
+
+ duplicateCounts[0] = 1;
+ long prevKey = vector[selected[0]];
+ serialize(0, prevKey);
+
+ long currentKey;
+ for (int logical = 1; logical < currentBatchSize; logical++) {
+ currentKey = vector[selected[logical]];
+ if (prevKey == currentKey) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ serialize(seriesCount, currentKey);
+ prevKey = currentKey;
+ }
+ }
+ Arrays.fill(seriesIsAllNull, 0, ++seriesCount, false);
+ nonNullKeyCount = seriesCount;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ } else {
+ boolean[] isNull = longColVector.isNull;
+
+ boolean prevKeyIsNull;
+ long prevKey = 0;
+ duplicateCounts[0] = 1;
+ int index = selected[0];
+ if (isNull[index]) {
+ seriesIsAllNull[0] = true;
+ prevKeyIsNull = true;
+ nonNullKeyCount = 0;
+ } else {
+ seriesIsAllNull[0] = false;
+ prevKeyIsNull = false;
+ prevKey = vector[index];
+ serialize(0, prevKey);
+ nonNullKeyCount = 1;
+ }
+
+ long currentKey;
+ for (int logical = 1; logical < currentBatchSize; logical++) {
+ index = selected[logical];
+ if (isNull[index]) {
+ if (prevKeyIsNull) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = true;
+ prevKeyIsNull = true;
+ }
+ } else {
+ currentKey = vector[index];
+ if (!prevKeyIsNull && prevKey == currentKey) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = false;
+ serialize(nonNullKeyCount++, currentKey);
+ prevKeyIsNull = false;
+ prevKey = currentKey;
+ }
+ }
+ }
+ seriesCount++;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ }
+ } else {
+
+ // NOT selectedInUse
+
+ if (longColVector.noNulls) {
+
+ duplicateCounts[0] = 1;
+ long prevKey = vector[0];
+ serialize(0, prevKey);
+ long currentKey;
+ for (int index = 1; index < currentBatchSize; index++) {
+ currentKey = vector[index];
+ if (prevKey == currentKey) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ serialize(seriesCount, currentKey);
+ prevKey = currentKey;
+ }
+ }
+ Arrays.fill(seriesIsAllNull, 0, ++seriesCount, false);
+ nonNullKeyCount = seriesCount;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ } else {
+ boolean[] isNull = longColVector.isNull;
+
+ boolean prevKeyIsNull;
+ long prevKey = 0;
+ duplicateCounts[0] = 1;
+ if (isNull[0]) {
+ seriesIsAllNull[0] = true;
+ prevKeyIsNull = true;
+ nonNullKeyCount = 0;
+ } else {
+ seriesIsAllNull[0] = false;
+ prevKeyIsNull = false;
+ prevKey = vector[0];
+ serialize(0, prevKey);
+ nonNullKeyCount = 1;
+ }
+
+ for (int index = 1; index < currentBatchSize; index++) {
+ if (isNull[index]) {
+ if (prevKeyIsNull) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = true;
+ prevKeyIsNull = true;
+ }
+ } else {
+ long currentKey = vector[index];
+ if (!prevKeyIsNull && prevKey == currentKey) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = false;
+ serialize(nonNullKeyCount++, currentKey);
+ prevKeyIsNull = false;
+ prevKey = currentKey;
+ }
+ }
+ }
+ seriesCount++;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ }
+ }
+ }
+
+ // Finally.
+ computeSerializedHashCodes();
+ positionToFirst();
+ Preconditions.checkState(validate());
+ }
+
+ private void serialize(int pos, long value) throws IOException {
+ serializeWrite.setAppend(output);
+
+ // UNDONE: Add support for DATE, TIMESTAMP, INTERVAL_YEAR_MONTH, INTERVAL_DAY_TIME...
+ switch (primitiveCategory) {
+ case BOOLEAN:
+ serializeWrite.writeBoolean(value != 0);
+ break;
+ case BYTE:
+ serializeWrite.writeByte((byte) value);
+ break;
+ case SHORT:
+ serializeWrite.writeShort((short) value);
+ break;
+ case INT:
+ serializeWrite.writeInt((int) value);
+ break;
+ case LONG:
+ serializeWrite.writeLong(value);
+ break;
+ default:
+ throw new RuntimeException("Unexpected primitive category " + primitiveCategory.name());
+ }
+ int outputNewPosition = output.getLength();
+ serializedKeyLengths[pos] = outputNewPosition - currentKeyStart;
+ currentKeyStart = outputNewPosition;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesMultiSerialized.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesMultiSerialized.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesMultiSerialized.java
new file mode 100644
index 0000000..2cc3531
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesMultiSerialized.java
@@ -0,0 +1,187 @@
+/**
+ * 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.keyseries;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hive.ql.exec.vector.VectorSerializeRow;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.StringExpr;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.serde2.fast.SerializeWrite;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * A key series of a multiple columns of keys where the keys get serialized.
+ * (Or, it can be 1 column).
+ */
+public class VectorKeySeriesMultiSerialized<T extends SerializeWrite>
+ extends VectorKeySeriesSerializedImpl<T> implements VectorKeySeriesSerialized {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorKeySeriesMultiSerialized.class.getName());
+
+ private VectorSerializeRow<T> keySerializeRow;
+
+ private boolean[] hasAnyNulls;
+
+ public VectorKeySeriesMultiSerialized(T serializeWrite) {
+ super(serializeWrite);
+ }
+
+ public void init(TypeInfo[] typeInfos, int[] columnNums) throws HiveException {
+ keySerializeRow = new VectorSerializeRow<T>(serializeWrite);
+ keySerializeRow.init(typeInfos, columnNums);
+ hasAnyNulls = new boolean[VectorizedRowBatch.DEFAULT_SIZE];
+ }
+
+ @Override
+ public void processBatch(VectorizedRowBatch batch) throws IOException {
+
+ currentBatchSize = batch.size;
+ Preconditions.checkState(currentBatchSize > 0);
+
+ // LOG.info("VectorKeySeriesMultiSerialized processBatch size " + currentBatchSize + " numCols " + batch.numCols + " selectedInUse " + batch.selectedInUse);
+
+ int prevKeyStart = 0;
+ int prevKeyLength;
+ int currentKeyStart = 0;
+ output.reset();
+
+ seriesCount = 0;
+ boolean prevKeyIsNull;
+ duplicateCounts[0] = 1;
+ if (batch.selectedInUse) {
+ int[] selected = batch.selected;
+ int index = selected[0];
+ keySerializeRow.setOutputAppend(output);
+ keySerializeRow.serializeWrite(batch, index);
+ if (keySerializeRow.getIsAllNulls()) {
+ seriesIsAllNull[0] = prevKeyIsNull = true;
+ prevKeyLength = 0;
+ output.setWritePosition(0);
+ nonNullKeyCount = 0;
+ } else {
+ seriesIsAllNull[0] = prevKeyIsNull = false;
+ serializedKeyLengths[0] = currentKeyStart = prevKeyLength = output.getLength();
+ hasAnyNulls[0] = keySerializeRow.getHasAnyNulls();
+ nonNullKeyCount = 1;
+ }
+
+ int keyLength;
+ for (int logical = 1; logical < currentBatchSize; logical++) {
+ index = selected[logical];
+ keySerializeRow.setOutputAppend(output);
+ keySerializeRow.serializeWrite(batch, index);
+ if (keySerializeRow.getIsAllNulls()) {
+ if (prevKeyIsNull) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = prevKeyIsNull = true;
+ }
+ output.setWritePosition(currentKeyStart);
+ } else {
+ keyLength = output.getLength() - currentKeyStart;
+ if (!prevKeyIsNull &&
+ StringExpr.equal(
+ output.getData(), prevKeyStart, prevKeyLength,
+ output.getData(), currentKeyStart, keyLength)) {
+ duplicateCounts[seriesCount]++;
+ output.setWritePosition(currentKeyStart);
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = prevKeyIsNull = false;
+ prevKeyStart = currentKeyStart;
+ serializedKeyLengths[nonNullKeyCount] = prevKeyLength = keyLength;
+ currentKeyStart += keyLength;
+ hasAnyNulls[nonNullKeyCount] = keySerializeRow.getHasAnyNulls();
+ nonNullKeyCount++;
+ }
+ }
+ }
+ seriesCount++;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ } else {
+ keySerializeRow.setOutputAppend(output);
+ keySerializeRow.serializeWrite(batch, 0);
+ if (keySerializeRow.getIsAllNulls()) {
+ seriesIsAllNull[0] = prevKeyIsNull = true;
+ prevKeyLength = 0;
+ output.setWritePosition(0);
+ nonNullKeyCount = 0;
+ } else {
+ seriesIsAllNull[0] = prevKeyIsNull = false;
+ serializedKeyLengths[0] = currentKeyStart = prevKeyLength = output.getLength();
+ hasAnyNulls[0] = keySerializeRow.getHasAnyNulls();
+ nonNullKeyCount = 1;
+ }
+
+ int keyLength;
+ for (int index = 1; index < currentBatchSize; index++) {
+ keySerializeRow.setOutputAppend(output);
+ keySerializeRow.serializeWrite(batch, index);
+ if (keySerializeRow.getIsAllNulls()) {
+ if (prevKeyIsNull) {
+ duplicateCounts[seriesCount]++;
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = prevKeyIsNull = true;
+ }
+ output.setWritePosition(currentKeyStart);
+ } else {
+ keyLength = output.getLength() - currentKeyStart;
+ if (!prevKeyIsNull &&
+ StringExpr.equal(
+ output.getData(), prevKeyStart, prevKeyLength,
+ output.getData(), currentKeyStart, keyLength)) {
+ duplicateCounts[seriesCount]++;
+ output.setWritePosition(currentKeyStart);
+ } else {
+ duplicateCounts[++seriesCount] = 1;
+ seriesIsAllNull[seriesCount] = prevKeyIsNull = false;
+ prevKeyStart = currentKeyStart;
+ serializedKeyLengths[nonNullKeyCount] = prevKeyLength = keyLength;
+ currentKeyStart += keyLength;
+ hasAnyNulls[nonNullKeyCount] = keySerializeRow.getHasAnyNulls();
+ nonNullKeyCount++;
+ }
+ }
+ }
+ seriesCount++;
+ Preconditions.checkState(seriesCount <= currentBatchSize);
+ }
+
+ // Finally.
+ computeSerializedHashCodes();
+ positionToFirst();
+ Preconditions.checkState(validate());
+ }
+
+ @Override
+ public void setNextNonNullKey(int nonNullKeyPosition) {
+ super.setNextNonNullKey(nonNullKeyPosition);
+
+ currentHasAnyNulls = hasAnyNulls[nonNullKeyPosition];
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesSerialized.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesSerialized.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesSerialized.java
new file mode 100644
index 0000000..1dfb3df
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesSerialized.java
@@ -0,0 +1,35 @@
+/**
+ * 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.keyseries;
+
+/**
+ * An abstract adding serialization to key series.
+ *
+ * A key with no or some nulls has serialized bytes, offset, and length.
+ */
+public interface VectorKeySeriesSerialized extends VectorKeySeries {
+
+ /**
+ * @return the serialized bytes (including optional key tag), start offset of the key in the
+ * bytes, and key byte length.
+ */
+ byte[] getSerializedBytes();
+ int getSerializedStart();
+ int getSerializedLength();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hive/blob/409db57d/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesSerializedImpl.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesSerializedImpl.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesSerializedImpl.java
new file mode 100644
index 0000000..1fbafa7
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/keyseries/VectorKeySeriesSerializedImpl.java
@@ -0,0 +1,130 @@
+/**
+ * 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.keyseries;
+
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.serde2.ByteStream.Output;
+import org.apache.hadoop.hive.serde2.fast.SerializeWrite;
+import org.apache.hive.common.util.HashCodeUtil;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Implementation of base serialization interface.
+ *
+ */
+public abstract class VectorKeySeriesSerializedImpl<T extends SerializeWrite>
+ extends VectorKeySeriesSingleImpl implements VectorKeySeriesSerialized {
+
+ protected T serializeWrite;
+
+ protected int bufferOffset;
+
+ // The serialized (non-NULL) series keys. These 3 members represent the value.
+ public int serializedStart;
+ public int serializedLength;
+ public byte[] serializedBytes;
+
+ protected final Output output;
+
+ protected final int[] serializedKeyLengths;
+
+ public VectorKeySeriesSerializedImpl(T serializeWrite) {
+ super();
+ this.serializeWrite = serializeWrite;
+ output = new Output();
+ serializedKeyLengths = new int[VectorizedRowBatch.DEFAULT_SIZE];
+ }
+
+ public boolean validate() {
+ super.validate();
+
+ int nullCount = 0;
+ for (int i = 0; i < seriesCount; i++) {
+ if (seriesIsAllNull[i]) {
+ nullCount++;
+ }
+ }
+ Preconditions.checkState(nullCount + nonNullKeyCount == seriesCount);
+
+ int lengthSum = 0;
+ int keyLength;
+ for (int i = 0; i < nonNullKeyCount; i++) {
+ keyLength = serializedKeyLengths[i];
+ Preconditions.checkState(keyLength > 0);
+ lengthSum += keyLength;
+ Preconditions.checkState(lengthSum <= output.getLength());
+ }
+ return true;
+ }
+
+ @Override
+ public byte[] getSerializedBytes() {
+ return serializedBytes;
+ }
+
+ @Override
+ public int getSerializedStart() {
+ return serializedStart;
+ }
+
+ @Override
+ public int getSerializedLength() {
+ return serializedLength;
+ }
+
+ /**
+ * Batch compute the hash codes for all the serialized keys.
+ *
+ * NOTE: MAJOR MAJOR ASSUMPTION:
+ * We assume that HashCodeUtil.murmurHash produces the same result
+ * as MurmurHash.hash with seed = 0 (the method used by ReduceSinkOperator for
+ * UNIFORM distribution).
+ */
+ protected void computeSerializedHashCodes() {
+ int offset = 0;
+ int keyLength;
+ byte[] bytes = output.getData();
+ for (int i = 0; i < nonNullKeyCount; i++) {
+ keyLength = serializedKeyLengths[i];
+ hashCodes[i] = HashCodeUtil.murmurHash(bytes, offset, keyLength);
+ offset += keyLength;
+ }
+ }
+
+ @Override
+ public void positionToFirst() {
+
+ // Reset this before calling positionToFirst.
+ bufferOffset = 0;
+
+ super.positionToFirst();
+
+ // This is constant for whole series.
+ serializedBytes = output.getData();
+ }
+
+ @Override
+ public void setNextNonNullKey(int nonNullKeyPosition) {
+ serializedStart = bufferOffset;
+ serializedLength = serializedKeyLengths[nonNullKeyPosition];
+ Preconditions.checkState(serializedStart + serializedLength <= output.getData().length);
+ bufferOffset += serializedLength;
+ }
+}
\ No newline at end of file