You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ha...@apache.org on 2013/09/20 02:33:01 UTC
svn commit: r1524874 [4/9] - in /hive/branches/vectorization: ./
cli/src/java/org/apache/hadoop/hive/cli/
cli/src/test/org/apache/hadoop/hive/cli/
common/src/java/org/apache/hadoop/hive/common/type/
common/src/java/org/apache/hadoop/hive/conf/ common/s...
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java Fri Sep 20 00:32:55 2013
@@ -20,7 +20,6 @@ package org.apache.hadoop.hive.ql.exec;
import java.io.Serializable;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -34,7 +33,6 @@ import org.apache.hadoop.hive.ql.plan.Op
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.ql.plan.api.OperatorType;
import org.apache.hadoop.hive.serde2.Deserializer;
-import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.util.ReflectionUtils;
@@ -74,17 +72,15 @@ public class DemuxOperator extends Opera
// We need to know the mapping from the newTag to oldTag and revert
// the newTag to oldTag to make operators in the operator tree
// function correctly.
- private Map<Integer, Integer> newTagToOldTag =
- new HashMap<Integer, Integer>();
+ private int[] newTagToOldTag;
// The mapping from a newTag to the index of the corresponding child
// of this operator.
- private Map<Integer, Integer> newTagToChildIndex =
- new HashMap<Integer, Integer>();
+ private int[] newTagToChildIndex;
// The mapping from the index of a child operator to its corresponding
// inputObjectInspectors
- private Map<Integer, ObjectInspector[]> childInputObjInspectors;
+ private ObjectInspector[][] childInputObjInspectors;
private int childrenDone;
@@ -107,7 +103,7 @@ public class DemuxOperator extends Opera
// In this case, the parent list of MUX is [DEMUX, GBY, DEMUX],
// so we need to have two childOperatorsTags (the index of this DemuxOperator in
// its children's parents lists, also see childOperatorsTag in Operator) at here.
- private List<List<Integer>> newChildOperatorsTag;
+ private int[][] newChildOperatorsTag;
@Override
protected void initializeOp(Configuration hconf) throws HiveException {
@@ -117,28 +113,28 @@ public class DemuxOperator extends Opera
"Expected number of children is at least 1. Found : " + childOperatorsArray.length);
}
- newTagToOldTag = conf.getNewTagToOldTag();
- newTagToChildIndex = conf.getNewTagToChildIndex();
- childInputObjInspectors = new HashMap<Integer, ObjectInspector[]>();
- cntrs = new long[newTagToOldTag.size()];
- nextCntrs = new long[newTagToOldTag.size()];
+ newTagToOldTag = toArray(conf.getNewTagToOldTag());
+ newTagToChildIndex = toArray(conf.getNewTagToChildIndex());
+ childInputObjInspectors = new ObjectInspector[childOperators.size()][];
+ cntrs = new long[newTagToOldTag.length];
+ nextCntrs = new long[newTagToOldTag.length];
try {
// We populate inputInspectors for all children of this DemuxOperator.
// Those inputObjectInspectors are stored in childInputObjInspectors.
- for (Entry<Integer, Integer> e1: newTagToOldTag.entrySet()) {
- int newTag = e1.getKey();
- int oldTag = e1.getValue();
- int childIndex = newTagToChildIndex.get(newTag);
+ for (int i = 0; i < newTagToOldTag.length; i++) {
+ int newTag = i;
+ int oldTag = newTagToOldTag[i];
+ int childIndex = newTagToChildIndex[newTag];
cntrs[newTag] = 0;
nextCntrs[newTag] = 0;
TableDesc keyTableDesc = conf.getKeysSerializeInfos().get(newTag);
- Deserializer inputKeyDeserializer = (SerDe) ReflectionUtils.newInstance(keyTableDesc
+ Deserializer inputKeyDeserializer = ReflectionUtils.newInstance(keyTableDesc
.getDeserializerClass(), null);
inputKeyDeserializer.initialize(null, keyTableDesc.getProperties());
TableDesc valueTableDesc = conf.getValuesSerializeInfos().get(newTag);
- Deserializer inputValueDeserializer = (SerDe) ReflectionUtils.newInstance(valueTableDesc
+ Deserializer inputValueDeserializer = ReflectionUtils.newInstance(valueTableDesc
.getDeserializerClass(), null);
inputValueDeserializer.initialize(null, valueTableDesc.getProperties());
@@ -148,19 +144,20 @@ public class DemuxOperator extends Opera
int childParentsCount = conf.getChildIndexToOriginalNumParents().get(childIndex);
// Multiple newTags can point to the same child (e.g. when the child is a JoinOperator).
// So, we first check if childInputObjInspectors contains the key of childIndex.
- if (!childInputObjInspectors.containsKey(childIndex)) {
- childInputObjInspectors.put(childIndex, new ObjectInspector[childParentsCount]);
+ if (childInputObjInspectors[childIndex] == null) {
+ childInputObjInspectors[childIndex] = new ObjectInspector[childParentsCount];
}
- ObjectInspector[] ois = childInputObjInspectors.get(childIndex);
+ ObjectInspector[] ois = childInputObjInspectors[childIndex];
ois[oldTag] = ObjectInspectorFactory
.getStandardStructObjectInspector(Utilities.reduceFieldNameList, oi);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
- this.childrenDone = 0;
- newChildOperatorsTag = new ArrayList<List<Integer>>();
- for (Operator<? extends OperatorDesc> child: childOperators) {
+ childrenDone = 0;
+ newChildOperatorsTag = new int[childOperators.size()][];
+ for (int i = 0; i < childOperators.size(); i++) {
+ Operator<? extends OperatorDesc> child = childOperators.get(i);
List<Integer> childOperatorTags = new ArrayList<Integer>();
if (child instanceof MuxOperator) {
// This DemuxOperator can appear multiple times in MuxOperator's
@@ -175,12 +172,28 @@ public class DemuxOperator extends Opera
} else {
childOperatorTags.add(child.getParentOperators().indexOf(this));
}
- newChildOperatorsTag.add(childOperatorTags);
+ newChildOperatorsTag[i] = toArray(childOperatorTags);
}
LOG.info("newChildOperatorsTag " + newChildOperatorsTag);
initializeChildren(hconf);
}
+ private int[] toArray(List<Integer> list) {
+ int[] array = new int[list.size()];
+ for (int i = 0; i < list.size(); i++) {
+ array[i] = list.get(i);
+ }
+ return array;
+ }
+
+ private int[] toArray(Map<Integer, Integer> map) {
+ int[] array = new int[map.size()];
+ for (Entry<Integer, Integer> entry : map.entrySet()) {
+ array[entry.getKey()] = entry.getValue();
+ }
+ return array;
+ }
+
// Each child should has its own outputObjInspector
@Override
protected void initializeChildren(Configuration hconf) throws HiveException {
@@ -190,7 +203,7 @@ public class DemuxOperator extends Opera
for (int i = 0; i < childOperatorsArray.length; i++) {
LOG.info("Initializing child " + i + " " + childOperatorsArray[i].getIdentifier() + " " +
childOperatorsArray[i].getName() +
- " " + childInputObjInspectors.get(i).length);
+ " " + childInputObjInspectors[i].length);
// We need to initialize those MuxOperators first because if we first
// initialize other operators, the states of all parents of those MuxOperators
// are INIT (including this DemuxOperator),
@@ -200,7 +213,7 @@ public class DemuxOperator extends Opera
// that MuxOperator must be the parent of a JoinOperator.
// In this case, that MuxOperator should be initialized
// by multiple parents (of that MuxOperator).
- ObjectInspector[] ois = childInputObjInspectors.get(i);
+ ObjectInspector[] ois = childInputObjInspectors[i];
for (int j = 0; j < ois.length; j++) {
if (ois[j] != null) {
childOperatorsArray[i].initialize(hconf, ois[j], j);
@@ -216,9 +229,9 @@ public class DemuxOperator extends Opera
for (int i = 0; i < childOperatorsArray.length; i++) {
LOG.info("Initializing child " + i + " " + childOperatorsArray[i].getIdentifier() + " " +
childOperatorsArray[i].getName() +
- " " + childInputObjInspectors.get(i).length);
+ " " + childInputObjInspectors[i].length);
if (!(childOperatorsArray[i] instanceof MuxOperator)) {
- childOperatorsArray[i].initialize(hconf, childInputObjInspectors.get(i));
+ childOperatorsArray[i].initialize(hconf, childInputObjInspectors[i]);
} else {
continue;
}
@@ -230,7 +243,7 @@ public class DemuxOperator extends Opera
@Override
public void processOp(Object row, int tag) throws HiveException {
- int currentChildIndex = newTagToChildIndex.get(tag);
+ int currentChildIndex = newTagToChildIndex[tag];
// Check if we start to forward rows to a new child.
// If so, in the current key group, rows will not be forwarded
@@ -239,7 +252,7 @@ public class DemuxOperator extends Opera
// to currentChildIndex (exclusive) and propagate processGroup to those children.
endGroupIfNecessary(currentChildIndex);
- int oldTag = newTagToOldTag.get(tag);
+ int oldTag = newTagToOldTag[tag];
if (isLogInfoEnabled) {
cntrs[tag]++;
if (cntrs[tag] == nextCntrs[tag]) {
@@ -272,10 +285,10 @@ public class DemuxOperator extends Opera
@Override
protected void closeOp(boolean abort) throws HiveException {
- for (Entry<Integer, Integer> entry: newTagToOldTag.entrySet()) {
- int newTag = entry.getKey();
- int oldTag = entry.getValue();
- int childIndex = newTagToChildIndex.get(newTag);
+ for (int i = 0 ; i < newTagToOldTag.length; i++) {
+ int newTag = i;
+ int oldTag = newTagToOldTag[i];
+ int childIndex = newTagToChildIndex[newTag];
LOG.info(id + " (newTag, childIndex, oldTag)=(" + newTag + ", " + childIndex + ", "
+ oldTag + "), forwarded " + cntrs[newTag] + " rows");
}
@@ -297,7 +310,7 @@ public class DemuxOperator extends Opera
Operator<? extends OperatorDesc> child = childOperatorsArray[i];
child.flush();
child.endGroup();
- for (Integer childTag: newChildOperatorsTag.get(i)) {
+ for (int childTag: newChildOperatorsTag[i]) {
child.processGroup(childTag);
}
}
@@ -328,7 +341,7 @@ public class DemuxOperator extends Opera
Operator<? extends OperatorDesc> child = childOperatorsArray[i];
child.flush();
child.endGroup();
- for (Integer childTag: newChildOperatorsTag.get(i)) {
+ for (int childTag: newChildOperatorsTag[i]) {
child.processGroup(childTag);
}
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java Fri Sep 20 00:32:55 2013
@@ -130,7 +130,7 @@ public class FetchOperator implements Se
List<ObjectInspector> inspectors = new ArrayList<ObjectInspector>(vcCols.size());
for (VirtualColumn vc : vcCols) {
inspectors.add(PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(
- vc.getTypeInfo().getPrimitiveCategory()));
+ vc.getTypeInfo()));
names.add(vc.getName());
}
vcsOI = ObjectInspectorFactory.getStandardStructObjectInspector(names, inspectors);
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java Fri Sep 20 00:32:55 2013
@@ -38,6 +38,7 @@ import org.apache.hadoop.hive.ql.ErrorMs
import org.apache.hadoop.hive.ql.io.HiveFileFormatUtils;
import org.apache.hadoop.hive.ql.io.HiveKey;
import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
+import org.apache.hadoop.hive.ql.io.HivePassThroughOutputFormat;
import org.apache.hadoop.hive.ql.io.HivePartitioner;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.DynamicPartitionCtx;
@@ -930,7 +931,19 @@ public class FileSinkOperator extends Te
public void checkOutputSpecs(FileSystem ignored, JobConf job) throws IOException {
if (hiveOutputFormat == null) {
try {
- hiveOutputFormat = conf.getTableInfo().getOutputFileFormatClass().newInstance();
+ if (getConf().getTableInfo().getJobProperties() != null) {
+ //Setting only for Storage Handler
+ if (getConf().getTableInfo().getJobProperties().get(HivePassThroughOutputFormat.HIVE_PASSTHROUGH_STORAGEHANDLER_OF_JOBCONFKEY) != null) {
+ job.set(HivePassThroughOutputFormat.HIVE_PASSTHROUGH_STORAGEHANDLER_OF_JOBCONFKEY,getConf().getTableInfo().getJobProperties().get(HivePassThroughOutputFormat.HIVE_PASSTHROUGH_STORAGEHANDLER_OF_JOBCONFKEY));
+ hiveOutputFormat = ReflectionUtils.newInstance(conf.getTableInfo().getOutputFileFormatClass(),job);
+ }
+ else {
+ hiveOutputFormat = conf.getTableInfo().getOutputFileFormatClass().newInstance();
+ }
+ }
+ else {
+ hiveOutputFormat = conf.getTableInfo().getOutputFileFormatClass().newInstance();
+ }
} catch (Exception ex) {
throw new IOException(ex);
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FunctionRegistry.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FunctionRegistry.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FunctionRegistry.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/FunctionRegistry.java Fri Sep 20 00:32:55 2013
@@ -39,12 +39,14 @@ import javax.xml.parsers.DocumentBuilder
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.common.type.HiveVarchar;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
import org.apache.hadoop.hive.ql.udf.GenericUDFDecode;
import org.apache.hadoop.hive.ql.udf.GenericUDFEncode;
+import org.apache.hadoop.hive.ql.udf.SettableUDF;
import org.apache.hadoop.hive.ql.udf.UDAFPercentile;
import org.apache.hadoop.hive.ql.udf.UDFAbs;
import org.apache.hadoop.hive.ql.udf.UDFAcos;
@@ -54,7 +56,6 @@ import org.apache.hadoop.hive.ql.udf.UDF
import org.apache.hadoop.hive.ql.udf.UDFBase64;
import org.apache.hadoop.hive.ql.udf.UDFBin;
import org.apache.hadoop.hive.ql.udf.UDFCeil;
-import org.apache.hadoop.hive.ql.udf.UDFConcat;
import org.apache.hadoop.hive.ql.udf.UDFConv;
import org.apache.hadoop.hive.ql.udf.UDFCos;
import org.apache.hadoop.hive.ql.udf.UDFDate;
@@ -78,7 +79,6 @@ import org.apache.hadoop.hive.ql.udf.UDF
import org.apache.hadoop.hive.ql.udf.UDFLog;
import org.apache.hadoop.hive.ql.udf.UDFLog10;
import org.apache.hadoop.hive.ql.udf.UDFLog2;
-import org.apache.hadoop.hive.ql.udf.UDFLower;
import org.apache.hadoop.hive.ql.udf.UDFLpad;
import org.apache.hadoop.hive.ql.udf.UDFMinute;
import org.apache.hadoop.hive.ql.udf.UDFMonth;
@@ -127,7 +127,6 @@ import org.apache.hadoop.hive.ql.udf.UDF
import org.apache.hadoop.hive.ql.udf.UDFType;
import org.apache.hadoop.hive.ql.udf.UDFUnbase64;
import org.apache.hadoop.hive.ql.udf.UDFUnhex;
-import org.apache.hadoop.hive.ql.udf.UDFUpper;
import org.apache.hadoop.hive.ql.udf.UDFWeekOfYear;
import org.apache.hadoop.hive.ql.udf.UDFYear;
import org.apache.hadoop.hive.ql.udf.generic.*;
@@ -158,6 +157,7 @@ import org.apache.hadoop.hive.serde2.typ
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.VarcharTypeParams;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.ReflectionUtils;
import org.w3c.dom.Document;
@@ -194,7 +194,7 @@ public final class FunctionRegistry {
static {
- registerUDF("concat", UDFConcat.class, false);
+ registerGenericUDF("concat", GenericUDFConcat.class);
registerUDF("substr", UDFSubstr.class, false);
registerUDF("substring", UDFSubstr.class, false);
registerUDF("space", UDFSpace.class, false);
@@ -243,10 +243,10 @@ public final class FunctionRegistry {
registerGenericUDF("encode", GenericUDFEncode.class);
registerGenericUDF("decode", GenericUDFDecode.class);
- registerUDF("upper", UDFUpper.class, false);
- registerUDF("lower", UDFLower.class, false);
- registerUDF("ucase", UDFUpper.class, false);
- registerUDF("lcase", UDFLower.class, false);
+ registerGenericUDF("upper", GenericUDFUpper.class);
+ registerGenericUDF("lower", GenericUDFLower.class);
+ registerGenericUDF("ucase", GenericUDFUpper.class);
+ registerGenericUDF("lcase", GenericUDFLower.class);
registerUDF("trim", UDFTrim.class, false);
registerUDF("ltrim", UDFLTrim.class, false);
registerUDF("rtrim", UDFRTrim.class, false);
@@ -360,6 +360,8 @@ public final class FunctionRegistry {
GenericUDFToBinary.class);
registerGenericUDF(serdeConstants.DECIMAL_TYPE_NAME,
GenericUDFToDecimal.class);
+ registerGenericUDF(serdeConstants.VARCHAR_TYPE_NAME,
+ GenericUDFToVarchar.class);
// Aggregate functions
registerGenericUDAF("max", new GenericUDAFMax());
@@ -381,6 +383,7 @@ public final class FunctionRegistry {
registerGenericUDAF("histogram_numeric", new GenericUDAFHistogramNumeric());
registerGenericUDAF("percentile_approx", new GenericUDAFPercentileApprox());
registerGenericUDAF("collect_set", new GenericUDAFCollectSet());
+ registerGenericUDAF("collect_list", new GenericUDAFCollectList());
registerGenericUDAF("ngrams", new GenericUDAFnGrams());
registerGenericUDAF("context_ngrams", new GenericUDAFContextNGrams());
@@ -625,6 +628,17 @@ public final class FunctionRegistry {
registerNumericType(PrimitiveCategory.STRING, 8);
}
+ static int getCommonLength(int aLen, int bLen) {
+ int maxLength;
+ if (aLen < 0 || bLen < 0) {
+ // negative length should take precedence over positive value?
+ maxLength = -1;
+ } else {
+ maxLength = Math.max(aLen, bLen);
+ }
+ return maxLength;
+ }
+
/**
* Given 2 TypeInfo types and the PrimitiveCategory selected as the common class between the two,
* return a TypeInfo corresponding to the common PrimitiveCategory, and with type qualifiers
@@ -642,6 +656,16 @@ public final class FunctionRegistry {
// For types with parameters (like varchar), we need to determine the type parameters
// that should be added to this type, based on the original 2 TypeInfos.
switch (typeCategory) {
+ case VARCHAR:
+ int maxLength = getCommonLength(
+ TypeInfoUtils.getCharacterLengthForType(a),
+ TypeInfoUtils.getCharacterLengthForType(b));
+ VarcharTypeParams varcharParams = new VarcharTypeParams();
+ varcharParams.setLength(maxLength);
+ // Generate type name so that we can retrieve the TypeInfo for that type.
+ String typeName = PrimitiveObjectInspectorUtils
+ .getTypeEntryFromTypeSpecs(typeCategory, varcharParams).toString();
+ return TypeInfoFactory.getPrimitiveTypeInfo(typeName);
default:
// Type doesn't require any qualifiers.
@@ -839,7 +863,6 @@ public final class FunctionRegistry {
return false;
}
-
/**
* Get the GenericUDAF evaluator for the name and argumentClasses.
*
@@ -1018,7 +1041,8 @@ public final class FunctionRegistry {
*/
public static int matchCost(TypeInfo argumentPassed,
TypeInfo argumentAccepted, boolean exact) {
- if (argumentAccepted.equals(argumentPassed)) {
+ if (argumentAccepted.equals(argumentPassed)
+ || TypeInfoUtils.doPrimitiveCategoriesMatch(argumentPassed, argumentAccepted)) {
// matches
return 0;
}
@@ -1281,18 +1305,38 @@ public final class FunctionRegistry {
return null;
}
+ GenericUDF clonedUDF = null;
if (genericUDF instanceof GenericUDFBridge) {
GenericUDFBridge bridge = (GenericUDFBridge) genericUDF;
- return new GenericUDFBridge(bridge.getUdfName(), bridge.isOperator(),
+ clonedUDF = new GenericUDFBridge(bridge.getUdfName(), bridge.isOperator(),
bridge.getUdfClassName());
} else if (genericUDF instanceof GenericUDFMacro) {
GenericUDFMacro bridge = (GenericUDFMacro) genericUDF;
- return new GenericUDFMacro(bridge.getMacroName(), bridge.getBody(),
+ clonedUDF = new GenericUDFMacro(bridge.getMacroName(), bridge.getBody(),
bridge.getColNames(), bridge.getColTypes());
+ } else {
+ clonedUDF = (GenericUDF) ReflectionUtils
+ .newInstance(genericUDF.getClass(), null);
+ }
+
+ if (clonedUDF != null) {
+ // The original may have settable info that needs to be added to the new copy.
+ if (genericUDF instanceof SettableUDF) {
+ try {
+ Object settableData = ((SettableUDF)genericUDF).getParams();
+ if (settableData != null) {
+ ((SettableUDF)clonedUDF).setParams(settableData);
+ }
+ } catch (UDFArgumentException err) {
+ // In theory this should not happen - if the original copy of the UDF had this
+ // data, we should be able to set the UDF copy with this same settableData.
+ LOG.error("Unable to add settable data to UDF " + genericUDF.getClass());
+ throw new IllegalArgumentException(err);
+ }
+ }
}
- return (GenericUDF) ReflectionUtils
- .newInstance(genericUDF.getClass(), null);
+ return clonedUDF;
}
/**
@@ -1447,6 +1491,7 @@ public final class FunctionRegistry {
udfClass == UDFToDouble.class || udfClass == UDFToFloat.class ||
udfClass == UDFToInteger.class || udfClass == UDFToLong.class ||
udfClass == UDFToShort.class || udfClass == UDFToString.class ||
+ udfClass == GenericUDFToVarchar.class ||
udfClass == GenericUDFTimestamp.class || udfClass == GenericUDFToBinary.class ||
udfClass == GenericUDFToDate.class;
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java Fri Sep 20 00:32:55 2013
@@ -226,7 +226,7 @@ public abstract class Operator<T extends
* optimizer and built during semantic analysis contains only key elements for
* reduce sink and group by op
*/
- protected transient Map<String, ExprNodeDesc> colExprMap;
+ protected Map<String, ExprNodeDesc> colExprMap;
public void setId(String id) {
this.id = id;
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java Fri Sep 20 00:32:55 2013
@@ -212,6 +212,16 @@ public class ScriptOperator extends Oper
if (f.isFile() && f.canRead()) {
return f;
}
+ if (Shell.WINDOWS) {
+ // Try filename with executable extentions
+ String[] exts = new String[] {".exe", ".bat"};
+ for (String ext : exts) {
+ File fileWithExt = new File(f.toString() + ext);
+ if (fileWithExt.isFile() && fileWithExt.canRead()) {
+ return fileWithExt;
+ }
+ }
+ }
} catch (Exception exp) {
}
classvalue = classvalue.substring(val + 1).trim();
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/SecureCmdDoAs.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/SecureCmdDoAs.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/SecureCmdDoAs.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/SecureCmdDoAs.java Fri Sep 20 00:32:55 2013
@@ -38,14 +38,6 @@ public class SecureCmdDoAs {
tokenPath = ShimLoader.getHadoopShims().createDelegationTokenFile(conf);
}
- public String addArg(String cmdline) throws HiveException {
- StringBuilder sb = new StringBuilder();
- sb.append(cmdline);
- sb.append(" -hadooptoken ");
- sb.append(tokenPath.toUri().getPath());
- return sb.toString();
- }
-
public void addEnv(Map<String, String> env){
env.put(ShimLoader.getHadoopShims().getTokenFileLocEnvName(),
tokenPath.toUri().getPath());
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java Fri Sep 20 00:32:55 2013
@@ -79,6 +79,7 @@ import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import org.antlr.runtime.CommonToken;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.WordUtils;
import org.apache.commons.logging.Log;
@@ -98,6 +99,7 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.ql.Context;
+import org.apache.hadoop.hive.ql.Driver;
import org.apache.hadoop.hive.ql.ErrorMsg;
import org.apache.hadoop.hive.ql.QueryPlan;
import org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter;
@@ -177,6 +179,7 @@ import org.apache.hadoop.util.Shell;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.serializers.FieldSerializer;
/**
* Utilities.
@@ -216,7 +219,8 @@ public final class Utilities {
private static Map<Path, BaseWork> gWorkMap = Collections
.synchronizedMap(new HashMap<Path, BaseWork>());
- private static final Log LOG = LogFactory.getLog(Utilities.class.getName());
+ private static final String CLASS_NAME = Utilities.class.getName();
+ private static final Log LOG = LogFactory.getLog(CLASS_NAME);
public static void clearWork(Configuration conf) {
Path mapPath = getPlanPath(conf, MAP_PLAN_NAME);
@@ -633,6 +637,34 @@ public final class Utilities {
}
+ private static class CommonTokenSerializer extends com.esotericsoftware.kryo.Serializer<CommonToken> {
+ @Override
+ public CommonToken read(Kryo kryo, Input input, Class<CommonToken> clazz) {
+ return new CommonToken(input.readInt(), input.readString());
+ }
+
+ @Override
+ public void write(Kryo kryo, Output output, CommonToken token) {
+ output.writeInt(token.getType());
+ output.writeString(token.getText());
+ }
+ }
+ private static void serializePlan(Object plan, OutputStream out, Configuration conf, boolean cloningPlan) {
+ PerfLogger perfLogger = PerfLogger.getPerfLogger();
+ perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.SERIALIZE_PLAN);
+ String serializationType = conf.get(HiveConf.ConfVars.PLAN_SERIALIZATION.varname, "kryo");
+ LOG.info("Serializing " + plan.getClass().getSimpleName() + " via " + serializationType);
+ if("javaXML".equalsIgnoreCase(serializationType)) {
+ serializeObjectByJavaXML(plan, out);
+ } else {
+ if(cloningPlan) {
+ serializeObjectByKryo(cloningQueryPlanKryo.get(), plan, out);
+ } else {
+ serializeObjectByKryo(runtimeSerializationKryo.get(), plan, out);
+ }
+ }
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.SERIALIZE_PLAN);
+ }
/**
* Serializes the plan.
* @param plan The plan, such as QueryPlan, MapredWork, etc.
@@ -640,16 +672,27 @@ public final class Utilities {
* @param conf to pick which serialization format is desired.
*/
public static void serializePlan(Object plan, OutputStream out, Configuration conf) {
+ serializePlan(plan, out, conf, false);
+ }
+
+ private static <T> T deserializePlan(InputStream in, Class<T> planClass, Configuration conf, boolean cloningPlan) {
PerfLogger perfLogger = PerfLogger.getPerfLogger();
- perfLogger.PerfLogBegin(LOG, PerfLogger.SERIALIZE_PLAN);
- if(conf.get(HiveConf.ConfVars.PLAN_SERIALIZATION.varname, "kryo").equals("javaXML")) {
- serializeObjectByJavaXML(plan, out);
+ perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.DESERIALIZE_PLAN);
+ T plan;
+ String serializationType = conf.get(HiveConf.ConfVars.PLAN_SERIALIZATION.varname, "kryo");
+ LOG.info("Deserializing " + planClass.getSimpleName() + " via " + serializationType);
+ if("javaXML".equalsIgnoreCase(serializationType)) {
+ plan = deserializeObjectByJavaXML(in);
} else {
- serializeObjectByKryo(plan, out);
+ if(cloningPlan) {
+ plan = deserializeObjectByKryo(cloningQueryPlanKryo.get(), in, planClass);
+ } else {
+ plan = deserializeObjectByKryo(runtimeSerializationKryo.get(), in, planClass);
+ }
}
- perfLogger.PerfLogEnd(LOG, PerfLogger.SERIALIZE_PLAN);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.DESERIALIZE_PLAN);
+ return plan;
}
-
/**
* Deserializes the plan.
* @param in The stream to read from.
@@ -657,16 +700,7 @@ public final class Utilities {
* @param To know what serialization format plan is in
*/
public static <T> T deserializePlan(InputStream in, Class<T> planClass, Configuration conf) {
- PerfLogger perfLogger = PerfLogger.getPerfLogger();
- perfLogger.PerfLogBegin(LOG, PerfLogger.DESERIALIZE_PLAN);
- T plan;
- if(conf.get(HiveConf.ConfVars.PLAN_SERIALIZATION.varname, "kryo").equals("javaXML")) {
- plan = deserializeObjectByJavaXML(in);
- } else {
- plan = deserializeObjectByKryo(in, planClass);
- }
- perfLogger.PerfLogEnd(LOG, PerfLogger.DESERIALIZE_PLAN);
- return plan;
+ return deserializePlan(in, planClass, conf, false);
}
/**
@@ -677,12 +711,13 @@ public final class Utilities {
public static MapredWork clonePlan(MapredWork plan) {
// TODO: need proper clone. Meanwhile, let's at least keep this horror in one place
PerfLogger perfLogger = PerfLogger.getPerfLogger();
- perfLogger.PerfLogBegin(LOG, PerfLogger.CLONE_PLAN);
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- Utilities.serializeObjectByJavaXML(plan, baos);
- MapredWork newPlan = Utilities.deserializeObjectByJavaXML(
- new ByteArrayInputStream(baos.toByteArray()));
- perfLogger.PerfLogEnd(LOG, PerfLogger.CLONE_PLAN);
+ perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.CLONE_PLAN);
+ ByteArrayOutputStream baos = new ByteArrayOutputStream(4096);
+ Configuration conf = new Configuration();
+ serializePlan(plan, baos, conf, true);
+ MapredWork newPlan = deserializePlan(new ByteArrayInputStream(baos.toByteArray()),
+ MapredWork.class, conf, true);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.CLONE_PLAN);
return newPlan;
}
@@ -716,9 +751,9 @@ public final class Utilities {
* @param plan Usually of type MapredWork, MapredLocalWork etc.
* @param out stream in which serialized plan is written into
*/
- private static void serializeObjectByKryo(Object plan, OutputStream out) {
+ private static void serializeObjectByKryo(Kryo kryo, Object plan, OutputStream out) {
Output output = new Output(out);
- kryo.get().writeObject(output, plan);
+ kryo.writeObject(output, plan);
output.close();
}
@@ -739,26 +774,44 @@ public final class Utilities {
}
}
- private static <T> T deserializeObjectByKryo(InputStream in, Class<T> clazz ) {
+ private static <T> T deserializeObjectByKryo(Kryo kryo, InputStream in, Class<T> clazz ) {
Input inp = new Input(in);
- T t = kryo.get().readObject(inp,clazz);
+ T t = kryo.readObject(inp,clazz);
inp.close();
return t;
}
// Kryo is not thread-safe,
// Also new Kryo() is expensive, so we want to do it just once.
- private static ThreadLocal<Kryo> kryo = new ThreadLocal<Kryo>() {
-
+ private static ThreadLocal<Kryo> runtimeSerializationKryo = new ThreadLocal<Kryo>() {
@Override
protected synchronized Kryo initialValue() {
Kryo kryo = new Kryo();
kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
kryo.register(java.sql.Date.class, new SqlDateSerializer());
+ removeField(kryo, Operator.class, "colExprMap");
+ removeField(kryo, ColumnInfo.class, "objectInspector");
+ removeField(kryo, MapWork.class, "opParseCtxMap");
+ removeField(kryo, MapWork.class, "joinTree");
+ return kryo;
+ };
+ };
+ @SuppressWarnings("rawtypes")
+ protected static void removeField(Kryo kryo, Class type, String fieldName) {
+ FieldSerializer fld = new FieldSerializer(kryo, type);
+ fld.removeField(fieldName);
+ kryo.register(type, fld);
+ }
+ private static ThreadLocal<Kryo> cloningQueryPlanKryo = new ThreadLocal<Kryo>() {
+ @Override
+ protected synchronized Kryo initialValue() {
+ Kryo kryo = new Kryo();
+ kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
+ kryo.register(CommonToken.class, new CommonTokenSerializer());
+ kryo.register(java.sql.Date.class, new SqlDateSerializer());
return kryo;
};
};
-
public static TableDesc defaultTd;
static {
@@ -1934,7 +1987,7 @@ public final class Utilities {
public static ContentSummary getInputSummary(Context ctx, MapWork work, PathFilter filter)
throws IOException {
PerfLogger perfLogger = PerfLogger.getPerfLogger();
- perfLogger.PerfLogBegin(LOG, PerfLogger.INPUT_SUMMARY);
+ perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.INPUT_SUMMARY);
long[] summary = {0, 0, 0};
@@ -2070,7 +2123,7 @@ public final class Utilities {
+ cs.getFileCount() + " directory count: " + cs.getDirectoryCount());
}
- perfLogger.PerfLogEnd(LOG, PerfLogger.INPUT_SUMMARY);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.INPUT_SUMMARY);
return new ContentSummary(summary[0], summary[1], summary[2]);
} finally {
HiveInterruptUtils.remove(interrup);
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java Fri Sep 20 00:32:55 2013
@@ -616,7 +616,6 @@ public class ExecDriver extends Task<Map
boolean noLog = false;
String files = null;
boolean localtask = false;
- String hadoopAuthToken = null;
try {
for (int i = 0; i < args.length; i++) {
if (args[i].equals("-plan")) {
@@ -629,9 +628,6 @@ public class ExecDriver extends Task<Map
files = args[++i];
} else if (args[i].equals("-localtask")) {
localtask = true;
- } else if (args[i].equals("-hadooptoken")) {
- //set with HS2 in secure mode with doAs
- hadoopAuthToken = args[++i];
}
}
} catch (IndexOutOfBoundsException e) {
@@ -653,6 +649,9 @@ public class ExecDriver extends Task<Map
if (files != null) {
conf.set("tmpfiles", files);
}
+
+ String hadoopAuthToken =
+ System.getenv(ShimLoader.getHadoopShims().getTokenFileLocEnvName());
if(hadoopAuthToken != null){
conf.set("mapreduce.job.credentials.binary", hadoopAuthToken);
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HadoopJobExecHelper.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HadoopJobExecHelper.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HadoopJobExecHelper.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HadoopJobExecHelper.java Fri Sep 20 00:32:55 2013
@@ -503,7 +503,7 @@ public class HadoopJobExecHelper {
}
} else {
console.printInfo("Execution completed successfully");
- console.printInfo("Mapred Local Task Succeeded . Convert the Join into MapJoin");
+ console.printInfo("MapredLocal task succeeded");
}
return exitVal;
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java Fri Sep 20 00:32:55 2013
@@ -236,7 +236,6 @@ public class MapredLocalTask extends Tas
// then additional params need to be set so that the command is run as
// intended user
SecureCmdDoAs secureDoAs = new SecureCmdDoAs(conf);
- cmdLine = secureDoAs.addArg(cmdLine);
secureDoAs.addEnv(variables);
}
@@ -271,7 +270,6 @@ public class MapredLocalTask extends Tas
}
} else {
LOG.info("Execution completed successfully");
- console.printInfo("Mapred Local Task Succeeded . Convert the Join into MapJoin");
}
return exitVal;
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizationContext.java Fri Sep 20 00:32:55 2013
@@ -79,7 +79,6 @@ import org.apache.hadoop.hive.ql.udf.UDF
import org.apache.hadoop.hive.ql.udf.UDFHour;
import org.apache.hadoop.hive.ql.udf.UDFLength;
import org.apache.hadoop.hive.ql.udf.UDFLike;
-import org.apache.hadoop.hive.ql.udf.UDFLower;
import org.apache.hadoop.hive.ql.udf.UDFMinute;
import org.apache.hadoop.hive.ql.udf.UDFMonth;
import org.apache.hadoop.hive.ql.udf.UDFOPDivide;
@@ -90,11 +89,11 @@ import org.apache.hadoop.hive.ql.udf.UDF
import org.apache.hadoop.hive.ql.udf.UDFOPPlus;
import org.apache.hadoop.hive.ql.udf.UDFOPPositive;
import org.apache.hadoop.hive.ql.udf.UDFSecond;
-import org.apache.hadoop.hive.ql.udf.UDFUpper;
import org.apache.hadoop.hive.ql.udf.UDFWeekOfYear;
import org.apache.hadoop.hive.ql.udf.UDFYear;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBridge;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFLower;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPAnd;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqual;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqualOrGreaterThan;
@@ -107,6 +106,7 @@ import org.apache.hadoop.hive.ql.udf.gen
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNull;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPOr;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFToUnixTimeStamp;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFUpper;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
import org.apache.hadoop.io.Text;
@@ -444,6 +444,10 @@ public class VectorizationContext {
return getVectorExpression((GenericUDFBridge) udf, childExpr);
} else if(udf instanceof GenericUDFToUnixTimeStamp) {
return getVectorExpression((GenericUDFToUnixTimeStamp) udf, childExpr);
+ } else if (udf instanceof GenericUDFLower) {
+ return getUnaryStringExpression("StringLower", "String", childExpr);
+ } else if (udf instanceof GenericUDFUpper) {
+ return getUnaryStringExpression("StringUpper", "String", childExpr);
}
throw new HiveException("Udf: "+udf.getClass().getSimpleName()+", is not supported");
}
@@ -497,10 +501,6 @@ public class VectorizationContext {
return getTimestampFieldExpression(cl.getSimpleName(), childExpr);
} else if (cl.equals(UDFLike.class)) {
return getLikeExpression(childExpr);
- } else if (cl.equals(UDFLower.class)) {
- return getUnaryStringExpression("StringLower", "String", childExpr);
- } else if (cl.equals(UDFUpper.class)) {
- return getUnaryStringExpression("StringUpper", "String", childExpr);
} else if (cl.equals(UDFLength.class)) {
return getUnaryStringExpression("StringLength", "Long", childExpr);
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringLower.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringLower.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringLower.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringLower.java Fri Sep 20 00:32:55 2013
@@ -18,8 +18,7 @@
package org.apache.hadoop.hive.ql.exec.vector.expressions;
-import org.apache.hadoop.hive.ql.udf.IUDFUnaryString;
-import org.apache.hadoop.hive.ql.udf.UDFLower;
+import org.apache.hadoop.io.Text;
/**
* Expression to convert a string to lower case.
@@ -29,7 +28,19 @@ public class StringLower extends StringU
private static final long serialVersionUID = 1L;
public StringLower(int colNum, int outputColumn) {
- super(colNum, outputColumn, (IUDFUnaryString) new UDFLower());
+ super(colNum, outputColumn, new IUDFUnaryString() {
+
+ private final Text t = new Text();
+
+ @Override
+ public Text evaluate(Text s) {
+ if (s == null) {
+ return null;
+ }
+ t.set(s.toString().toLowerCase());
+ return t;
+ }
+ });
}
public StringLower() {
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringUnaryUDF.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringUnaryUDF.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringUnaryUDF.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringUnaryUDF.java Fri Sep 20 00:32:55 2013
@@ -20,7 +20,6 @@ package org.apache.hadoop.hive.ql.exec.v
import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
-import org.apache.hadoop.hive.ql.udf.IUDFUnaryString;
import org.apache.hadoop.io.Text;
/**
@@ -29,6 +28,11 @@ import org.apache.hadoop.io.Text;
* the vector.
*/
public class StringUnaryUDF extends VectorExpression {
+
+ public interface IUDFUnaryString {
+ Text evaluate(Text s);
+ }
+
private static final long serialVersionUID = 1L;
private int colNum;
private int outputColumn;
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringUpper.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringUpper.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringUpper.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/StringUpper.java Fri Sep 20 00:32:55 2013
@@ -18,8 +18,7 @@
package org.apache.hadoop.hive.ql.exec.vector.expressions;
-import org.apache.hadoop.hive.ql.udf.IUDFUnaryString;
-import org.apache.hadoop.hive.ql.udf.UDFUpper;
+import org.apache.hadoop.io.Text;
/**
* Expression to convert a string to lower case.
@@ -29,7 +28,19 @@ public class StringUpper extends StringU
private static final long serialVersionUID = 1L;
public StringUpper(int colNum, int outputColumn) {
- super(colNum, outputColumn, (IUDFUnaryString) new UDFUpper());
+ super(colNum, outputColumn, new IUDFUnaryString() {
+
+ Text t = new Text();
+
+ @Override
+ public Text evaluate(Text s) {
+ if (s == null) {
+ return null;
+ }
+ t.set(s.toString().toUpperCase());
+ return t;
+ }
+ });
}
public StringUpper() {
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/CombineHiveInputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/CombineHiveInputFormat.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/CombineHiveInputFormat.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/CombineHiveInputFormat.java Fri Sep 20 00:32:55 2013
@@ -68,8 +68,8 @@ import org.apache.hadoop.mapred.TextInpu
public class CombineHiveInputFormat<K extends WritableComparable, V extends Writable>
extends HiveInputFormat<K, V> {
- public static final Log LOG = LogFactory
- .getLog("org.apache.hadoop.hive.ql.io.CombineHiveInputFormat");
+ private static final String CLASS_NAME = CombineHiveInputFormat.class.getName();
+ public static final Log LOG = LogFactory.getLog(CLASS_NAME);
/**
* CombineHiveInputSplit encapsulates an InputSplit with its corresponding
@@ -264,7 +264,7 @@ public class CombineHiveInputFormat<K ex
@Override
public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
PerfLogger perfLogger = PerfLogger.getPerfLogger();
- perfLogger.PerfLogBegin(LOG, PerfLogger.GET_SPLITS);
+ perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.GET_SPLITS);
init(job);
Map<String, ArrayList<String>> pathToAliases = mrwork.getPathToAliases();
Map<String, Operator<? extends OperatorDesc>> aliasToWork =
@@ -275,7 +275,7 @@ public class CombineHiveInputFormat<K ex
InputSplit[] splits = null;
if (combine == null) {
splits = super.getSplits(job, numSplits);
- perfLogger.PerfLogEnd(LOG, PerfLogger.GET_SPLITS);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
return splits;
}
@@ -328,7 +328,7 @@ public class CombineHiveInputFormat<K ex
dirs.offer(path);
} else if ((new CompressionCodecFactory(job)).getCodec(path) != null) {
splits = super.getSplits(job, numSplits);
- perfLogger.PerfLogEnd(LOG, PerfLogger.GET_SPLITS);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
return splits;
}
@@ -341,7 +341,7 @@ public class CombineHiveInputFormat<K ex
} else if ((new CompressionCodecFactory(job)).getCodec(
fStatus[idx].getPath()) != null) {
splits = super.getSplits(job, numSplits);
- perfLogger.PerfLogEnd(LOG, PerfLogger.GET_SPLITS);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
return splits;
}
}
@@ -351,7 +351,7 @@ public class CombineHiveInputFormat<K ex
if (inputFormat instanceof SymlinkTextInputFormat) {
splits = super.getSplits(job, numSplits);
- perfLogger.PerfLogEnd(LOG, PerfLogger.GET_SPLITS);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
return splits;
}
@@ -422,7 +422,7 @@ public class CombineHiveInputFormat<K ex
}
LOG.info("number of splits " + result.size());
- perfLogger.PerfLogEnd(LOG, PerfLogger.GET_SPLITS);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
return result.toArray(new CombineHiveInputSplit[result.size()]);
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/HiveFileFormatUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/HiveFileFormatUtils.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/HiveFileFormatUtils.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/HiveFileFormatUtils.java Fri Sep 20 00:32:55 2013
@@ -19,6 +19,7 @@
package org.apache.hadoop.hive.ql.io;
import java.io.IOException;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -34,6 +35,7 @@ import org.apache.hadoop.hive.conf.HiveC
import org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.io.HivePassThroughOutputFormat;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
@@ -51,6 +53,7 @@ import org.apache.hadoop.mapred.Sequence
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.util.Shell;
+import org.apache.hadoop.util.ReflectionUtils;
/**
* An util class for various Hive file format tasks.
@@ -70,6 +73,8 @@ public final class HiveFileFormatUtils {
SequenceFileOutputFormat.class, HiveSequenceFileOutputFormat.class);
}
+ static String realoutputFormat;
+
@SuppressWarnings("unchecked")
private static Map<Class<? extends OutputFormat>, Class<? extends HiveOutputFormat>>
outputFormatSubstituteMap;
@@ -93,16 +98,45 @@ public final class HiveFileFormatUtils {
*/
@SuppressWarnings("unchecked")
public static synchronized Class<? extends HiveOutputFormat> getOutputFormatSubstitute(
- Class<?> origin) {
+ Class<?> origin, boolean storagehandlerflag) {
if (HiveOutputFormat.class.isAssignableFrom(origin)) {
return (Class<? extends HiveOutputFormat>) origin;
}
Class<? extends HiveOutputFormat> result = outputFormatSubstituteMap
.get(origin);
+ //register this output format into the map for the first time
+ if ((storagehandlerflag == true) && (result == null)) {
+ HiveFileFormatUtils.setRealOutputFormatClassName(origin.getName());
+ result = HivePassThroughOutputFormat.class;
+ HiveFileFormatUtils.registerOutputFormatSubstitute((Class<? extends OutputFormat>) origin,HivePassThroughOutputFormat.class);
+ }
return result;
}
/**
+ * get a RealOutputFormatClassName corresponding to the HivePassThroughOutputFormat
+ */
+ @SuppressWarnings("unchecked")
+ public static String getRealOutputFormatClassName()
+ {
+ return realoutputFormat;
+ }
+
+ /**
+ * set a RealOutputFormatClassName corresponding to the HivePassThroughOutputFormat
+ */
+ public static void setRealOutputFormatClassName(
+ String destination) {
+ if (destination != null){
+ realoutputFormat = destination;
+ }
+ else {
+ return;
+ }
+ }
+
+
+ /**
* get the final output path of a given FileOutputFormat.
*
* @param parent
@@ -215,9 +249,21 @@ public final class HiveFileFormatUtils {
public static RecordWriter getHiveRecordWriter(JobConf jc,
TableDesc tableInfo, Class<? extends Writable> outputClass,
FileSinkDesc conf, Path outPath, Reporter reporter) throws HiveException {
+ boolean storagehandlerofhivepassthru = false;
+ HiveOutputFormat<?, ?> hiveOutputFormat;
try {
- HiveOutputFormat<?, ?> hiveOutputFormat = tableInfo
- .getOutputFileFormatClass().newInstance();
+ if (tableInfo.getJobProperties() != null) {
+ if (tableInfo.getJobProperties().get(HivePassThroughOutputFormat.HIVE_PASSTHROUGH_STORAGEHANDLER_OF_JOBCONFKEY) != null) {
+ jc.set(HivePassThroughOutputFormat.HIVE_PASSTHROUGH_STORAGEHANDLER_OF_JOBCONFKEY,tableInfo.getJobProperties().get(HivePassThroughOutputFormat.HIVE_PASSTHROUGH_STORAGEHANDLER_OF_JOBCONFKEY));
+ storagehandlerofhivepassthru = true;
+ }
+ }
+ if (storagehandlerofhivepassthru) {
+ hiveOutputFormat = ReflectionUtils.newInstance(tableInfo.getOutputFileFormatClass(),jc);
+ }
+ else {
+ hiveOutputFormat = tableInfo.getOutputFileFormatClass().newInstance();
+ }
boolean isCompressed = conf.getCompressed();
JobConf jc_output = jc;
if (isCompressed) {
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java Fri Sep 20 00:32:55 2013
@@ -65,8 +65,8 @@ import org.apache.hadoop.util.Reflection
public class HiveInputFormat<K extends WritableComparable, V extends Writable>
implements InputFormat<K, V>, JobConfigurable {
- public static final Log LOG = LogFactory
- .getLog("org.apache.hadoop.hive.ql.io.HiveInputFormat");
+ public static final String CLASS_NAME = HiveInputFormat.class.getName();
+ public static final Log LOG = LogFactory.getLog(CLASS_NAME);
/**
* HiveInputSplit encapsulates an InputSplit with its corresponding
@@ -259,7 +259,7 @@ public class HiveInputFormat<K extends W
public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
PerfLogger perfLogger = PerfLogger.getPerfLogger();
- perfLogger.PerfLogBegin(LOG, PerfLogger.GET_SPLITS);
+ perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.GET_SPLITS);
init(job);
Path[] dirs = FileInputFormat.getInputPaths(job);
@@ -298,7 +298,7 @@ public class HiveInputFormat<K extends W
}
LOG.info("number of splits " + result.size());
- perfLogger.PerfLogEnd(LOG, PerfLogger.GET_SPLITS);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
return result.toArray(new HiveInputSplit[result.size()]);
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ColumnStatisticsImpl.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ColumnStatisticsImpl.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ColumnStatisticsImpl.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ColumnStatisticsImpl.java Fri Sep 20 00:32:55 2013
@@ -714,6 +714,7 @@ class ColumnStatisticsImpl implements Co
case DOUBLE:
return new DoubleStatisticsImpl();
case STRING:
+ case VARCHAR:
return new StringStatisticsImpl();
case DECIMAL:
return new DecimalStatisticsImpl();
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java Fri Sep 20 00:32:55 2013
@@ -26,37 +26,62 @@ import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedInputFormatInterface;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.hadoop.hive.ql.io.InputFormatChecker;
-import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.ql.plan.TableScanDesc;
import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
-import org.apache.hadoop.hive.serde2.SerDeException;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.shims.HadoopShims;
+import org.apache.hadoop.hive.shims.ShimLoader;
+import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileSplit;
+import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.InvalidInputException;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.util.StringUtils;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
/**
* A MapReduce/Hive input format for ORC files.
*/
-public class OrcInputFormat extends FileInputFormat<NullWritable, OrcStruct>
- implements InputFormatChecker, VectorizedInputFormatInterface {
+public class OrcInputFormat implements InputFormat<NullWritable, OrcStruct>,
+ InputFormatChecker, VectorizedInputFormatInterface {
VectorizedOrcInputFormat voif = new VectorizedOrcInputFormat();
private static final Log LOG = LogFactory.getLog(OrcInputFormat.class);
+ static final String MIN_SPLIT_SIZE = "mapred.min.split.size";
+ static final String MAX_SPLIT_SIZE = "mapred.max.split.size";
+ private static final long DEFAULT_MIN_SPLIT_SIZE = 16 * 1024 * 1024;
+ private static final long DEFAULT_MAX_SPLIT_SIZE = 256 * 1024 * 1024;
+
+ /**
+ * When picking the hosts for a split that crosses block boundaries,
+ * any drop any host that has fewer than MIN_INCLUDED_LOCATION of the
+ * number of bytes available on the host with the most.
+ * If host1 has 10MB of the split, host2 has 20MB, and host3 has 18MB the
+ * split will contain host2 (100% of host2) and host3 (90% of host2). Host1
+ * with 50% will be dropped.
+ */
+ private static final double MIN_INCLUDED_LOCATION = 0.80;
private static class OrcRecordReader
implements RecordReader<NullWritable, OrcStruct> {
@@ -142,10 +167,12 @@ public class OrcInputFormat extends Fil
}
}
- public OrcInputFormat() {
- // just set a really small lower bound
- setMinSplitSize(16 * 1024);
- }
+ private static final PathFilter hiddenFileFilter = new PathFilter(){
+ public boolean accept(Path p){
+ String name = p.getName();
+ return !name.startsWith("_") && !name.startsWith(".");
+ }
+ };
/**
* Recurse down into a type subtree turning on all of the sub-columns.
@@ -153,7 +180,7 @@ public class OrcInputFormat extends Fil
* @param result the global view of columns that should be included
* @param typeId the root of tree to enable
*/
- private static void includeColumnRecursive(List<OrcProto.Type> types,
+ static void includeColumnRecursive(List<OrcProto.Type> types,
boolean[] result,
int typeId) {
result[typeId] = true;
@@ -170,7 +197,7 @@ public class OrcInputFormat extends Fil
* @param conf the configuration
* @return true for each column that should be included
*/
- private static boolean[] findIncludedColumns(List<OrcProto.Type> types,
+ static boolean[] findIncludedColumns(List<OrcProto.Type> types,
Configuration conf) {
String includedStr =
conf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR);
@@ -244,4 +271,324 @@ public class OrcInputFormat extends Fil
}
return false;
}
+
+ /**
+ * Get the list of input {@link Path}s for the map-reduce job.
+ *
+ * @param conf The configuration of the job
+ * @return the list of input {@link Path}s for the map-reduce job.
+ */
+ static Path[] getInputPaths(JobConf conf) throws IOException {
+ String dirs = conf.get("mapred.input.dir");
+ if (dirs == null) {
+ throw new IOException("Configuration mapred.input.dir is not defined.");
+ }
+ String [] list = StringUtils.split(dirs);
+ Path[] result = new Path[list.length];
+ for (int i = 0; i < list.length; i++) {
+ result[i] = new Path(StringUtils.unEscapeString(list[i]));
+ }
+ return result;
+ }
+
+ /**
+ * The global information about the split generation that we pass around to
+ * the different worker threads.
+ */
+ static class Context {
+ private final ExecutorService threadPool = Executors.newFixedThreadPool(10);
+ private final List<FileSplit> splits = new ArrayList<FileSplit>(10000);
+ private final List<Throwable> errors = new ArrayList<Throwable>();
+ private final HadoopShims shims = ShimLoader.getHadoopShims();
+ private final Configuration conf;
+ private final long maxSize;
+ private final long minSize;
+
+ /**
+ * A count of the number of threads that may create more work for the
+ * thread pool.
+ */
+ private int schedulers = 0;
+
+ Context(Configuration conf) {
+ this.conf = conf;
+ minSize = conf.getLong(MIN_SPLIT_SIZE, DEFAULT_MIN_SPLIT_SIZE);
+ maxSize = conf.getLong(MAX_SPLIT_SIZE, DEFAULT_MAX_SPLIT_SIZE);
+ }
+
+ int getSchedulers() {
+ return schedulers;
+ }
+
+ /**
+ * Get the Nth split.
+ * @param index if index >= 0, count from the front, otherwise count from
+ * the back.
+ * @result the Nth file split
+ */
+ FileSplit getResult(int index) {
+ if (index >= 0) {
+ return splits.get(index);
+ } else {
+ return splits.get(splits.size() + index);
+ }
+ }
+
+ List<Throwable> getErrors() {
+ return errors;
+ }
+
+ /**
+ * Add a unit of work.
+ * @param runnable the object to run
+ */
+ synchronized void schedule(Runnable runnable) {
+ if (runnable instanceof FileGenerator) {
+ schedulers += 1;
+ }
+ threadPool.execute(runnable);
+ }
+
+ /**
+ * Mark a worker that may generate more work as done.
+ */
+ synchronized void decrementSchedulers() {
+ schedulers -= 1;
+ if (schedulers == 0) {
+ notify();
+ }
+ }
+
+ /**
+ * Wait until all of the tasks are done. It waits until all of the
+ * threads that may create more work are done and then shuts down the
+ * thread pool and waits for the final threads to finish.
+ */
+ synchronized void waitForTasks() {
+ try {
+ while (schedulers != 0) {
+ wait();
+ }
+ threadPool.shutdown();
+ threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
+ } catch (InterruptedException ie) {
+ throw new IllegalStateException("interrupted", ie);
+ }
+ }
+ }
+
+ /**
+ * Given a directory, get the list of files and blocks in those files.
+ * A thread is used for each directory.
+ */
+ static final class FileGenerator implements Runnable {
+ private final Context context;
+ private final FileSystem fs;
+ private final Path dir;
+
+ FileGenerator(Context context, FileSystem fs, Path dir) {
+ this.context = context;
+ this.fs = fs;
+ this.dir = dir;
+ }
+
+ /**
+ * For each path, get the list of files and blocks that they consist of.
+ */
+ @Override
+ public void run() {
+ try {
+ Iterator<FileStatus> itr = context.shims.listLocatedStatus(fs, dir,
+ hiddenFileFilter);
+ while (itr.hasNext()) {
+ FileStatus file = itr.next();
+ if (!file.isDir()) {
+ context.schedule(new SplitGenerator(context, fs, file));
+ }
+ }
+ // mark the fact that we are done
+ context.decrementSchedulers();
+ } catch (Throwable th) {
+ context.decrementSchedulers();
+ synchronized (context.errors) {
+ context.errors.add(th);
+ }
+ }
+ }
+ }
+
+ /**
+ * Split the stripes of a given file into input splits.
+ * A thread is used for each file.
+ */
+ static final class SplitGenerator implements Runnable {
+ private final Context context;
+ private final FileSystem fs;
+ private final FileStatus file;
+ private final long blockSize;
+ private final BlockLocation[] locations;
+
+ SplitGenerator(Context context, FileSystem fs,
+ FileStatus file) throws IOException {
+ this.context = context;
+ this.fs = fs;
+ this.file = file;
+ this.blockSize = file.getBlockSize();
+ locations = context.shims.getLocations(fs, file);
+ }
+
+ Path getPath() {
+ return file.getPath();
+ }
+
+ @Override
+ public String toString() {
+ return "splitter(" + file.getPath() + ")";
+ }
+
+ /**
+ * Compute the number of bytes that overlap between the two ranges.
+ * @param offset1 start of range1
+ * @param length1 length of range1
+ * @param offset2 start of range2
+ * @param length2 length of range2
+ * @return the number of bytes in the overlap range
+ */
+ static long getOverlap(long offset1, long length1,
+ long offset2, long length2) {
+ long end1 = offset1 + length1;
+ long end2 = offset2 + length2;
+ if (end2 <= offset1 || end1 <= offset2) {
+ return 0;
+ } else {
+ return Math.min(end1, end2) - Math.max(offset1, offset2);
+ }
+ }
+
+ /**
+ * Create an input split over the given range of bytes. The location of the
+ * split is based on where the majority of the byte are coming from. ORC
+ * files are unlikely to have splits that cross between blocks because they
+ * are written with large block sizes.
+ * @param offset the start of the split
+ * @param length the length of the split
+ * @throws IOException
+ */
+ void createSplit(long offset, long length) throws IOException {
+ String[] hosts;
+ if ((offset % blockSize) + length <= blockSize) {
+ // handle the single block case
+ hosts = locations[(int) (offset / blockSize)].getHosts();
+ } else {
+ // Calculate the number of bytes in the split that are local to each
+ // host.
+ Map<String, LongWritable> sizes = new HashMap<String, LongWritable>();
+ long maxSize = 0;
+ for(BlockLocation block: locations) {
+ long overlap = getOverlap(offset, length, block.getOffset(),
+ block.getLength());
+ if (overlap > 0) {
+ for(String host: block.getHosts()) {
+ LongWritable val = sizes.get(host);
+ if (val == null) {
+ val = new LongWritable();
+ sizes.put(host, val);
+ }
+ val.set(val.get() + overlap);
+ maxSize = Math.max(maxSize, val.get());
+ }
+ }
+ }
+ // filter the list of locations to those that have at least 80% of the
+ // max
+ long threshold = (long) (maxSize * MIN_INCLUDED_LOCATION);
+ List<String> hostList = new ArrayList<String>();
+ // build the locations in a predictable order to simplify testing
+ for(BlockLocation block: locations) {
+ for(String host: block.getHosts()) {
+ if (sizes.containsKey(host)) {
+ if (sizes.get(host).get() >= threshold) {
+ hostList.add(host);
+ }
+ sizes.remove(host);
+ }
+ }
+ }
+ hosts = new String[hostList.size()];
+ hostList.toArray(hosts);
+ }
+ synchronized (context.splits) {
+ context.splits.add(new FileSplit(file.getPath(), offset, length,
+ hosts));
+ }
+ }
+
+ /**
+ * Divide the adjacent stripes in the file into input splits based on the
+ * block size and the configured minimum and maximum sizes.
+ */
+ @Override
+ public void run() {
+ try {
+ Reader orcReader = OrcFile.createReader(fs, file.getPath());
+ long currentOffset = -1;
+ long currentLength = 0;
+ for(StripeInformation stripe: orcReader.getStripes()) {
+ // if we are working on a stripe, over the min stripe size, and
+ // crossed a block boundary, cut the input split here.
+ if (currentOffset != -1 && currentLength > context.minSize &&
+ (currentOffset / blockSize != stripe.getOffset() / blockSize)) {
+ createSplit(currentOffset, currentLength);
+ currentOffset = -1;
+ }
+ // if we aren't building a split, start a new one.
+ if (currentOffset == -1) {
+ currentOffset = stripe.getOffset();
+ currentLength = stripe.getLength();
+ } else {
+ currentLength += stripe.getLength();
+ }
+ if (currentLength >= context.maxSize) {
+ createSplit(currentOffset, currentLength);
+ currentOffset = -1;
+ }
+ }
+ if (currentOffset != -1) {
+ createSplit(currentOffset, currentLength);
+ }
+ } catch (Throwable th) {
+ synchronized (context.errors) {
+ context.errors.add(th);
+ }
+ }
+ }
+ }
+
+ @Override
+ public InputSplit[] getSplits(JobConf job,
+ int numSplits) throws IOException {
+ // use threads to resolve directories into splits
+ Context context = new Context(job);
+ for(Path dir: getInputPaths(job)) {
+ FileSystem fs = dir.getFileSystem(job);
+ context.schedule(new FileGenerator(context, fs, dir));
+ }
+ context.waitForTasks();
+ // deal with exceptions
+ if (!context.errors.isEmpty()) {
+ List<IOException> errors =
+ new ArrayList<IOException>(context.errors.size());
+ for(Throwable th: context.errors) {
+ if (th instanceof IOException) {
+ errors.add((IOException) th);
+ } else {
+ throw new IOException("serious problem", th);
+ }
+ }
+ throw new InvalidInputException(errors);
+ }
+ InputSplit[] result = new InputSplit[context.splits.size()];
+ context.splits.toArray(result);
+ return result;
+ }
}
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcStruct.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcStruct.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcStruct.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcStruct.java Fri Sep 20 00:32:55 2013
@@ -33,13 +33,17 @@ import org.apache.hadoop.hive.serde2.obj
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.PrimitiveObjectInspector.PrimitiveCategory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.typeinfo.ListTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.MapTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.ParameterizedPrimitiveTypeUtils;
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.UnionTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.VarcharTypeParams;
import org.apache.hadoop.io.Writable;
final class OrcStruct implements Writable {
@@ -163,8 +167,11 @@ final class OrcStruct implements Writabl
}
static class OrcStructInspector extends SettableStructObjectInspector {
- private final List<StructField> fields;
+ private List<StructField> fields;
+ protected OrcStructInspector() {
+ super();
+ }
OrcStructInspector(StructTypeInfo info) {
ArrayList<String> fieldNames = info.getAllStructFieldNames();
ArrayList<TypeInfo> fieldTypes = info.getAllStructFieldTypeInfos();
@@ -283,9 +290,12 @@ final class OrcStruct implements Writabl
static class OrcMapObjectInspector
implements MapObjectInspector, SettableMapObjectInspector {
- private final ObjectInspector key;
- private final ObjectInspector value;
+ private ObjectInspector key;
+ private ObjectInspector value;
+ private OrcMapObjectInspector() {
+ super();
+ }
OrcMapObjectInspector(MapTypeInfo info) {
key = createObjectInspector(info.getMapKeyTypeInfo());
value = createObjectInspector(info.getMapValueTypeInfo());
@@ -371,8 +381,11 @@ final class OrcStruct implements Writabl
static class OrcListObjectInspector
implements ListObjectInspector, SettableListObjectInspector {
- private final ObjectInspector child;
+ private ObjectInspector child;
+ private OrcListObjectInspector() {
+ super();
+ }
OrcListObjectInspector(ListTypeInfo info) {
child = createObjectInspector(info.getListElementTypeInfo());
}
@@ -473,6 +486,15 @@ final class OrcStruct implements Writabl
return PrimitiveObjectInspectorFactory.writableBinaryObjectInspector;
case STRING:
return PrimitiveObjectInspectorFactory.writableStringObjectInspector;
+ case VARCHAR:
+ // For varchar we need to retrieve the string length from the TypeInfo.
+ VarcharTypeParams varcharParams = (VarcharTypeParams)
+ ParameterizedPrimitiveTypeUtils.getTypeParamsFromTypeInfo(info);
+ if (varcharParams == null) {
+ throw new IllegalArgumentException("varchar type used without type params");
+ }
+ return PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(
+ (PrimitiveTypeInfo) info);
case TIMESTAMP:
return PrimitiveObjectInspectorFactory.javaTimestampObjectInspector;
case DATE:
@@ -519,6 +541,16 @@ final class OrcStruct implements Writabl
return PrimitiveObjectInspectorFactory.writableBinaryObjectInspector;
case STRING:
return PrimitiveObjectInspectorFactory.writableStringObjectInspector;
+ case VARCHAR:
+ if (!type.hasMaximumLength()) {
+ throw new UnsupportedOperationException(
+ "Illegal use of varchar type without length in ORC type definition.");
+ }
+ VarcharTypeParams varcharParams = new VarcharTypeParams();
+ varcharParams.setLength(type.getMaximumLength());
+ return PrimitiveObjectInspectorFactory.getPrimitiveWritableObjectInspector(
+ PrimitiveObjectInspectorUtils.getTypeEntryFromTypeSpecs(
+ PrimitiveCategory.VARCHAR, varcharParams));
case TIMESTAMP:
return PrimitiveObjectInspectorFactory.javaTimestampObjectInspector;
case DATE:
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcUnion.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcUnion.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcUnion.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcUnion.java Fri Sep 20 00:32:55 2013
@@ -79,8 +79,11 @@ final class OrcUnion implements UnionObj
}
static class OrcUnionObjectInspector implements UnionObjectInspector {
- private final List<ObjectInspector> children;
+ private List<ObjectInspector> children;
+ protected OrcUnionObjectInspector() {
+ super();
+ }
OrcUnionObjectInspector(int columnId,
List<OrcProto.Type> types) {
OrcProto.Type type = types.get(columnId);
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ReaderImpl.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ReaderImpl.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ReaderImpl.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ReaderImpl.java Fri Sep 20 00:32:55 2013
@@ -27,6 +27,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.util.StringUtils;
import java.io.IOException;
import java.io.InputStream;
@@ -63,6 +64,11 @@ final class ReaderImpl implements Reader
}
@Override
+ public long getLength() {
+ return stripe.getDataLength() + getIndexLength() + getFooterLength();
+ }
+
+ @Override
public long getDataLength() {
return stripe.getDataLength();
}
@@ -269,7 +275,7 @@ final class ReaderImpl implements Reader
ByteBuffer buffer = ByteBuffer.allocate(readSize);
file.readFully(buffer.array(), buffer.arrayOffset() + buffer.position(),
buffer.remaining());
- int psLen = buffer.get(readSize - 1);
+ int psLen = buffer.get(readSize - 1) & 0xff;
ensureOrcFooter(file, path, psLen, buffer);
int psOffset = readSize - 1 - psLen;
CodedInputStream in = CodedInputStream.newInstance(buffer.array(),
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java Fri Sep 20 00:32:55 2013
@@ -44,6 +44,7 @@ import org.apache.hadoop.hive.ql.io.sarg
import org.apache.hadoop.hive.serde2.io.ByteWritable;
import org.apache.hadoop.hive.serde2.io.DateWritable;
import org.apache.hadoop.hive.serde2.io.DoubleWritable;
+import org.apache.hadoop.hive.serde2.io.HiveVarcharWritable;
import org.apache.hadoop.hive.serde2.io.ShortWritable;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.BytesWritable;
@@ -1460,6 +1461,34 @@ class RecordReaderImpl implements Record
}
}
+ private static class VarcharTreeReader extends StringTreeReader {
+ int maxLength;
+
+ VarcharTreeReader(Path path, int columnId, int maxLength) {
+ super(path, columnId);
+ this.maxLength = maxLength;
+ }
+
+ @Override
+ Object next(Object previous) throws IOException {
+ HiveVarcharWritable result = null;
+ if (previous == null) {
+ result = new HiveVarcharWritable();
+ } else {
+ result = (HiveVarcharWritable) previous;
+ }
+ // Use the string reader implementation to populate the internal Text value
+ Object textVal = super.next(result.getTextValue());
+ if (textVal == null) {
+ return null;
+ }
+ // result should now hold the value that was read in.
+ // enforce varchar length
+ result.enforceMaxLength(maxLength);
+ return result;
+ }
+ }
+
private static class StructTreeReader extends TreeReader {
private final TreeReader[] fields;
private final String[] fieldNames;
@@ -1849,6 +1878,11 @@ class RecordReaderImpl implements Record
return new LongTreeReader(path, columnId);
case STRING:
return new StringTreeReader(path, columnId);
+ case VARCHAR:
+ if (!type.hasMaximumLength()) {
+ throw new IllegalArgumentException("ORC varchar type has no length specified");
+ }
+ return new VarcharTreeReader(path, columnId, type.getMaximumLength());
case BINARY:
return new BinaryTreeReader(path, columnId);
case TIMESTAMP:
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/StripeInformation.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/StripeInformation.java?rev=1524874&r1=1524873&r2=1524874&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/StripeInformation.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/StripeInformation.java Fri Sep 20 00:32:55 2013
@@ -28,6 +28,12 @@ public interface StripeInformation {
long getOffset();
/**
+ * Get the total length of the stripe in bytes.
+ * @return the number of bytes in the stripe
+ */
+ long getLength();
+
+ /**
* Get the length of the stripe's indexes.
* @return the number of bytes in the index
*/