You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by gu...@apache.org on 2013/11/26 09:19:34 UTC
svn commit: r1545564 [4/22] - in /hive/branches/tez: ./ ant/ beeline/ bin/
cli/ common/ common/src/java/org/apache/hadoop/hive/common/
common/src/java/org/apache/hadoop/hive/common/type/
common/src/java/org/apache/hadoop/hive/conf/ common/src/test/org/...
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/VectorExpressionWriterFactory.java Tue Nov 26 08:19:25 2013
@@ -20,8 +20,12 @@ package org.apache.hadoop.hive.ql.exec.v
import java.sql.Timestamp;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.hive.common.type.HiveVarchar;
import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
@@ -35,16 +39,39 @@ import org.apache.hadoop.hive.serde2.io.
import org.apache.hadoop.hive.serde2.io.TimestampWritable;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.SettableStructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableBinaryObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableBooleanObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableByteObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableDateObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableDoubleObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableFloatObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableHiveDecimalObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableHiveVarcharObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableIntObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableLongObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableHiveVarcharObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableTimestampObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableShortObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableStringObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.SettableTimestampObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.ListTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.MapTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+import org.apache.hadoop.hive.serde2.typeinfo.UnionTypeInfo;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
/**
* VectorExpressionWritableFactory helper class for generating VectorExpressionWritable objects.
@@ -67,20 +94,11 @@ public final class VectorExpressionWrite
return objectInspector;
}
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- objectInspector = nodeDesc.getWritableObjectInspector();
- if (null == objectInspector) {
- objectInspector = TypeInfoUtils
- .getStandardWritableObjectInspectorFromTypeInfo(nodeDesc.getTypeInfo());
- }
- if (null == objectInspector) {
- throw new HiveException(String.format(
- "Failed to initialize VectorExpressionWriter for expr: %s",
- nodeDesc.getExprString()));
- }
+ public VectorExpressionWriter init(ObjectInspector objectInspector) throws HiveException {
+ this.objectInspector = objectInspector;
return this;
}
-
+
/**
* The base implementation must be overridden by the Long specialization
*/
@@ -88,6 +106,13 @@ public final class VectorExpressionWrite
public Object writeValue(long value) throws HiveException {
throw new HiveException("Internal error: should not reach here");
}
+
+ /**
+ * The base implementation must be overridden by the Long specialization
+ */
+ public Object setValue(Object field, long value) throws HiveException {
+ throw new HiveException("Internal error: should not reach here");
+ }
/**
* The base implementation must be overridden by the Double specialization
@@ -98,12 +123,26 @@ public final class VectorExpressionWrite
}
/**
+ * The base implementation must be overridden by the Double specialization
+ */
+ public Object setValue(Object field, double value) throws HiveException {
+ throw new HiveException("Internal error: should not reach here");
+ }
+
+ /**
* The base implementation must be overridden by the Bytes specialization
*/
@Override
public Object writeValue(byte[] value, int start, int length) throws HiveException {
throw new HiveException("Internal error: should not reach here");
}
+
+ /**
+ * The base implementation must be overridden by the Bytes specialization
+ */
+ public Object setValue(Object field, byte[] value, int start, int length) throws HiveException {
+ throw new HiveException("Internal error: should not reach here");
+ }
}
/**
@@ -133,7 +172,29 @@ public final class VectorExpressionWrite
"Incorrect null/repeating: row:%d noNulls:%b isRepeating:%b isNull[row]:%b isNull[0]:%b",
row, lcv.noNulls, lcv.isRepeating, lcv.isNull[row], lcv.isNull[0]));
}
- }
+
+ @Override
+ public Object setValue(Object field, ColumnVector column, int row) throws HiveException {
+ LongColumnVector lcv = (LongColumnVector) column;
+ if (lcv.noNulls && !lcv.isRepeating) {
+ return setValue(field, lcv.vector[row]);
+ } else if (lcv.noNulls && lcv.isRepeating) {
+ return setValue(field, lcv.vector[0]);
+ } else if (!lcv.noNulls && !lcv.isRepeating && !lcv.isNull[row]) {
+ return setValue(field, lcv.vector[row]);
+ } else if (!lcv.noNulls && !lcv.isRepeating && lcv.isNull[row]) {
+ return null;
+ } else if (!lcv.noNulls && lcv.isRepeating && !lcv.isNull[0]) {
+ return setValue(field, lcv.vector[0]);
+ } else if (!lcv.noNulls && lcv.isRepeating && lcv.isNull[0]) {
+ return null;
+ }
+ throw new HiveException(
+ String.format(
+ "Incorrect null/repeating: row:%d noNulls:%b isRepeating:%b isNull[row]:%b isNull[0]:%b",
+ row, lcv.noNulls, lcv.isRepeating, lcv.isNull[row], lcv.isNull[0]));
+ }
+ }
/**
* Specialized writer for DoubleColumnVector. Will throw cast exception
@@ -161,6 +222,28 @@ public final class VectorExpressionWrite
"Incorrect null/repeating: row:%d noNulls:%b isRepeating:%b isNull[row]:%b isNull[0]:%b",
row, dcv.noNulls, dcv.isRepeating, dcv.isNull[row], dcv.isNull[0]));
}
+
+ @Override
+ public Object setValue(Object field, ColumnVector column, int row) throws HiveException {
+ DoubleColumnVector dcv = (DoubleColumnVector) column;
+ if (dcv.noNulls && !dcv.isRepeating) {
+ return setValue(field, dcv.vector[row]);
+ } else if (dcv.noNulls && dcv.isRepeating) {
+ return setValue(field, dcv.vector[0]);
+ } else if (!dcv.noNulls && !dcv.isRepeating && !dcv.isNull[row]) {
+ return setValue(field, dcv.vector[row]);
+ } else if (!dcv.noNulls && !dcv.isRepeating && dcv.isNull[row]) {
+ return null;
+ } else if (!dcv.noNulls && dcv.isRepeating && !dcv.isNull[0]) {
+ return setValue(field, dcv.vector[0]);
+ } else if (!dcv.noNulls && dcv.isRepeating && dcv.isNull[0]) {
+ return null;
+ }
+ throw new HiveException(
+ String.format(
+ "Incorrect null/repeating: row:%d noNulls:%b isRepeating:%b isNull[row]:%b isNull[0]:%b",
+ row, dcv.noNulls, dcv.isRepeating, dcv.isNull[row], dcv.isNull[0]));
+ }
}
/**
@@ -189,176 +272,528 @@ public final class VectorExpressionWrite
"Incorrect null/repeating: row:%d noNulls:%b isRepeating:%b isNull[row]:%b isNull[0]:%b",
row, bcv.noNulls, bcv.isRepeating, bcv.isNull[row], bcv.isNull[0]));
}
+
+ @Override
+ public Object setValue(Object field, ColumnVector column, int row) throws HiveException {
+ BytesColumnVector bcv = (BytesColumnVector) column;
+ if (bcv.noNulls && !bcv.isRepeating) {
+ return setValue(field, bcv.vector[row], bcv.start[row], bcv.length[row]);
+ } else if (bcv.noNulls && bcv.isRepeating) {
+ return setValue(field, bcv.vector[0], bcv.start[0], bcv.length[0]);
+ } else if (!bcv.noNulls && !bcv.isRepeating && !bcv.isNull[row]) {
+ return setValue(field, bcv.vector[row], bcv.start[row], bcv.length[row]);
+ } else if (!bcv.noNulls && !bcv.isRepeating && bcv.isNull[row]) {
+ return null;
+ } else if (!bcv.noNulls && bcv.isRepeating && !bcv.isNull[0]) {
+ return setValue(field, bcv.vector[0], bcv.start[0], bcv.length[0]);
+ } else if (!bcv.noNulls && bcv.isRepeating && bcv.isNull[0]) {
+ return null;
+ }
+ throw new HiveException(
+ String.format(
+ "Incorrect null/repeating: row:%d noNulls:%b isRepeating:%b isNull[row]:%b isNull[0]:%b",
+ row, bcv.noNulls, bcv.isRepeating, bcv.isNull[row], bcv.isNull[0]));
+ }
}
- /**
- * Compiles the appropriate vector expression writer based on an expression info (ExprNodeDesc)
- */
- public static VectorExpressionWriter genVectorExpressionWritable(ExprNodeDesc nodeDesc)
- throws HiveException {
- String nodeType = nodeDesc.getTypeString();
- if (nodeType.equalsIgnoreCase("tinyint")) {
- return new VectorExpressionWriterLong()
- {
- private ByteWritable writable;
-
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new ByteWritable();
- return this;
- }
-
- @Override
- public Object writeValue(long value) {
- writable.set((byte) value);
- return writable;
- }
- }.init(nodeDesc);
- } else if (nodeType.equalsIgnoreCase("smallint")) {
- return new VectorExpressionWriterLong()
- {
- private ShortWritable writable;
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new ShortWritable();
- return this;
- }
-
- @Override
- public Object writeValue(long value) {
- writable.set((short) value);
- return writable;
- }
- }.init(nodeDesc);
- } else if (nodeType.equalsIgnoreCase("int")) {
- return new VectorExpressionWriterLong()
- {
- private IntWritable writable;
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new IntWritable();
- return this;
- }
-
- @Override
- public Object writeValue(long value) {
- writable.set((int) value);
- return writable;
- }
- }.init(nodeDesc);
- } else if (nodeType.equalsIgnoreCase("bigint")) {
- return new VectorExpressionWriterLong()
- {
- private LongWritable writable;
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new LongWritable();
- return this;
- }
-
- @Override
- public Object writeValue(long value) {
- writable.set(value);
- return writable;
- }
- }.init(nodeDesc);
- } else if (nodeType.equalsIgnoreCase("boolean")) {
- return new VectorExpressionWriterLong()
- {
- private BooleanWritable writable;
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new BooleanWritable();
- return this;
- }
-
- @Override
- public Object writeValue(long value) {
- writable.set(value != 0 ? true : false);
- return writable;
- }
- }.init(nodeDesc);
- } else if (nodeType.equalsIgnoreCase("timestamp")) {
- return new VectorExpressionWriterLong()
- {
- private TimestampWritable writable;
- private Timestamp timestamp;
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new TimestampWritable();
- timestamp = new Timestamp(0);
- return this;
- }
-
- @Override
- public Object writeValue(long value) {
- TimestampUtils.assignTimeInNanoSec(value, timestamp);
- writable.set(timestamp);
- return writable;
- }
- }.init(nodeDesc);
- } else if (nodeType.equalsIgnoreCase("string")) {
- return new VectorExpressionWriterBytes()
- {
- private Text writable;
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new Text();
- return this;
- }
-
- @Override
- public Object writeValue(byte[] value, int start, int length) throws HiveException {
- writable.set(value, start, length);
- return writable;
- }
- }.init(nodeDesc);
- } else if (nodeType.equalsIgnoreCase("float")) {
- return new VectorExpressionWriterDouble()
- {
- private FloatWritable writable;
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new FloatWritable();
- return this;
- }
-
- @Override
- public Object writeValue(double value) {
- writable.set((float)value);
- return writable;
- }
- }.init(nodeDesc);
- } else if (nodeType.equalsIgnoreCase("double")) {
- return new VectorExpressionWriterDouble()
- {
- private DoubleWritable writable;
- @Override
- public VectorExpressionWriter init(ExprNodeDesc nodeDesc) throws HiveException {
- super.init(nodeDesc);
- writable = new DoubleWritable();
- return this;
- }
-
- @Override
- public Object writeValue(double value) {
- writable.set(value);
- return writable;
- }
- }.init(nodeDesc);
- }
-
- throw new HiveException(String.format(
- "Unimplemented genVectorExpressionWritable type: %s for expression: %s",
- nodeType, nodeDesc));
+ /**
+ * Compiles the appropriate vector expression writer based on an expression info (ExprNodeDesc)
+ */
+ public static VectorExpressionWriter genVectorExpressionWritable(ExprNodeDesc nodeDesc)
+ throws HiveException {
+ String nodeType = nodeDesc.getTypeString();
+ ObjectInspector objectInspector = nodeDesc.getWritableObjectInspector();
+ if (null == objectInspector) {
+ objectInspector = TypeInfoUtils
+ .getStandardWritableObjectInspectorFromTypeInfo(nodeDesc.getTypeInfo());
+ }
+ if (null == objectInspector) {
+ throw new HiveException(String.format(
+ "Failed to initialize VectorExpressionWriter for expr: %s",
+ nodeDesc.getExprString()));
+ }
+ return genVectorExpressionWritable(objectInspector);
+ }
+
+ /**
+ * Compiles the appropriate vector expression writer based on an expression info (ExprNodeDesc)
+ */
+ public static VectorExpressionWriter genVectorExpressionWritable(
+ ObjectInspector fieldObjInspector) throws HiveException {
+
+ switch (fieldObjInspector.getCategory()) {
+ case PRIMITIVE:
+ switch (((PrimitiveObjectInspector) fieldObjInspector).getPrimitiveCategory()) {
+ case FLOAT:
+ return genVectorExpressionWritableFloat(
+ (SettableFloatObjectInspector) fieldObjInspector);
+ case DOUBLE:
+ return genVectorExpressionWritableDouble(
+ (SettableDoubleObjectInspector) fieldObjInspector);
+ case BOOLEAN:
+ return genVectorExpressionWritableBoolean(
+ (SettableBooleanObjectInspector) fieldObjInspector);
+ case BYTE:
+ return genVectorExpressionWritableByte(
+ (SettableByteObjectInspector) fieldObjInspector);
+ case SHORT:
+ return genVectorExpressionWritableShort(
+ (SettableShortObjectInspector) fieldObjInspector);
+ case INT:
+ return genVectorExpressionWritableInt(
+ (SettableIntObjectInspector) fieldObjInspector);
+ case LONG:
+ return genVectorExpressionWritableLong(
+ (SettableLongObjectInspector) fieldObjInspector);
+ case BINARY:
+ return genVectorExpressionWritableBinary(
+ (SettableBinaryObjectInspector) fieldObjInspector);
+ case STRING:
+ return genVectorExpressionWritableString(
+ (SettableStringObjectInspector) fieldObjInspector);
+ case VARCHAR:
+ return genVectorExpressionWritableVarchar(
+ (SettableHiveVarcharObjectInspector) fieldObjInspector);
+ case TIMESTAMP:
+ return genVectorExpressionWritableTimestamp(
+ (SettableTimestampObjectInspector) fieldObjInspector);
+ case DATE:
+ return genVectorExpressionWritableDate(
+ (SettableDateObjectInspector) fieldObjInspector);
+ case DECIMAL:
+ return genVectorExpressionWritableDecimal(
+ (SettableHiveDecimalObjectInspector) fieldObjInspector);
+ default:
+ throw new IllegalArgumentException("Unknown primitive type: " +
+ ((PrimitiveObjectInspector) fieldObjInspector).getPrimitiveCategory());
+ }
+
+ case STRUCT:
+ case UNION:
+ case MAP:
+ case LIST:
+ throw new IllegalArgumentException("Unsupported complex type: " +
+ fieldObjInspector.getCategory());
+ default:
+ throw new IllegalArgumentException("Unknown type " +
+ fieldObjInspector.getCategory());
+ }
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableDecimal(
+ SettableHiveDecimalObjectInspector fieldObjInspector) throws HiveException {
+
+ // We should never reach this, the compile validation should guard us
+ throw new HiveException("DECIMAL primitive type not supported in vectorization.");
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableDate(
+ SettableDateObjectInspector fieldObjInspector) throws HiveException {
+ // We should never reach this, the compile validation should guard us
+ throw new HiveException("DATE primitive type not supported in vectorization.");
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableTimestamp(
+ SettableTimestampObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterLong() {
+ private Object obj;
+ private Timestamp ts;
+
+ public VectorExpressionWriter init(SettableTimestampObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ ts = new Timestamp(0);
+ obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(long value) {
+ TimestampUtils.assignTimeInNanoSec(value, ts);
+ ((SettableTimestampObjectInspector) this.objectInspector).set(obj, ts);
+ return obj;
+ }
+
+ @Override
+ public Object setValue(Object field, long value) {
+ if (null == field) {
+ field = initValue(null);
+ }
+ TimestampUtils.assignTimeInNanoSec(value, ts);
+ ((SettableTimestampObjectInspector) this.objectInspector).set(field, ts);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableTimestampObjectInspector) this.objectInspector).create(new Timestamp(0));
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableVarchar(
+ SettableHiveVarcharObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterBytes() {
+ private Object obj;
+ private Text text;
+
+ public VectorExpressionWriter init(SettableHiveVarcharObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.text = new Text();
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(byte[] value, int start, int length) throws HiveException {
+ text.set(value, start, length);
+ ((SettableHiveVarcharObjectInspector) this.objectInspector).set(this.obj, text.toString());
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, byte[] value, int start, int length)
+ throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ text.set(value, start, length);
+ ((SettableHiveVarcharObjectInspector) this.objectInspector).set(field, text.toString());
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableHiveVarcharObjectInspector) this.objectInspector)
+ .create(new HiveVarchar(StringUtils.EMPTY, -1));
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableString(
+ SettableStringObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterBytes() {
+ private Object obj;
+ private Text text;
+
+ public VectorExpressionWriter init(SettableStringObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.text = new Text();
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(byte[] value, int start, int length) throws HiveException {
+ this.text.set(value, start, length);
+ ((SettableStringObjectInspector) this.objectInspector).set(this.obj, this.text.toString());
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, byte[] value, int start, int length)
+ throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ this.text.set(value, start, length);
+ ((SettableStringObjectInspector) this.objectInspector).set(field, this.text.toString());
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableStringObjectInspector) this.objectInspector).create(StringUtils.EMPTY);
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableBinary(
+ SettableBinaryObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterBytes() {
+ private Object obj;
+ private byte[] bytes;
+
+ public VectorExpressionWriter init(SettableBinaryObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.bytes = ArrayUtils.EMPTY_BYTE_ARRAY;
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(byte[] value, int start, int length) throws HiveException {
+ bytes = Arrays.copyOfRange(value, start, start + length);
+ ((SettableBinaryObjectInspector) this.objectInspector).set(this.obj, bytes);
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, byte[] value, int start, int length) throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ bytes = Arrays.copyOfRange(value, start, start + length);
+ ((SettableBinaryObjectInspector) this.objectInspector).set(field, bytes);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableBinaryObjectInspector) this.objectInspector)
+ .create(ArrayUtils.EMPTY_BYTE_ARRAY);
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableLong(
+ SettableLongObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterLong() {
+ private Object obj;
+
+ public VectorExpressionWriter init(SettableLongObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(long value) throws HiveException {
+ ((SettableLongObjectInspector) this.objectInspector).set(this.obj, value);
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, long value) throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ ((SettableLongObjectInspector) this.objectInspector).set(field, value);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableLongObjectInspector) this.objectInspector)
+ .create(0L);
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableInt(
+ SettableIntObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterLong() {
+ private Object obj;
+
+ public VectorExpressionWriter init(SettableIntObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(long value) throws HiveException {
+ ((SettableIntObjectInspector) this.objectInspector).set(this.obj, (int) value);
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, long value) throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ ((SettableIntObjectInspector) this.objectInspector).set(field, (int) value);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableIntObjectInspector) this.objectInspector)
+ .create(0);
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableShort(
+ SettableShortObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterLong() {
+ private Object obj;
+
+ public VectorExpressionWriter init(SettableShortObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(long value) throws HiveException {
+ ((SettableShortObjectInspector) this.objectInspector).set(this.obj, (short) value);
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, long value) throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ ((SettableShortObjectInspector) this.objectInspector).set(field, (short) value);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableShortObjectInspector) this.objectInspector)
+ .create((short) 0);
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableByte(
+ SettableByteObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterLong() {
+ private Object obj;
+
+ public VectorExpressionWriter init(SettableByteObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(long value) throws HiveException {
+ ((SettableByteObjectInspector) this.objectInspector).set(this.obj, (byte) value);
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, long value) throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ ((SettableByteObjectInspector) this.objectInspector).set(field, (byte) value);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableByteObjectInspector) this.objectInspector)
+ .create((byte) 0);
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableBoolean(
+ SettableBooleanObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterLong() {
+ private Object obj;
+
+ public VectorExpressionWriter init(SettableBooleanObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(long value) throws HiveException {
+ ((SettableBooleanObjectInspector) this.objectInspector).set(this.obj,
+ value == 0 ? false : true);
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, long value) throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ ((SettableBooleanObjectInspector) this.objectInspector).set(field,
+ value == 0 ? false : true);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableBooleanObjectInspector) this.objectInspector)
+ .create(false);
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableDouble(
+ SettableDoubleObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterDouble() {
+ private Object obj;
+
+ public VectorExpressionWriter init(SettableDoubleObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(double value) throws HiveException {
+ ((SettableDoubleObjectInspector) this.objectInspector).set(this.obj, value);
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, double value) throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ ((SettableDoubleObjectInspector) this.objectInspector).set(field, value);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableDoubleObjectInspector) this.objectInspector)
+ .create(0f);
+ }
+ }.init(fieldObjInspector);
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritableFloat(
+ SettableFloatObjectInspector fieldObjInspector) throws HiveException {
+ return new VectorExpressionWriterDouble() {
+ private Object obj;
+
+ public VectorExpressionWriter init(SettableFloatObjectInspector objInspector)
+ throws HiveException {
+ super.init(objInspector);
+ this.obj = initValue(null);
+ return this;
+ }
+
+ @Override
+ public Object writeValue(double value) throws HiveException {
+ ((SettableFloatObjectInspector) this.objectInspector).set(this.obj, (float) value);
+ return this.obj;
+ }
+
+ @Override
+ public Object setValue(Object field, double value) throws HiveException {
+ if (null == field) {
+ field = initValue(null);
+ }
+ ((SettableFloatObjectInspector) this.objectInspector).set(field, (float) value);
+ return field;
+ }
+
+ @Override
+ public Object initValue(Object ignored) {
+ return ((SettableFloatObjectInspector) this.objectInspector)
+ .create(0f);
+ }
+ }.init(fieldObjInspector);
}
/**
@@ -432,25 +867,87 @@ public final class VectorExpressionWrite
*/
public static VectorExpressionWriter[] getExpressionWriters(StructObjectInspector objInspector)
throws HiveException {
- List<ExprNodeDesc> outputNodeDescs = new ArrayList<ExprNodeDesc>();
- for (StructField fieldRef : (objInspector)
- .getAllStructFieldRefs()) {
- String typeName = fieldRef.getFieldObjectInspector().getTypeName();
- TypeInfo ti = TypeInfoFactory.getPrimitiveTypeInfo(typeName);
- outputNodeDescs.add(new ExprNodeDesc(ti) {
- private static final long serialVersionUID = 1L;
-
- @Override
- public ExprNodeDesc clone() { /* Not needed */
- return null;
- }
-
- @Override
- public boolean isSame(Object o) { /* Not needed */
- return false;
- }
- });
+
+ if (objInspector.isSettable()) {
+ return getSettableExpressionWriters((SettableStructObjectInspector) objInspector);
+ }
+
+ List<? extends StructField> allFieldRefs = objInspector.getAllStructFieldRefs();
+
+ VectorExpressionWriter[] expressionWriters = new VectorExpressionWriter[allFieldRefs.size()];
+
+ for(int i=0; i<expressionWriters.length; ++i) {
+ expressionWriters[i] = genVectorExpressionWritable(allFieldRefs.get(i).getFieldObjectInspector());
+ }
+
+ return expressionWriters;
+ }
+
+ public static VectorExpressionWriter[] getSettableExpressionWriters(
+ SettableStructObjectInspector objInspector) throws HiveException {
+ List<? extends StructField> fieldsRef = objInspector.getAllStructFieldRefs();
+ VectorExpressionWriter[] writers = new VectorExpressionWriter[fieldsRef.size()];
+ for(int i=0; i<writers.length; ++i) {
+ StructField fieldRef = fieldsRef.get(i);
+ VectorExpressionWriter baseWriter = genVectorExpressionWritable(
+ fieldRef.getFieldObjectInspector());
+ writers[i] = genVectorExpressionWritable(objInspector, fieldRef, baseWriter);
+ }
+ return writers;
+
+ }
+
+ /**
+ * VectorExpressionWriterSetter helper for vector expression writers that use
+ * settable ObjectInspector fields to assign the values.
+ * This is used by the OrcStruct serialization (eg. CREATE TABLE ... AS ...)
+ */
+ private static class VectorExpressionWriterSetter extends VectorExpressionWriterBase {
+ private SettableStructObjectInspector settableObjInspector;
+ private StructField fieldRef;
+ private VectorExpressionWriter baseWriter;
+
+ public VectorExpressionWriterSetter init(
+ SettableStructObjectInspector objInspector,
+ StructField fieldRef,
+ VectorExpressionWriter baseWriter) {
+ this.fieldRef = fieldRef;
+ this.settableObjInspector = objInspector;
+ this.objectInspector = fieldRef.getFieldObjectInspector();
+ this.baseWriter = baseWriter;
+ return this;
+ }
+
+ @Override
+ public Object writeValue(ColumnVector column, int row)
+ throws HiveException {
+ throw new HiveException("Should never reach here");
}
- return VectorExpressionWriterFactory.getExpressionWriters(outputNodeDescs);
+
+ @Override
+ public Object setValue(Object row, ColumnVector column, int columnRow)
+ throws HiveException {
+
+ // NULLs are handled by each individual base writer setter
+ // We could handle NULLs centrally here but that would result in spurious allocs
+
+ Object fieldValue = this.settableObjInspector.getStructFieldData(row, fieldRef);
+ fieldValue = baseWriter.setValue(fieldValue, column, columnRow);
+ return this.settableObjInspector.setStructFieldData(row, fieldRef, fieldValue);
+ }
+
+ @Override
+ public Object initValue(Object struct) throws HiveException {
+ Object initValue = this.baseWriter.initValue(null);
+ this.settableObjInspector.setStructFieldData(struct, fieldRef, initValue);
+ return struct;
+ }
+ }
+
+ private static VectorExpressionWriter genVectorExpressionWritable(
+ SettableStructObjectInspector objInspector,
+ StructField fieldRef,
+ VectorExpressionWriter baseWriter) throws HiveException {
+ return new VectorExpressionWriterSetter().init(objInspector, fieldRef, baseWriter);
}
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java Tue Nov 26 08:19:25 2013
@@ -29,10 +29,8 @@ abstract class InStream extends InputStr
private final long[] offsets;
private final long length;
private long currentOffset;
- private byte[] range;
+ private ByteBuffer range;
private int currentRange;
- private int offsetInRange;
- private int limitInRange;
public UncompressedStream(String name, ByteBuffer[] input, long[] offsets,
long length) {
@@ -41,42 +39,39 @@ abstract class InStream extends InputStr
this.offsets = offsets;
this.length = length;
currentRange = 0;
- offsetInRange = 0;
- limitInRange = 0;
currentOffset = 0;
}
@Override
public int read() {
- if (offsetInRange >= limitInRange) {
+ if (range == null || range.remaining() == 0) {
if (currentOffset == length) {
return -1;
}
seek(currentOffset);
}
currentOffset += 1;
- return 0xff & range[offsetInRange++];
+ return 0xff & range.get();
}
@Override
public int read(byte[] data, int offset, int length) {
- if (offsetInRange >= limitInRange) {
+ if (range == null || range.remaining() == 0) {
if (currentOffset == this.length) {
return -1;
}
seek(currentOffset);
}
- int actualLength = Math.min(length, limitInRange - offsetInRange);
- System.arraycopy(range, offsetInRange, data, offset, actualLength);
- offsetInRange += actualLength;
+ int actualLength = Math.min(length, range.remaining());
+ range.get(data, offset, actualLength);
currentOffset += actualLength;
return actualLength;
}
@Override
public int available() {
- if (offsetInRange < limitInRange) {
- return limitInRange - offsetInRange;
+ if (range != null && range.remaining() > 0) {
+ return range.remaining();
}
return (int) (length - currentOffset);
}
@@ -85,6 +80,10 @@ abstract class InStream extends InputStr
public void close() {
currentRange = bytes.length;
currentOffset = length;
+ // explicit de-ref of bytes[]
+ for(int i = 0; i < bytes.length; i++) {
+ bytes[i] = null;
+ }
}
@Override
@@ -98,10 +97,10 @@ abstract class InStream extends InputStr
desired - offsets[i] < bytes[i].remaining()) {
currentOffset = desired;
currentRange = i;
- this.range = bytes[i].array();
- offsetInRange = bytes[i].arrayOffset() + bytes[i].position();
- limitInRange = bytes[i].arrayOffset() + bytes[i].limit();
- offsetInRange += desired - offsets[i];
+ this.range = bytes[i].duplicate();
+ int pos = range.position();
+ pos += (int)(desired - offsets[i]); // this is why we duplicate
+ this.range.position(pos);
return;
}
}
@@ -113,7 +112,7 @@ abstract class InStream extends InputStr
public String toString() {
return "uncompressed stream " + name + " position: " + currentOffset +
" length: " + length + " range: " + currentRange +
- " offset: " + offsetInRange + " limit: " + limitInRange;
+ " offset: " + (range == null ? 0 : range.position()) + " limit: " + (range == null ? 0 : range.limit());
}
}
@@ -123,14 +122,13 @@ abstract class InStream extends InputStr
private final long[] offsets;
private final int bufferSize;
private final long length;
- private ByteBuffer uncompressed = null;
+ private ByteBuffer uncompressed;
private final CompressionCodec codec;
- private byte[] compressed = null;
+ private ByteBuffer compressed;
private long currentOffset;
private int currentRange;
- private int offsetInCompressed;
- private int limitInCompressed;
private boolean isUncompressedOriginal;
+ private boolean isDirect = false;
public CompressedStream(String name, ByteBuffer[] input,
long[] offsets, long length,
@@ -140,49 +138,59 @@ abstract class InStream extends InputStr
this.name = name;
this.codec = codec;
this.length = length;
+ if(this.length > 0) {
+ isDirect = this.bytes[0].isDirect();
+ }
this.offsets = offsets;
this.bufferSize = bufferSize;
currentOffset = 0;
currentRange = 0;
- offsetInCompressed = 0;
- limitInCompressed = 0;
+ }
+
+ private ByteBuffer allocateBuffer(int size) {
+ // TODO: use the same pool as the ORC readers
+ if(isDirect == true) {
+ return ByteBuffer.allocateDirect(size);
+ } else {
+ return ByteBuffer.allocate(size);
+ }
}
private void readHeader() throws IOException {
- if (compressed == null || offsetInCompressed >= limitInCompressed) {
+ if (compressed == null || compressed.remaining() <= 0) {
seek(currentOffset);
}
- if (limitInCompressed - offsetInCompressed > OutStream.HEADER_SIZE) {
- int chunkLength = ((0xff & compressed[offsetInCompressed + 2]) << 15) |
- ((0xff & compressed[offsetInCompressed + 1]) << 7) |
- ((0xff & compressed[offsetInCompressed]) >> 1);
+ if (compressed.remaining() > OutStream.HEADER_SIZE) {
+ int b0 = compressed.get() & 0xff;
+ int b1 = compressed.get() & 0xff;
+ int b2 = compressed.get() & 0xff;
+ boolean isOriginal = (b0 & 0x01) == 1;
+ int chunkLength = (b2 << 15) | (b1 << 7) | (b0 >> 1);
+
if (chunkLength > bufferSize) {
throw new IllegalArgumentException("Buffer size too small. size = " +
bufferSize + " needed = " + chunkLength);
}
- boolean isOriginal = (compressed[offsetInCompressed] & 0x01) == 1;
- offsetInCompressed += OutStream.HEADER_SIZE;
+ // read 3 bytes, which should be equal to OutStream.HEADER_SIZE always
+ assert OutStream.HEADER_SIZE == 3 : "The Orc HEADER_SIZE must be the same in OutStream and InStream";
+ currentOffset += OutStream.HEADER_SIZE;
+
+ ByteBuffer slice = this.slice(chunkLength);
+
if (isOriginal) {
+ uncompressed = slice;
isUncompressedOriginal = true;
- uncompressed = bytes[currentRange].duplicate();
- uncompressed.position(offsetInCompressed -
- bytes[currentRange].arrayOffset());
- uncompressed.limit(offsetInCompressed + chunkLength);
} else {
if (isUncompressedOriginal) {
- uncompressed = ByteBuffer.allocate(bufferSize);
+ uncompressed = allocateBuffer(bufferSize);
isUncompressedOriginal = false;
} else if (uncompressed == null) {
- uncompressed = ByteBuffer.allocate(bufferSize);
+ uncompressed = allocateBuffer(bufferSize);
} else {
uncompressed.clear();
}
- codec.decompress(ByteBuffer.wrap(compressed, offsetInCompressed,
- chunkLength),
- uncompressed);
+ codec.decompress(slice, uncompressed);
}
- offsetInCompressed += chunkLength;
- currentOffset += chunkLength + OutStream.HEADER_SIZE;
} else {
throw new IllegalStateException("Can't read header at " + this);
}
@@ -208,10 +216,7 @@ abstract class InStream extends InputStr
readHeader();
}
int actualLength = Math.min(length, uncompressed.remaining());
- System.arraycopy(uncompressed.array(),
- uncompressed.arrayOffset() + uncompressed.position(), data,
- offset, actualLength);
- uncompressed.position(uncompressed.position() + actualLength);
+ uncompressed.get(data, offset, actualLength);
return actualLength;
}
@@ -229,10 +234,12 @@ abstract class InStream extends InputStr
@Override
public void close() {
uncompressed = null;
+ compressed = null;
currentRange = bytes.length;
- offsetInCompressed = 0;
- limitInCompressed = 0;
currentOffset = length;
+ for(int i = 0; i < bytes.length; i++) {
+ bytes[i] = null;
+ }
}
@Override
@@ -249,16 +256,62 @@ abstract class InStream extends InputStr
}
}
+ /* slices a read only contigous buffer of chunkLength */
+ private ByteBuffer slice(int chunkLength) throws IOException {
+ int len = chunkLength;
+ final long oldOffset = currentOffset;
+ ByteBuffer slice;
+ if (compressed.remaining() >= len) {
+ slice = compressed.slice();
+ // simple case
+ slice.limit(len);
+ currentOffset += len;
+ compressed.position(compressed.position() + len);
+ return slice;
+ } else if (currentRange >= (bytes.length - 1)) {
+ // nothing has been modified yet
+ throw new IOException("EOF in " + this + " while trying to read " +
+ chunkLength + " bytes");
+ }
+
+ // we need to consolidate 2 or more buffers into 1
+ // first clear out compressed buffers
+ ByteBuffer copy = allocateBuffer(chunkLength);
+ currentOffset += compressed.remaining();
+ len -= compressed.remaining();
+ copy.put(compressed);
+
+ while (len > 0 && (++currentRange) < bytes.length) {
+ compressed = bytes[currentRange].duplicate();
+ if (compressed.remaining() >= len) {
+ slice = compressed.slice();
+ slice.limit(len);
+ copy.put(slice);
+ currentOffset += len;
+ compressed.position(compressed.position() + len);
+ return copy;
+ }
+ currentOffset += compressed.remaining();
+ len -= compressed.remaining();
+ copy.put(compressed);
+ }
+
+ // restore offsets for exception clarity
+ seek(oldOffset);
+ throw new IOException("EOF in " + this + " while trying to read " +
+ chunkLength + " bytes");
+ }
+
private void seek(long desired) throws IOException {
for(int i = 0; i < bytes.length; ++i) {
if (offsets[i] <= desired &&
desired - offsets[i] < bytes[i].remaining()) {
currentRange = i;
- compressed = bytes[i].array();
- offsetInCompressed = (int) (bytes[i].arrayOffset() +
- bytes[i].position() + (desired - offsets[i]));
+ compressed = bytes[i].duplicate();
+ int pos = compressed.position();
+ pos += (int)(desired - offsets[i]);
+ compressed.position(pos);
currentOffset = desired;
- limitInCompressed = bytes[i].arrayOffset() + bytes[i].limit();
return;
}
}
@@ -267,11 +320,9 @@ abstract class InStream extends InputStr
if (segments != 0 &&
desired == offsets[segments - 1] + bytes[segments - 1].remaining()) {
currentRange = segments - 1;
- compressed = bytes[currentRange].array();
- offsetInCompressed = bytes[currentRange].arrayOffset() +
- bytes[currentRange].limit();
+ compressed = bytes[currentRange].duplicate();
+ compressed.position(compressed.limit());
currentOffset = desired;
- limitInCompressed = offsetInCompressed;
return;
}
throw new IOException("Seek outside of data in " + this + " to " +
@@ -294,7 +345,7 @@ abstract class InStream extends InputStr
public String toString() {
return "compressed stream " + name + " position: " + currentOffset +
" length: " + length + " range: " + currentRange +
- " offset: " + offsetInCompressed + " limit: " + limitInCompressed +
+ " offset: " + (compressed == null ? 0 : compressed.position()) + " limit: " + (compressed == null ? 0 : compressed.limit()) +
rangeString() +
(uncompressed == null ? "" :
" uncompressed: " + uncompressed.position() + " to " +
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSerde.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSerde.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSerde.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSerde.java Tue Nov 26 08:19:25 2013
@@ -143,9 +143,9 @@ public class OrcSerde implements SerDe,
public Writable serializeVector(VectorizedRowBatch vrg, ObjectInspector objInspector)
throws SerDeException {
if (vos == null) {
- vos = new VectorizedOrcSerde(objInspector);
+ vos = new VectorizedOrcSerde(getObjectInspector());
}
- return vos.serialize(vrg, objInspector);
+ return vos.serialize(vrg, getObjectInspector());
}
@Override
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcStruct.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcStruct.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcStruct.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcStruct.java Tue Nov 26 08:19:25 2013
@@ -545,8 +545,8 @@ final class OrcStruct implements Writabl
case DATE:
return PrimitiveObjectInspectorFactory.javaDateObjectInspector;
case DECIMAL:
- int precision = type.hasPrecision() ? type.getPrecision() : HiveDecimal.MAX_PRECISION;
- int scale = type.hasScale()? type.getScale() : HiveDecimal.MAX_SCALE;
+ int precision = type.hasPrecision() ? type.getPrecision() : HiveDecimal.SYSTEM_DEFAULT_PRECISION;
+ int scale = type.hasScale()? type.getScale() : HiveDecimal.SYSTEM_DEFAULT_SCALE;
return PrimitiveObjectInspectorFactory.getPrimitiveJavaObjectInspector(
TypeInfoFactory.getDecimalTypeInfo(precision, scale));
case STRUCT:
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java Tue Nov 26 08:19:25 2013
@@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
import java.sql.Date;
import java.sql.Timestamp;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -75,6 +76,7 @@ class RecordReaderImpl implements Record
private long rowCountInStripe = 0;
private final Map<StreamName, InStream> streams =
new HashMap<StreamName, InStream>();
+ List<BufferChunk> bufferChunks = new ArrayList<BufferChunk>(0);
private final TreeReader reader;
private final OrcProto.RowIndex[] indexes;
private final SearchArgument sarg;
@@ -125,6 +127,7 @@ class RecordReaderImpl implements Record
rows += stripe.getNumberOfRows();
}
}
+
firstRow = skippedRows;
totalRowCount = rows;
reader = createTreeReader(path, 0, types, included);
@@ -1899,8 +1902,8 @@ class RecordReaderImpl implements Record
case DATE:
return new DateTreeReader(path, columnId);
case DECIMAL:
- int precision = type.hasPrecision() ? type.getPrecision() : HiveDecimal.MAX_PRECISION;
- int scale = type.hasScale()? type.getScale() : HiveDecimal.MAX_SCALE;
+ int precision = type.hasPrecision() ? type.getPrecision() : HiveDecimal.SYSTEM_DEFAULT_PRECISION;
+ int scale = type.hasScale()? type.getScale() : HiveDecimal.SYSTEM_DEFAULT_SCALE;
return new DecimalTreeReader(path, columnId, precision, scale);
case STRUCT:
return new StructTreeReader(path, columnId, types, included);
@@ -2176,6 +2179,17 @@ class RecordReaderImpl implements Record
return null;
}
+ private void clearStreams() throws IOException {
+ // explicit close of all streams to de-ref ByteBuffers
+ for(InStream is: streams.values()) {
+ is.close();
+ }
+ if(bufferChunks != null) {
+ bufferChunks.clear();
+ }
+ streams.clear();
+ }
+
/**
* Read the current stripe into memory.
* @throws IOException
@@ -2183,7 +2197,7 @@ class RecordReaderImpl implements Record
private void readStripe() throws IOException {
StripeInformation stripe = stripes.get(currentStripe);
stripeFooter = readStripeFooter(stripe);
- streams.clear();
+ clearStreams();
// setup the position in the stripe
rowCountInStripe = stripe.getNumberOfRows();
rowInStripe = 0;
@@ -2223,28 +2237,17 @@ class RecordReaderImpl implements Record
private void readAllDataStreams(StripeInformation stripe
) throws IOException {
- byte[] buffer =
- new byte[(int) (stripe.getDataLength())];
- file.seek(stripe.getOffset() + stripe.getIndexLength());
- file.readFully(buffer, 0, buffer.length);
- int sectionOffset = 0;
- for(OrcProto.Stream section: stripeFooter.getStreamsList()) {
- if (StreamName.getArea(section.getKind()) == StreamName.Area.DATA) {
- int sectionLength = (int) section.getLength();
- ByteBuffer sectionBuffer = ByteBuffer.wrap(buffer, sectionOffset,
- sectionLength);
- StreamName name = new StreamName(section.getColumn(),
- section.getKind());
- streams.put(name,
- InStream.create(name.toString(), new ByteBuffer[]{sectionBuffer},
- new long[]{0}, sectionLength, codec, bufferSize));
- sectionOffset += sectionLength;
- }
- }
+ long start = stripe.getIndexLength();
+ long end = start + stripe.getDataLength();
+ // explicitly trigger 1 big read
+ DiskRange[] ranges = new DiskRange[]{new DiskRange(start, end)};
+ bufferChunks = readDiskRanges(file, stripe.getOffset(), Arrays.asList(ranges));
+ List<OrcProto.Stream> streamDescriptions = stripeFooter.getStreamsList();
+ createStreams(streamDescriptions, bufferChunks, null, codec, bufferSize, streams);
}
/**
- * The secionts of stripe that we need to read.
+ * The sections of stripe that we need to read.
*/
static class DiskRange {
/** the first address we need to read. */
@@ -2275,6 +2278,30 @@ class RecordReaderImpl implements Record
}
}
+ /**
+ * The sections of stripe that we have read.
+ * This might not match diskRange - 1 disk range can be multiple buffer chunks, depending on DFS block boundaries.
+ */
+ static class BufferChunk {
+ final ByteBuffer chunk;
+ /** the first address we need to read. */
+ final long offset;
+ /** end of the buffer **/
+ final long end;
+
+ BufferChunk(ByteBuffer chunk, long offset) {
+ this.offset = offset;
+ this.chunk = chunk;
+ end = offset + chunk.remaining();
+ }
+
+ @Override
+ public final String toString() {
+ return "range start: " + offset + " size: " + chunk.remaining() + " type: "
+ + (chunk.isDirect() ? "direct" : "array-backed");
+ }
+ }
+
private static final int BYTE_STREAM_POSITIONS = 1;
private static final int RUN_LENGTH_BYTE_POSITIONS =
BYTE_STREAM_POSITIONS + 1;
@@ -2460,17 +2487,17 @@ class RecordReaderImpl implements Record
* ranges
* @throws IOException
*/
- static byte[][] readDiskRanges(FSDataInputStream file,
+ List<BufferChunk> readDiskRanges(FSDataInputStream file,
long base,
List<DiskRange> ranges) throws IOException {
- byte[][] result = new byte[ranges.size()][];
- int i = 0;
+ ArrayList<BufferChunk> result = new ArrayList<RecordReaderImpl.BufferChunk>(ranges.size());
for(DiskRange range: ranges) {
int len = (int) (range.end - range.offset);
- result[i] = new byte[len];
- file.seek(base + range.offset);
- file.readFully(result[i]);
- i += 1;
+ long off = range.offset;
+ file.seek(base + off);
+ byte[] buffer = new byte[len];
+ file.readFully(buffer, 0, buffer.length);
+ result.add(new BufferChunk(ByteBuffer.wrap(buffer), range.offset));
}
return result;
}
@@ -2509,8 +2536,7 @@ class RecordReaderImpl implements Record
}
static void createStreams(List<OrcProto.Stream> streamDescriptions,
- List<DiskRange> ranges,
- byte[][] bytes,
+ List<BufferChunk> ranges,
boolean[] includeColumn,
CompressionCodec codec,
int bufferSize,
@@ -2519,13 +2545,13 @@ class RecordReaderImpl implements Record
long offset = 0;
for(OrcProto.Stream streamDesc: streamDescriptions) {
int column = streamDesc.getColumn();
- if (includeColumn[column] &&
+ if ((includeColumn == null || includeColumn[column]) &&
StreamName.getArea(streamDesc.getKind()) == StreamName.Area.DATA) {
long length = streamDesc.getLength();
int first = -1;
int last = -2;
- for(int i=0; i < bytes.length; ++i) {
- DiskRange range = ranges.get(i);
+ for(int i=0; i < ranges.size(); ++i) {
+ BufferChunk range = ranges.get(i);
if (overlap(offset, offset+length, range.offset, range.end)) {
if (first == -1) {
first = i;
@@ -2536,12 +2562,24 @@ class RecordReaderImpl implements Record
ByteBuffer[] buffers = new ByteBuffer[last - first + 1];
long[] offsets = new long[last - first + 1];
for(int i=0; i < buffers.length; ++i) {
- DiskRange range = ranges.get(i + first);
+ BufferChunk range = ranges.get(i + first);
long start = Math.max(range.offset, offset);
long end = Math.min(range.end, offset+length);
- buffers[i] = ByteBuffer.wrap(bytes[first + i],
- Math.max(0, (int) (offset - range.offset)), (int) (end - start));
- offsets[i] = Math.max(0, range.offset - offset);
+ buffers[i] = range.chunk.slice();
+ assert range.chunk.position() == 0; // otherwise we'll mix up positions
+ /*
+ * buffers are positioned in-wards if the offset > range.offset
+ * offsets[i] == range.offset - offset, except if offset > range.offset
+ */
+ if(offset > range.offset) {
+ buffers[i].position((int)(offset - range.offset));
+ buffers[i].limit((int)(end - range.offset));
+ offsets[i] = 0;
+ } else {
+ buffers[i].position(0);
+ buffers[i].limit((int)(end - range.offset));
+ offsets[i] = (range.offset - offset);
+ }
}
StreamName name = new StreamName(column, streamDesc.getKind());
streams.put(name, InStream.create(name.toString(), buffers, offsets,
@@ -2565,8 +2603,8 @@ class RecordReaderImpl implements Record
if (LOG.isDebugEnabled()) {
LOG.debug("merge = " + stringifyDiskRanges(chunks));
}
- byte[][] bytes = readDiskRanges(file, stripe.getOffset(), chunks);
- createStreams(streamList, chunks, bytes, included, codec, bufferSize,
+ bufferChunks = readDiskRanges(file, stripe.getOffset(), chunks);
+ createStreams(streamList, bufferChunks, included, codec, bufferSize,
streams);
}
@@ -2666,6 +2704,7 @@ class RecordReaderImpl implements Record
@Override
public void close() throws IOException {
+ clearStreams();
file.close();
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/VectorizedOrcSerde.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/VectorizedOrcSerde.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/VectorizedOrcSerde.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/VectorizedOrcSerde.java Tue Nov 26 08:19:25 2013
@@ -69,13 +69,11 @@ public class VectorizedOrcSerde extends
}
for (int p = 0; p < batch.projectionSize; p++) {
int k = batch.projectedColumns[p];
- Writable w;
if (batch.cols[k].isRepeating) {
- w = (Writable) valueWriters[p].writeValue(batch.cols[k], 0);
+ valueWriters[p].setValue(ost, batch.cols[k], 0);
} else {
- w = (Writable) valueWriters[p].writeValue(batch.cols[k], index);
+ valueWriters[p].setValue(ost, batch.cols[k], index);
}
- ost.setFieldValue(k, w);
}
OrcSerdeRow row = (OrcSerdeRow) orcRowArray[i];
row.realRow = ost;
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/merge/BlockMergeTask.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/merge/BlockMergeTask.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/merge/BlockMergeTask.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/merge/BlockMergeTask.java Tue Nov 26 08:19:25 2013
@@ -372,10 +372,5 @@ public class BlockMergeTask extends Task
@Override
public void logPlanProgress(SessionState ss) throws IOException {
// no op
- }
-
- @Override
- public void updateCounters(Counters ctrs, RunningJob rj) throws IOException {
- // no op
- }
+ }
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/stats/PartialScanTask.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/stats/PartialScanTask.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/stats/PartialScanTask.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/stats/PartialScanTask.java Tue Nov 26 08:19:25 2013
@@ -205,9 +205,9 @@ public class PartialScanTask extends Tas
if (work.isGatheringStats()) {
// initialize stats publishing table
StatsPublisher statsPublisher;
- String statsImplementationClass = HiveConf.getVar(job, HiveConf.ConfVars.HIVESTATSDBCLASS);
- if (StatsFactory.setImplementation(statsImplementationClass, job)) {
- statsPublisher = StatsFactory.getStatsPublisher();
+ StatsFactory factory = StatsFactory.newFactory(job);
+ if (factory != null) {
+ statsPublisher = factory.getStatsPublisher();
if (!statsPublisher.init(job)) { // creating stats table if not exists
if (HiveConf.getBoolVar(job, HiveConf.ConfVars.HIVE_STATS_RELIABLE)) {
throw
@@ -377,9 +377,4 @@ public class PartialScanTask extends Tas
public void logPlanProgress(SessionState ss) throws IOException {
// no op
}
-
- @Override
- public void updateCounters(Counters ctrs, RunningJob rj) throws IOException {
- // no op
- }
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateTask.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateTask.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateTask.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateTask.java Tue Nov 26 08:19:25 2013
@@ -252,9 +252,4 @@ public class ColumnTruncateTask extends
public void logPlanProgress(SessionState ss) throws IOException {
// no op
}
-
- @Override
- public void updateCounters(Counters ctrs, RunningJob rj) throws IOException {
- // no op
- }
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java Tue Nov 26 08:19:25 2013
@@ -93,10 +93,24 @@ public class ZooKeeperHiveLockManager im
* Get the quorum server address from the configuration. The format is:
* host1:port, host2:port..
**/
- private static String getQuorumServers(HiveConf conf) {
- String hosts = conf.getVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_QUORUM);
+ @VisibleForTesting
+ static String getQuorumServers(HiveConf conf) {
+ String[] hosts = conf.getVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_QUORUM).split(",");
String port = conf.getVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_CLIENT_PORT);
- return hosts + ":" + port;
+ StringBuilder quorum = new StringBuilder();
+ for(int i=0; i<hosts.length; i++) {
+ quorum.append(hosts[i].trim());
+ if (!hosts[i].contains(":")) {
+ // if the hostname doesn't contain a port, add the configured port to hostname
+ quorum.append(":");
+ quorum.append(port);
+ }
+
+ if (i != hosts.length-1)
+ quorum.append(",");
+ }
+
+ return quorum.toString();
}
/**
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java Tue Nov 26 08:19:25 2013
@@ -2478,7 +2478,7 @@ private void constructOneLBLocationMap(F
try {
return getMSC().updateTableColumnStatistics(statsObj);
} catch (Exception e) {
- LOG.error(StringUtils.stringifyException(e));
+ LOG.debug(StringUtils.stringifyException(e));
throw new HiveException(e);
}
}
@@ -2487,7 +2487,7 @@ private void constructOneLBLocationMap(F
try {
return getMSC().updatePartitionColumnStatistics(statsObj);
} catch (Exception e) {
- LOG.error(StringUtils.stringifyException(e));
+ LOG.debug(StringUtils.stringifyException(e));
throw new HiveException(e);
}
}
@@ -2497,7 +2497,7 @@ private void constructOneLBLocationMap(F
try {
return getMSC().getTableColumnStatistics(dbName, tableName, colName);
} catch (Exception e) {
- LOG.error(StringUtils.stringifyException(e));
+ LOG.debug(StringUtils.stringifyException(e));
throw new HiveException(e);
}
@@ -2508,7 +2508,7 @@ private void constructOneLBLocationMap(F
try {
return getMSC().getPartitionColumnStatistics(dbName, tableName, partName, colName);
} catch (Exception e) {
- LOG.error(StringUtils.stringifyException(e));
+ LOG.debug(StringUtils.stringifyException(e));
throw new HiveException(e);
}
}
@@ -2518,7 +2518,7 @@ private void constructOneLBLocationMap(F
try {
return getMSC().deleteTableColumnStatistics(dbName, tableName, colName);
} catch(Exception e) {
- LOG.error(StringUtils.stringifyException(e));
+ LOG.debug(StringUtils.stringifyException(e));
throw new HiveException(e);
}
}
@@ -2528,7 +2528,7 @@ private void constructOneLBLocationMap(F
try {
return getMSC().deletePartitionColumnStatistics(dbName, tableName, partName, colName);
} catch(Exception e) {
- LOG.error(StringUtils.stringifyException(e));
+ LOG.debug(StringUtils.stringifyException(e));
throw new HiveException(e);
}
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java Tue Nov 26 08:19:25 2013
@@ -86,17 +86,16 @@ public class ConvertJoinMapJoin implemen
for (Operator<? extends OperatorDesc> parentOp : joinOp.getParentOperators()) {
- Statistics currInputStat = null;
- try {
- currInputStat = parentOp.getStatistics(context.conf);
- } catch (HiveException e) {
+ Statistics currInputStat = parentOp.getStatistics();
+ if (currInputStat == null) {
+ LOG.warn("Couldn't get statistics from: "+parentOp);
return null;
}
- long inputSize = currInputStat.getNumberOfBytes();
+ long inputSize = currInputStat.getDataSize();
if ((bigInputStat == null) ||
((bigInputStat != null) &&
- (inputSize > bigInputStat.getNumberOfBytes()))) {
+ (inputSize > bigInputStat.getDataSize()))) {
if (bigTableFound) {
// cannot convert to map join; we've already chosen a big table
@@ -117,7 +116,7 @@ public class ConvertJoinMapJoin implemen
if (bigInputStat != null) {
// we're replacing the current big table with a new one. Need
// to count the current one as a map table then.
- totalSize += bigInputStat.getNumberOfBytes();
+ totalSize += bigInputStat.getDataSize();
}
if (totalSize > maxSize) {
@@ -131,7 +130,7 @@ public class ConvertJoinMapJoin implemen
bigInputStat = currInputStat;
}
} else {
- totalSize += currInputStat.getNumberOfBytes();
+ totalSize += currInputStat.getDataSize();
if (totalSize > maxSize) {
// cannot hold all map tables in memory. Cannot convert.
return null;
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java Tue Nov 26 08:19:25 2013
@@ -1366,7 +1366,7 @@ public final class GenMapRedUtils {
public static void addStatsTask(FileSinkOperator nd, MoveTask mvTask,
Task<? extends Serializable> currTask, HiveConf hconf) {
- MoveWork mvWork = ((MoveTask) mvTask).getWork();
+ MoveWork mvWork = mvTask.getWork();
StatsWork statsWork = null;
if (mvWork.getLoadTableWork() != null) {
statsWork = new StatsWork(mvWork.getLoadTableWork());
@@ -1374,6 +1374,8 @@ public final class GenMapRedUtils {
statsWork = new StatsWork(mvWork.getLoadFileWork());
}
assert statsWork != null : "Error when genereting StatsTask";
+
+ statsWork.setSourceTask(currTask);
statsWork.setStatsReliable(hconf.getBoolVar(ConfVars.HIVE_STATS_RELIABLE));
if (currTask.getWork() instanceof MapredWork) {
@@ -1391,7 +1393,7 @@ public final class GenMapRedUtils {
// AggKey in StatsWork is used for stats aggregation while StatsAggPrefix
// in FileSinkDesc is used for stats publishing. They should be consistent.
- statsWork.setAggKey(((FileSinkOperator) nd).getConf().getStatsAggPrefix());
+ statsWork.setAggKey(nd.getConf().getStatsAggPrefix());
Task<? extends Serializable> statsTask = TaskFactory.get(statsWork, hconf);
// mark the MapredWork and FileSinkOperator for gathering stats
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java Tue Nov 26 08:19:25 2013
@@ -543,6 +543,7 @@ public class MapJoinProcessor implements
MapJoinDesc mapJoinDescriptor = new MapJoinDesc(keyExprMap, keyTableDesc, newValueExprs,
valueTableDescs, valueFiltedTableDescs, outputColumnNames, mapJoinPos, joinCondns,
filters, op.getConf().getNoOuterJoin(), dumpFilePrefix);
+ mapJoinDescriptor.setStatistics(op.getConf().getStatistics());
mapJoinDescriptor.setTagOrder(tagOrder);
mapJoinDescriptor.setNullSafes(desc.getNullSafes());
mapJoinDescriptor.setFilterMap(desc.getFilterMap());
@@ -601,6 +602,8 @@ public class MapJoinProcessor implements
bigTablePos, smbJoinDesc.getConds(),
smbJoinDesc.getFilters(), smbJoinDesc.isNoOuterJoin(), smbJoinDesc.getDumpFilePrefix());
+ mapJoinDesc.setStatistics(smbJoinDesc.getStatistics());
+
RowResolver joinRS = opParseCtxMap.get(smbJoinOp).getRowResolver();
// The mapjoin has the same schema as the join operator
MapJoinOperator mapJoinOp = (MapJoinOperator) OperatorFactory.getAndMakeChild(
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java Tue Nov 26 08:19:25 2013
@@ -29,6 +29,7 @@ import org.apache.hadoop.hive.ql.optimiz
import org.apache.hadoop.hive.ql.optimizer.listbucketingpruner.ListBucketingPruner;
import org.apache.hadoop.hive.ql.optimizer.pcr.PartitionConditionRemover;
import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
+import org.apache.hadoop.hive.ql.optimizer.stats.annotation.AnnotateWithStatistics;
import org.apache.hadoop.hive.ql.optimizer.unionproc.UnionProcessor;
import org.apache.hadoop.hive.ql.parse.ParseContext;
import org.apache.hadoop.hive.ql.parse.SemanticException;
@@ -114,6 +115,10 @@ public class Optimizer {
if(HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTIMIZEMETADATAQUERIES)) {
transformations.add(new StatsOptimizer());
}
+ if (pctx.getContext().getExplain() ||
+ HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVE_OPTIMIZE_TEZ)) {
+ transformations.add(new AnnotateWithStatistics());
+ }
transformations.add(new SimpleFetchOptimizer()); // must be called last
if (HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEFETCHTASKAGGR)) {
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SetReducerParallelism.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SetReducerParallelism.java?rev=1545564&r1=1545563&r2=1545564&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SetReducerParallelism.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SetReducerParallelism.java Tue Nov 26 08:19:25 2013
@@ -29,7 +29,6 @@ import org.apache.hadoop.hive.ql.exec.Ut
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.lib.NodeProcessor;
import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.OptimizeTezProcContext;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
@@ -66,30 +65,30 @@ public class SetReducerParallelism imple
context.visitedReduceSinks.add(sink);
- try {
- if (desc.getNumReducers() <= 0) {
- if (constantReducers > 0) {
- LOG.info("Parallelism for reduce sink "+sink+" set by user to "+constantReducers);
- desc.setNumReducers(constantReducers);
- } else {
- long numberOfBytes = 0;
-
- // we need to add up all the estimates from the siblings of this reduce sink
- for (Operator<? extends OperatorDesc> sibling:
- sink.getChildOperators().get(0).getParentOperators()) {
- numberOfBytes += sibling.getStatistics(context.conf).getNumberOfBytes();
- }
+ if (desc.getNumReducers() <= 0) {
+ if (constantReducers > 0) {
+ LOG.info("Parallelism for reduce sink "+sink+" set by user to "+constantReducers);
+ desc.setNumReducers(constantReducers);
+ } else {
+ long numberOfBytes = 0;
- int numReducers = Utilities.estimateReducers(numberOfBytes, bytesPerReducer,
- maxReducers, false);
- LOG.info("Set parallelism for reduce sink "+sink+" to: "+numReducers);
- desc.setNumReducers(numReducers);
+ // we need to add up all the estimates from the siblings of this reduce sink
+ for (Operator<? extends OperatorDesc> sibling:
+ sink.getChildOperators().get(0).getParentOperators()) {
+ if (sibling.getStatistics() != null) {
+ numberOfBytes += sibling.getStatistics().getDataSize();
+ } else {
+ LOG.warn("No stats available from: "+sibling);
+ }
}
- } else {
- LOG.info("Number of reducers determined to be: "+desc.getNumReducers());
+
+ int numReducers = Utilities.estimateReducers(numberOfBytes, bytesPerReducer,
+ maxReducers, false);
+ LOG.info("Set parallelism for reduce sink "+sink+" to: "+numReducers);
+ desc.setNumReducers(numReducers);
}
- } catch (HiveException e) {
- throw new SemanticException(e);
+ } else {
+ LOG.info("Number of reducers determined to be: "+desc.getNumReducers());
}
return false;