You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by jp...@apache.org on 2015/03/25 19:44:16 UTC
svn commit: r1669206 [2/2] - in /hive/branches/cbo: metastore/bin/ ql/
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/
ql/src/java/org/apache/hadoop/hive/ql/parse/ ql/src/java...
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java?rev=1669206&r1=1669205&r2=1669206&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java Wed Mar 25 18:44:16 2015
@@ -61,6 +61,7 @@ import org.apache.hadoop.hive.ql.optimiz
import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil.JoinLeafPredicateInfo;
import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil.JoinPredicateInfo;
import org.apache.hadoop.hive.ql.optimizer.calcite.RelOptHiveTable;
+import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveAggregate;
import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveFilter;
import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveJoin;
import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveProject;
@@ -105,7 +106,7 @@ public class HiveOpConverter {
NO_SKEW_NO_MAP_SIDE_AGG, // Corresponds to SemAnalyzer genGroupByPlan1MR
SKEW_NO_MAP_SIDE_AGG, // Corresponds to SemAnalyzer genGroupByPlan2MR
NO_SKEW_MAP_SIDE_AGG, // Corresponds to SemAnalyzer
- // genGroupByPlanMapAggrNoSkew
+ // genGroupByPlanMapAggrNoSkew
SKEW_MAP_SIDE_AGG // Corresponds to SemAnalyzer genGroupByPlanMapAggr2MR
};
@@ -114,45 +115,26 @@ public class HiveOpConverter {
private final HiveConf hiveConf;
private final UnparseTranslator unparseTranslator;
private final Map<String, Operator<? extends OperatorDesc>> topOps;
- private final HIVEAGGOPMODE aggMode;
private final boolean strictMode;
private int reduceSinkTagGenerator;
- public static HIVEAGGOPMODE getAggOPMode(HiveConf hc) {
- HIVEAGGOPMODE aggOpMode = HIVEAGGOPMODE.NO_SKEW_NO_MAP_SIDE_AGG;
-
- if (hc.getBoolVar(HiveConf.ConfVars.HIVEMAPSIDEAGGREGATE)) {
- if (!hc.getBoolVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)) {
- aggOpMode = HIVEAGGOPMODE.NO_SKEW_MAP_SIDE_AGG;
- } else {
- aggOpMode = HIVEAGGOPMODE.SKEW_MAP_SIDE_AGG;
- }
- } else if (hc.getBoolVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)) {
- aggOpMode = HIVEAGGOPMODE.SKEW_NO_MAP_SIDE_AGG;
- }
-
- return aggOpMode;
- }
-
- public HiveOpConverter(SemanticAnalyzer semanticAnalyzer,
- HiveConf hiveConf, UnparseTranslator unparseTranslator,
- Map<String, Operator<? extends OperatorDesc>> topOps,
- HIVEAGGOPMODE aggMode, boolean strictMode) {
+ public HiveOpConverter(SemanticAnalyzer semanticAnalyzer, HiveConf hiveConf,
+ UnparseTranslator unparseTranslator, Map<String, Operator<? extends OperatorDesc>> topOps,
+ boolean strictMode) {
this.semanticAnalyzer = semanticAnalyzer;
this.hiveConf = hiveConf;
this.unparseTranslator = unparseTranslator;
this.topOps = topOps;
- this.aggMode = aggMode;
this.strictMode = strictMode;
this.reduceSinkTagGenerator = 0;
}
- private class OpAttr {
- private final String tabAlias;
+ static class OpAttr {
+ final String tabAlias;
ImmutableList<Operator> inputs;
ImmutableMap<Integer, VirtualColumn> vcolMap;
- private OpAttr(String tabAlias, Map<Integer, VirtualColumn> vcolMap, Operator... inputs) {
+ OpAttr(String tabAlias, Map<Integer, VirtualColumn> vcolMap, Operator... inputs) {
this.tabAlias = tabAlias;
this.vcolMap = ImmutableMap.copyOf(vcolMap);
this.inputs = ImmutableList.copyOf(inputs);
@@ -178,7 +160,7 @@ public class HiveOpConverter {
} else if (rn instanceof SemiJoin) {
SemiJoin sj = (SemiJoin) rn;
HiveJoin hj = HiveJoin.getJoin(sj.getCluster(), sj.getLeft(), sj.getRight(),
- sj.getCondition(), sj.getJoinType(), true);
+ sj.getCondition(), sj.getJoinType(), true);
return visit(hj);
} else if (rn instanceof HiveFilter) {
return visit((HiveFilter) rn);
@@ -188,9 +170,11 @@ public class HiveOpConverter {
return visit((HiveUnion) rn);
} else if (rn instanceof LogicalExchange) {
return visit((LogicalExchange) rn);
+ } else if (rn instanceof HiveAggregate) {
+ return visit((HiveAggregate) rn);
}
LOG.error(rn.getClass().getCanonicalName() + "operator translation not supported"
- + " yet in return path.");
+ + " yet in return path.");
return null;
}
@@ -203,8 +187,8 @@ public class HiveOpConverter {
OpAttr visit(HiveTableScan scanRel) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Translating operator rel#" + scanRel.getId() + ":" + scanRel.getRelTypeName() +
- " with row type: [" + scanRel.getRowType() + "]");
+ LOG.debug("Translating operator rel#" + scanRel.getId() + ":" + scanRel.getRelTypeName()
+ + " with row type: [" + scanRel.getRowType() + "]");
}
RelOptHiveTable ht = (RelOptHiveTable) scanRel.getTable();
@@ -264,24 +248,23 @@ public class HiveOpConverter {
OpAttr inputOpAf = dispatch(projectRel.getInput());
if (LOG.isDebugEnabled()) {
- LOG.debug("Translating operator rel#" + projectRel.getId() + ":" + projectRel.getRelTypeName() +
- " with row type: [" + projectRel.getRowType() + "]");
+ LOG.debug("Translating operator rel#" + projectRel.getId() + ":"
+ + projectRel.getRelTypeName() + " with row type: [" + projectRel.getRowType() + "]");
}
WindowingSpec windowingSpec = new WindowingSpec();
List<ExprNodeDesc> exprCols = new ArrayList<ExprNodeDesc>();
- for (int pos=0; pos<projectRel.getChildExps().size(); pos++) {
- ExprNodeConverter converter = new ExprNodeConverter(inputOpAf.tabAlias,
- projectRel.getRowType().getFieldNames().get(pos),
- projectRel.getInput().getRowType(), projectRel.getRowType(), false);
- exprCols.add((ExprNodeDesc) projectRel.getChildExps().get(pos).
- accept(converter));
+ for (int pos = 0; pos < projectRel.getChildExps().size(); pos++) {
+ ExprNodeConverter converter = new ExprNodeConverter(inputOpAf.tabAlias, projectRel
+ .getRowType().getFieldNames().get(pos), projectRel.getInput().getRowType(),
+ projectRel.getRowType(), false, projectRel.getCluster().getTypeFactory());
+ exprCols.add((ExprNodeDesc) projectRel.getChildExps().get(pos).accept(converter));
if (converter.getWindowFunctionSpec() != null) {
windowingSpec.addWindowFunction(converter.getWindowFunctionSpec());
}
}
- if (windowingSpec.getWindowExpressions() != null &&
- !windowingSpec.getWindowExpressions().isEmpty()) {
+ if (windowingSpec.getWindowExpressions() != null
+ && !windowingSpec.getWindowExpressions().isEmpty()) {
inputOpAf = genPTF(inputOpAf, windowingSpec);
}
// TODO: is this a safe assumption (name collision, external names...)
@@ -302,16 +285,15 @@ public class HiveOpConverter {
OpAttr visit(HiveJoin joinRel) throws SemanticException {
// 1. Convert inputs
OpAttr[] inputs = new OpAttr[joinRel.getInputs().size()];
- List<Operator<?>> children = new ArrayList<Operator<?>>(
- joinRel.getInputs().size());
- for (int i=0; i<inputs.length; i++) {
+ List<Operator<?>> children = new ArrayList<Operator<?>>(joinRel.getInputs().size());
+ for (int i = 0; i < inputs.length; i++) {
inputs[i] = dispatch(joinRel.getInput(i));
children.add(inputs[i].inputs.get(0));
}
if (LOG.isDebugEnabled()) {
- LOG.debug("Translating operator rel#" + joinRel.getId() + ":" + joinRel.getRelTypeName() +
- " with row type: [" + joinRel.getRowType() + "]");
+ LOG.debug("Translating operator rel#" + joinRel.getId() + ":" + joinRel.getRelTypeName()
+ + " with row type: [" + joinRel.getRowType() + "]");
}
// 2. Convert join condition
@@ -323,15 +305,16 @@ public class HiveOpConverter {
// 4. Generate Join operator
JoinOperator joinOp = genJoin(joinRel, joinPredInfo, children, joinKeys);
- // 5. TODO: Extract condition for non-equi join elements (if any) and add it
+ // 5. TODO: Extract condition for non-equi join elements (if any) and
+ // add it
// 6. Virtual columns
- Map<Integer,VirtualColumn> vcolMap = new HashMap<Integer,VirtualColumn>();
+ Map<Integer, VirtualColumn> vcolMap = new HashMap<Integer, VirtualColumn>();
vcolMap.putAll(inputs[0].vcolMap);
if (extractJoinType(joinRel) != JoinType.LEFTSEMI) {
int shift = inputs[0].inputs.get(0).getSchema().getSignature().size();
- for (int i=1; i<inputs.length; i++) {
- vcolMap.putAll(HiveCalciteUtil.shiftVColsMap(inputs[i].vcolMap,shift));
+ for (int i = 1; i < inputs.length; i++) {
+ vcolMap.putAll(HiveCalciteUtil.shiftVColsMap(inputs[i].vcolMap, shift));
shift += inputs[i].inputs.get(0).getSchema().getSignature().size();
}
}
@@ -339,66 +322,68 @@ public class HiveOpConverter {
// 8. Return result
return new OpAttr(null, vcolMap, joinOp);
}
-
+
+ OpAttr visit(HiveAggregate aggRel) throws SemanticException {
+ OpAttr inputOpAf = dispatch(aggRel.getInput());
+ return HiveGBOpConvUtil.translateGB(inputOpAf, aggRel, hiveConf);
+ }
+
OpAttr visit(HiveSort sortRel) throws SemanticException {
OpAttr inputOpAf = dispatch(sortRel.getInput());
-
+
if (LOG.isDebugEnabled()) {
- LOG.debug("Translating operator rel#" + sortRel.getId() + ":" + sortRel.getRelTypeName() +
- " with row type: [" + sortRel.getRowType() + "]");
+ LOG.debug("Translating operator rel#" + sortRel.getId() + ":" + sortRel.getRelTypeName()
+ + " with row type: [" + sortRel.getRowType() + "]");
if (sortRel.getCollation() == RelCollations.EMPTY) {
- LOG.debug("Operator rel#" + sortRel.getId() + ":" + sortRel.getRelTypeName() +
- " consists of limit");
- }
- else if (sortRel.fetch == null) {
- LOG.debug("Operator rel#" + sortRel.getId() + ":" + sortRel.getRelTypeName() +
- " consists of sort");
- }
- else {
- LOG.debug("Operator rel#" + sortRel.getId() + ":" + sortRel.getRelTypeName() +
- " consists of sort+limit");
+ LOG.debug("Operator rel#" + sortRel.getId() + ":" + sortRel.getRelTypeName()
+ + " consists of limit");
+ } else if (sortRel.fetch == null) {
+ LOG.debug("Operator rel#" + sortRel.getId() + ":" + sortRel.getRelTypeName()
+ + " consists of sort");
+ } else {
+ LOG.debug("Operator rel#" + sortRel.getId() + ":" + sortRel.getRelTypeName()
+ + " consists of sort+limit");
}
}
-
+
Operator<?> inputOp = inputOpAf.inputs.get(0);
Operator<?> resultOp = inputOpAf.inputs.get(0);
// 1. If we need to sort tuples based on the value of some
- // of their columns
+ // of their columns
if (sortRel.getCollation() != RelCollations.EMPTY) {
-
- // In strict mode, in the presence of order by, limit must be specified
+
+ // In strict mode, in the presence of order by, limit must be
+ // specified
if (strictMode && sortRel.fetch == null) {
throw new SemanticException(ErrorMsg.NO_LIMIT_WITH_ORDERBY.getMsg());
}
-
+
// 1.a. Extract order for each column from collation
- // Generate sortCols and order
+ // Generate sortCols and order
List<ExprNodeDesc> sortCols = new ArrayList<ExprNodeDesc>();
StringBuilder order = new StringBuilder();
for (RelCollation collation : sortRel.getCollationList()) {
for (RelFieldCollation sortInfo : collation.getFieldCollations()) {
int sortColumnPos = sortInfo.getFieldIndex();
- ColumnInfo columnInfo = new ColumnInfo(inputOp.getSchema().getSignature().
- get(sortColumnPos));
+ ColumnInfo columnInfo = new ColumnInfo(inputOp.getSchema().getSignature()
+ .get(sortColumnPos));
ExprNodeColumnDesc sortColumn = new ExprNodeColumnDesc(columnInfo.getType(),
- columnInfo.getInternalName(), columnInfo.getTabAlias(),
- columnInfo.getIsVirtualCol());
+ columnInfo.getInternalName(), columnInfo.getTabAlias(), columnInfo.getIsVirtualCol());
sortCols.add(sortColumn);
if (sortInfo.getDirection() == RelFieldCollation.Direction.DESCENDING) {
order.append("-");
- }
- else {
+ } else {
order.append("+");
}
}
}
// Use only 1 reducer for order by
int numReducers = 1;
-
+
// 1.b. Generate reduce sink and project operator
- resultOp = genReduceSinkAndBacktrackSelect(resultOp, sortCols.toArray(new ExprNodeDesc[sortCols.size()]),
- -1, new ArrayList<ExprNodeDesc>(), order.toString(), numReducers,
- Operation.NOT_ACID, strictMode);
+ resultOp = genReduceSinkAndBacktrackSelect(resultOp,
+ sortCols.toArray(new ExprNodeDesc[sortCols.size()]), -1, new ArrayList<ExprNodeDesc>(),
+ order.toString(), numReducers, Operation.NOT_ACID, strictMode);
}
// 2. If we need to generate limit
@@ -407,14 +392,14 @@ public class HiveOpConverter {
LimitDesc limitDesc = new LimitDesc(limit);
// TODO: Set 'last limit' global property
ArrayList<ColumnInfo> cinfoLst = createColInfos(inputOp);
- resultOp = (LimitOperator) OperatorFactory.getAndMakeChild(
- limitDesc, new RowSchema(cinfoLst), resultOp);
-
+ resultOp = (LimitOperator) OperatorFactory.getAndMakeChild(limitDesc,
+ new RowSchema(cinfoLst), resultOp);
+
if (LOG.isDebugEnabled()) {
LOG.debug("Generated " + resultOp + " with row schema: [" + resultOp.getSchema() + "]");
}
}
-
+
// 3. Return result
return inputOpAf.clone(resultOp);
}
@@ -426,12 +411,13 @@ public class HiveOpConverter {
OpAttr inputOpAf = dispatch(filterRel.getInput());
if (LOG.isDebugEnabled()) {
- LOG.debug("Translating operator rel#" + filterRel.getId() + ":" + filterRel.getRelTypeName() +
- " with row type: [" + filterRel.getRowType() + "]");
+ LOG.debug("Translating operator rel#" + filterRel.getId() + ":" + filterRel.getRelTypeName()
+ + " with row type: [" + filterRel.getRowType() + "]");
}
ExprNodeDesc filCondExpr = filterRel.getCondition().accept(
- new ExprNodeConverter(inputOpAf.tabAlias, filterRel.getInput().getRowType(), false));
+ new ExprNodeConverter(inputOpAf.tabAlias, filterRel.getInput().getRowType(), false,
+ filterRel.getCluster().getTypeFactory()));
FilterDesc filDesc = new FilterDesc(filCondExpr, false);
ArrayList<ColumnInfo> cinfoLst = createColInfos(inputOpAf.inputs.get(0));
FilterOperator filOp = (FilterOperator) OperatorFactory.getAndMakeChild(filDesc, new RowSchema(
@@ -447,13 +433,13 @@ public class HiveOpConverter {
OpAttr visit(HiveUnion unionRel) throws SemanticException {
// 1. Convert inputs
OpAttr[] inputs = new OpAttr[unionRel.getInputs().size()];
- for (int i=0; i<inputs.length; i++) {
+ for (int i = 0; i < inputs.length; i++) {
inputs[i] = dispatch(unionRel.getInput(i));
}
if (LOG.isDebugEnabled()) {
- LOG.debug("Translating operator rel#" + unionRel.getId() + ":" + unionRel.getRelTypeName() +
- " with row type: [" + unionRel.getRowType() + "]");
+ LOG.debug("Translating operator rel#" + unionRel.getId() + ":" + unionRel.getRelTypeName()
+ + " with row type: [" + unionRel.getRowType() + "]");
}
// 2. Create a new union operator
@@ -461,11 +447,11 @@ public class HiveOpConverter {
unionDesc.setNumInputs(inputs.length);
ArrayList<ColumnInfo> cinfoLst = createColInfos(inputs[0].inputs.get(0));
Operator<?>[] children = new Operator<?>[inputs.length];
- for (int i=0; i<children.length; i++) {
+ for (int i = 0; i < children.length; i++) {
children[i] = inputs[i].inputs.get(0);
}
- Operator<? extends OperatorDesc> unionOp = OperatorFactory.getAndMakeChild(
- unionDesc, new RowSchema(cinfoLst), children);
+ Operator<? extends OperatorDesc> unionOp = OperatorFactory.getAndMakeChild(unionDesc,
+ new RowSchema(cinfoLst), children);
if (LOG.isDebugEnabled()) {
LOG.debug("Generated " + unionOp + " with row schema: [" + unionOp.getSchema() + "]");
@@ -479,8 +465,8 @@ public class HiveOpConverter {
OpAttr inputOpAf = dispatch(exchangeRel.getInput());
if (LOG.isDebugEnabled()) {
- LOG.debug("Translating operator rel#" + exchangeRel.getId() + ":" + exchangeRel.getRelTypeName() +
- " with row type: [" + exchangeRel.getRowType() + "]");
+ LOG.debug("Translating operator rel#" + exchangeRel.getId() + ":"
+ + exchangeRel.getRelTypeName() + " with row type: [" + exchangeRel.getRowType() + "]");
}
RelDistribution distribution = exchangeRel.getDistribution();
@@ -488,23 +474,22 @@ public class HiveOpConverter {
throw new SemanticException("Only hash distribution supported for LogicalExchange");
}
ExprNodeDesc[] expressions = new ExprNodeDesc[distribution.getKeys().size()];
- for (int i=0; i<distribution.getKeys().size(); i++) {
+ for (int i = 0; i < distribution.getKeys().size(); i++) {
int key = distribution.getKeys().get(i);
- ColumnInfo colInfo = inputOpAf.inputs.get(0).getSchema().
- getSignature().get(key);
+ ColumnInfo colInfo = inputOpAf.inputs.get(0).getSchema().getSignature().get(key);
ExprNodeDesc column = new ExprNodeColumnDesc(colInfo);
expressions[i] = column;
}
- ReduceSinkOperator rsOp = genReduceSink(inputOpAf.inputs.get(0),
- expressions, reduceSinkTagGenerator++, -1, Operation.NOT_ACID, strictMode);
+ ReduceSinkOperator rsOp = genReduceSink(inputOpAf.inputs.get(0), expressions,
+ reduceSinkTagGenerator++, -1, Operation.NOT_ACID, strictMode);
return inputOpAf.clone(rsOp);
}
private OpAttr genPTF(OpAttr inputOpAf, WindowingSpec wSpec) throws SemanticException {
Operator<?> input = inputOpAf.inputs.get(0);
-
+
wSpec.validateAndMakeEffective();
WindowingComponentizer groups = new WindowingComponentizer(wSpec);
RowResolver rr = new RowResolver();
@@ -512,7 +497,7 @@ public class HiveOpConverter {
rr.put(ci.getTabAlias(), ci.getInternalName(), ci);
}
- while(groups.hasNext() ) {
+ while (groups.hasNext()) {
wSpec = groups.next(hiveConf, semanticAnalyzer, unparseTranslator, rr);
// 1. Create RS and backtrack Select operator on top
@@ -544,17 +529,17 @@ public class HiveOpConverter {
}
SelectOperator selectOp = genReduceSinkAndBacktrackSelect(input,
- keyCols.toArray(new ExprNodeDesc[keyCols.size()]),
- reduceSinkTagGenerator++, partCols, order.toString(),
- -1, Operation.NOT_ACID, strictMode);
+ keyCols.toArray(new ExprNodeDesc[keyCols.size()]), reduceSinkTagGenerator++, partCols,
+ order.toString(), -1, Operation.NOT_ACID, strictMode);
// 2. Finally create PTF
PTFTranslator translator = new PTFTranslator();
- PTFDesc ptfDesc = translator.translate(wSpec, semanticAnalyzer, hiveConf, rr, unparseTranslator);
+ PTFDesc ptfDesc = translator.translate(wSpec, semanticAnalyzer, hiveConf, rr,
+ unparseTranslator);
RowResolver ptfOpRR = ptfDesc.getFuncDef().getOutputShape().getRr();
Operator<?> ptfOp = OperatorFactory.getAndMakeChild(ptfDesc,
- new RowSchema(ptfOpRR.getColumnInfos()), selectOp);
+ new RowSchema(ptfOpRR.getColumnInfos()), selectOp);
if (LOG.isDebugEnabled()) {
LOG.debug("Generated " + ptfOp + " with row schema: [" + ptfOp.getSchema() + "]");
@@ -564,12 +549,11 @@ public class HiveOpConverter {
rr = ptfOpRR;
input = ptfOp;
}
-
+
return inputOpAf.clone(input);
}
- private ExprNodeDesc[][] extractJoinKeys(JoinPredicateInfo joinPredInfo,
- List<RelNode> inputs) {
+ private ExprNodeDesc[][] extractJoinKeys(JoinPredicateInfo joinPredInfo, List<RelNode> inputs) {
ExprNodeDesc[][] joinKeys = new ExprNodeDesc[inputs.size()][];
for (int i = 0; i < inputs.size(); i++) {
joinKeys[i] = new ExprNodeDesc[joinPredInfo.getEquiJoinPredicateElements().size()];
@@ -583,22 +567,20 @@ public class HiveOpConverter {
}
private static SelectOperator genReduceSinkAndBacktrackSelect(Operator<?> input,
- ExprNodeDesc[] keys, int tag, ArrayList<ExprNodeDesc> partitionCols,
- String order, int numReducers, Operation acidOperation,
- boolean strictMode) throws SemanticException {
+ ExprNodeDesc[] keys, int tag, ArrayList<ExprNodeDesc> partitionCols, String order,
+ int numReducers, Operation acidOperation, boolean strictMode) throws SemanticException {
// 1. Generate RS operator
- ReduceSinkOperator rsOp = genReduceSink(input, keys, tag, partitionCols,
- order, numReducers, acidOperation, strictMode);
+ ReduceSinkOperator rsOp = genReduceSink(input, keys, tag, partitionCols, order, numReducers,
+ acidOperation, strictMode);
// 2. Generate backtrack Select operator
- Map<String, ExprNodeDesc> descriptors = buildBacktrackFromReduceSink(
- (ReduceSinkOperator) rsOp, input);
- SelectDesc selectDesc = new SelectDesc(
- new ArrayList<ExprNodeDesc>(descriptors.values()),
- new ArrayList<String>(descriptors.keySet()));
+ Map<String, ExprNodeDesc> descriptors = buildBacktrackFromReduceSink((ReduceSinkOperator) rsOp,
+ input);
+ SelectDesc selectDesc = new SelectDesc(new ArrayList<ExprNodeDesc>(descriptors.values()),
+ new ArrayList<String>(descriptors.keySet()));
ArrayList<ColumnInfo> cinfoLst = createColInfos(input);
SelectOperator selectOp = (SelectOperator) OperatorFactory.getAndMakeChild(selectDesc,
- new RowSchema(cinfoLst), rsOp);
+ new RowSchema(cinfoLst), rsOp);
selectOp.setColumnExprMap(descriptors);
if (LOG.isDebugEnabled()) {
@@ -608,19 +590,17 @@ public class HiveOpConverter {
return selectOp;
}
- private static ReduceSinkOperator genReduceSink(Operator<?> input, ExprNodeDesc[] keys,
- int tag, int numReducers, Operation acidOperation,
- boolean strictMode) throws SemanticException {
- return genReduceSink(input, keys, tag, new ArrayList<ExprNodeDesc>(),
- "", numReducers, acidOperation, strictMode);
+ private static ReduceSinkOperator genReduceSink(Operator<?> input, ExprNodeDesc[] keys, int tag,
+ int numReducers, Operation acidOperation, boolean strictMode) throws SemanticException {
+ return genReduceSink(input, keys, tag, new ArrayList<ExprNodeDesc>(), "", numReducers,
+ acidOperation, strictMode);
}
-
+
@SuppressWarnings({ "rawtypes", "unchecked" })
- private static ReduceSinkOperator genReduceSink(Operator<?> input,
- ExprNodeDesc[] keys, int tag, ArrayList<ExprNodeDesc> partitionCols,
- String order, int numReducers, Operation acidOperation,
- boolean strictMode) throws SemanticException {
- Operator dummy = Operator.createDummy(); // dummy for backtracking
+ private static ReduceSinkOperator genReduceSink(Operator<?> input, ExprNodeDesc[] keys, int tag,
+ ArrayList<ExprNodeDesc> partitionCols, String order, int numReducers,
+ Operation acidOperation, boolean strictMode) throws SemanticException {
+ Operator dummy = Operator.createDummy(); // dummy for backtracking
dummy.setParentOperators(Arrays.asList(input));
ArrayList<ExprNodeDesc> reduceKeys = new ArrayList<ExprNodeDesc>();
@@ -690,31 +670,29 @@ public class HiveOpConverter {
ReduceSinkDesc rsDesc;
if (order.isEmpty()) {
- rsDesc = PlanUtils.getReduceSinkDesc(reduceKeys, reduceValues,
- outputColumnNames, false, tag, reduceKeys.size(),
- numReducers, acidOperation);
+ rsDesc = PlanUtils.getReduceSinkDesc(reduceKeys, reduceValues, outputColumnNames, false, tag,
+ reduceKeys.size(), numReducers, acidOperation);
} else {
- rsDesc = PlanUtils.getReduceSinkDesc(reduceKeys, reduceValues,
- outputColumnNames, false, tag, partitionCols,
- order, numReducers, acidOperation);
+ rsDesc = PlanUtils.getReduceSinkDesc(reduceKeys, reduceValues, outputColumnNames, false, tag,
+ partitionCols, order, numReducers, acidOperation);
}
ReduceSinkOperator rsOp = (ReduceSinkOperator) OperatorFactory.getAndMakeChild(rsDesc,
- new RowSchema(outputColumns), input);
+ new RowSchema(outputColumns), input);
List<String> keyColNames = rsDesc.getOutputKeyColumnNames();
- for (int i = 0 ; i < keyColNames.size(); i++) {
+ for (int i = 0; i < keyColNames.size(); i++) {
colExprMap.put(Utilities.ReduceField.KEY + "." + keyColNames.get(i), reduceKeys.get(i));
}
List<String> valColNames = rsDesc.getOutputValueColumnNames();
- for (int i = 0 ; i < valColNames.size(); i++) {
+ for (int i = 0; i < valColNames.size(); i++) {
colExprMap.put(Utilities.ReduceField.VALUE + "." + valColNames.get(i), reduceValues.get(i));
}
rsOp.setValueIndex(index);
rsOp.setColumnExprMap(colExprMap);
- rsOp.setInputAliases(input.getSchema().getColumnNames().toArray(
- new String[input.getSchema().getColumnNames().size()]));
+ rsOp.setInputAliases(input.getSchema().getColumnNames()
+ .toArray(new String[input.getSchema().getColumnNames().size()]));
if (LOG.isDebugEnabled()) {
LOG.debug("Generated " + rsOp + " with row schema: [" + rsOp.getSchema() + "]");
@@ -728,14 +706,14 @@ public class HiveOpConverter {
// Extract join type
JoinType joinType = extractJoinType(hiveJoin);
-
+
// NOTE: Currently binary joins only
JoinCondDesc[] joinCondns = new JoinCondDesc[1];
joinCondns[0] = new JoinCondDesc(new JoinCond(0, 1, joinType));
ArrayList<ColumnInfo> outputColumns = new ArrayList<ColumnInfo>();
- ArrayList<String> outputColumnNames =
- new ArrayList<String>(hiveJoin.getRowType().getFieldNames());
+ ArrayList<String> outputColumnNames = new ArrayList<String>(hiveJoin.getRowType()
+ .getFieldNames());
Operator<?>[] childOps = new Operator[children.size()];
Map<String, Byte> reversedExprs = new HashMap<String, Byte>();
@@ -769,7 +747,7 @@ public class HiveOpConverter {
posToAliasMap.put(pos, new HashSet<String>(inputRS.getSchema().getTableNames()));
Map<String, ExprNodeDesc> descriptors = buildBacktrackFromReduceSink(outputPos,
- outputColumnNames, keyColNames, valColNames, index, parent);
+ outputColumnNames, keyColNames, valColNames, index, parent);
List<ColumnInfo> parentColumns = parent.getSchema().getSignature();
for (int i = 0; i < index.length; i++) {
@@ -785,15 +763,13 @@ public class HiveOpConverter {
childOps[pos] = inputRS;
}
- boolean noOuterJoin = joinType != JoinType.FULLOUTER
- && joinType != JoinType.LEFTOUTER
- && joinType != JoinType.RIGHTOUTER;
- JoinDesc desc = new JoinDesc(exprMap, outputColumnNames, noOuterJoin,
- joinCondns, joinKeys);
+ boolean noOuterJoin = joinType != JoinType.FULLOUTER && joinType != JoinType.LEFTOUTER
+ && joinType != JoinType.RIGHTOUTER;
+ JoinDesc desc = new JoinDesc(exprMap, outputColumnNames, noOuterJoin, joinCondns, joinKeys);
desc.setReversedExprs(reversedExprs);
- JoinOperator joinOp = (JoinOperator) OperatorFactory.getAndMakeChild(desc,
- new RowSchema(outputColumns), childOps);
+ JoinOperator joinOp = (JoinOperator) OperatorFactory.getAndMakeChild(desc, new RowSchema(
+ outputColumns), childOps);
joinOp.setColumnExprMap(colExprMap);
joinOp.setPosToAliasMap(posToAliasMap);
@@ -818,32 +794,32 @@ public class HiveOpConverter {
// OUTER AND INNER JOINS
JoinType resultJoinType;
switch (join.getJoinType()) {
- case FULL:
- resultJoinType = JoinType.FULLOUTER;
- break;
- case LEFT:
- resultJoinType = JoinType.LEFTOUTER;
- break;
- case RIGHT:
- resultJoinType = JoinType.RIGHTOUTER;
- break;
- default:
- resultJoinType = JoinType.INNER;
- break;
+ case FULL:
+ resultJoinType = JoinType.FULLOUTER;
+ break;
+ case LEFT:
+ resultJoinType = JoinType.LEFTOUTER;
+ break;
+ case RIGHT:
+ resultJoinType = JoinType.RIGHTOUTER;
+ break;
+ default:
+ resultJoinType = JoinType.INNER;
+ break;
}
return resultJoinType;
}
- private static Map<String, ExprNodeDesc> buildBacktrackFromReduceSink(
- ReduceSinkOperator rsOp, Operator<?> inputOp) {
- return buildBacktrackFromReduceSink(0, inputOp.getSchema().getColumnNames(),
- rsOp.getConf().getOutputKeyColumnNames(), rsOp.getConf().getOutputValueColumnNames(),
- rsOp.getValueIndex(), inputOp);
+ private static Map<String, ExprNodeDesc> buildBacktrackFromReduceSink(ReduceSinkOperator rsOp,
+ Operator<?> inputOp) {
+ return buildBacktrackFromReduceSink(0, inputOp.getSchema().getColumnNames(), rsOp.getConf()
+ .getOutputKeyColumnNames(), rsOp.getConf().getOutputValueColumnNames(),
+ rsOp.getValueIndex(), inputOp);
}
private static Map<String, ExprNodeDesc> buildBacktrackFromReduceSink(int initialPos,
- List<String> outputColumnNames, List<String> keyColNames, List<String> valueColNames,
- int[] index, Operator<?> inputOp) {
+ List<String> outputColumnNames, List<String> keyColNames, List<String> valueColNames,
+ int[] index, Operator<?> inputOp) {
Map<String, ExprNodeDesc> columnDescriptors = new LinkedHashMap<String, ExprNodeDesc>();
for (int i = 0; i < index.length; i++) {
ColumnInfo info = new ColumnInfo(inputOp.getSchema().getSignature().get(i));
@@ -853,17 +829,16 @@ public class HiveOpConverter {
} else {
field = Utilities.ReduceField.VALUE + "." + valueColNames.get(-index[i] - 1);
}
- ExprNodeColumnDesc desc = new ExprNodeColumnDesc(info.getType(),
- field, info.getTabAlias(), info.getIsVirtualCol());
- columnDescriptors.put(outputColumnNames.get(initialPos+i), desc);
+ ExprNodeColumnDesc desc = new ExprNodeColumnDesc(info.getType(), field, info.getTabAlias(),
+ info.getIsVirtualCol());
+ columnDescriptors.put(outputColumnNames.get(initialPos + i), desc);
}
return columnDescriptors;
}
- private static ExprNodeDesc convertToExprNode(RexNode rn, RelNode inputRel,
- String tabAlias) {
- return (ExprNodeDesc) rn.accept(new ExprNodeConverter(tabAlias,
- inputRel.getRowType(), false));
+ private static ExprNodeDesc convertToExprNode(RexNode rn, RelNode inputRel, String tabAlias) {
+ return (ExprNodeDesc) rn.accept(new ExprNodeConverter(tabAlias, inputRel.getRowType(), false,
+ inputRel.getCluster().getTypeFactory()));
}
private static ArrayList<ColumnInfo> createColInfos(Operator<?> input) {
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java?rev=1669206&r1=1669205&r2=1669206&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java Wed Mar 25 18:44:16 2015
@@ -611,7 +611,7 @@ public class CalcitePlanner extends Sema
RelNode modifiedOptimizedOptiqPlan = introduceProjectIfNeeded(optimizedOptiqPlan);
- Operator<?> hiveRoot = new HiveOpConverter(this, conf, unparseTranslator, topOps, HiveOpConverter.getAggOPMode(conf),
+ Operator<?> hiveRoot = new HiveOpConverter(this, conf, unparseTranslator, topOps,
conf.getVar(HiveConf.ConfVars.HIVEMAPREDMODE).equalsIgnoreCase("strict")).convert(modifiedOptimizedOptiqPlan);
RowResolver hiveRootRR = genRowResolver(hiveRoot, getQB());
opParseCtx.put(hiveRoot, new OpParseContext(hiveRootRR));
@@ -1836,10 +1836,40 @@ public class CalcitePlanner extends Sema
private RelNode genGBLogicalPlan(QB qb, RelNode srcRel) throws SemanticException {
RelNode gbRel = null;
QBParseInfo qbp = getQBParseInfo(qb);
-
- // 1. Gather GB Expressions (AST) (GB + Aggregations)
// NOTE: Multi Insert is not supported
String detsClauseName = qbp.getClauseNames().iterator().next();
+ List<ASTNode> grpByAstExprs = SemanticAnalyzer.getGroupByForClause(qbp, detsClauseName);
+ HashMap<String, ASTNode> aggregationTrees = qbp.getAggregationExprsForClause(detsClauseName);
+ // NOTE: Multi Insert is not supported
+ boolean cubeRollupGrpSetPresent = (!qbp.getDestRollups().isEmpty()
+ || !qbp.getDestGroupingSets().isEmpty() || !qbp.getDestCubes().isEmpty());
+
+ // 0. Sanity check
+ if (conf.getBoolVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)
+ && qbp.getDistinctFuncExprsForClause(detsClauseName).size() > 1) {
+ throw new SemanticException(ErrorMsg.UNSUPPORTED_MULTIPLE_DISTINCTS.getMsg());
+ }
+ if (cubeRollupGrpSetPresent) {
+ if (!HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVEMAPSIDEAGGREGATE)) {
+ throw new SemanticException(ErrorMsg.HIVE_GROUPING_SETS_AGGR_NOMAPAGGR.getMsg());
+ }
+
+ if (conf.getBoolVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)) {
+ checkExpressionsForGroupingSet(grpByAstExprs, qb.getParseInfo()
+ .getDistinctFuncExprsForClause(detsClauseName), aggregationTrees,
+ this.relToHiveRR.get(srcRel));
+
+ if (qbp.getDestGroupingSets().size() > conf
+ .getIntVar(HiveConf.ConfVars.HIVE_NEW_JOB_GROUPING_SET_CARDINALITY)) {
+ String errorMsg = "The number of rows per input row due to grouping sets is "
+ + qbp.getDestGroupingSets().size();
+ throw new SemanticException(
+ ErrorMsg.HIVE_GROUPING_SETS_THRESHOLD_NOT_ALLOWED_WITH_SKEW.getMsg(errorMsg));
+ }
+ }
+ }
+
+ // 1. Gather GB Expressions (AST) (GB + Aggregations)
// Check and transform group by *. This will only happen for select distinct *.
// Here the "genSelectPlan" is being leveraged.
// The main benefits are (1) remove virtual columns that should
@@ -1857,8 +1887,6 @@ public class CalcitePlanner extends Sema
qbp.setSelExprForClause(detsClauseName, SemanticAnalyzer.genSelectDIAST(rr));
}
}
- List<ASTNode> grpByAstExprs = SemanticAnalyzer.getGroupByForClause(qbp, detsClauseName);
- HashMap<String, ASTNode> aggregationTrees = qbp.getAggregationExprsForClause(detsClauseName);
boolean hasGrpByAstExprs = (grpByAstExprs != null && !grpByAstExprs.isEmpty()) ? true : false;
boolean hasAggregationTrees = (aggregationTrees != null && !aggregationTrees.isEmpty()) ? true
: false;
@@ -1890,9 +1918,7 @@ public class CalcitePlanner extends Sema
// 4. GroupingSets, Cube, Rollup
int groupingColsSize = gbExprNDescLst.size();
List<Integer> groupingSets = null;
- if (!qbp.getDestRollups().isEmpty()
- || !qbp.getDestGroupingSets().isEmpty()
- || !qbp.getDestCubes().isEmpty()) {
+ if (cubeRollupGrpSetPresent) {
if (qbp.getDestRollups().contains(detsClauseName)) {
groupingSets = getGroupingSetsForRollup(grpByAstExprs.size());
} else if (qbp.getDestCubes().contains(detsClauseName)) {
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java?rev=1669206&r1=1669205&r2=1669206&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java Wed Mar 25 18:44:16 2015
@@ -3443,7 +3443,7 @@ public class SemanticAnalyzer extends Ba
return ret;
}
- private int setBit(int bitmap, int bitIdx) {
+ public static int setBit(int bitmap, int bitIdx) {
return bitmap | (1 << bitIdx);
}
@@ -3979,10 +3979,10 @@ public class SemanticAnalyzer extends Ba
/**
* Class to store GenericUDAF related information.
*/
- static class GenericUDAFInfo {
- ArrayList<ExprNodeDesc> convertedParameters;
- GenericUDAFEvaluator genericUDAFEvaluator;
- TypeInfo returnType;
+ public static class GenericUDAFInfo {
+ public ArrayList<ExprNodeDesc> convertedParameters;
+ public GenericUDAFEvaluator genericUDAFEvaluator;
+ public TypeInfo returnType;
}
/**
@@ -4023,7 +4023,7 @@ public class SemanticAnalyzer extends Ba
* Returns the GenericUDAFEvaluator for the aggregation. This is called once
* for each GroupBy aggregation.
*/
- static GenericUDAFEvaluator getGenericUDAFEvaluator(String aggName,
+ public static GenericUDAFEvaluator getGenericUDAFEvaluator(String aggName,
ArrayList<ExprNodeDesc> aggParameters, ASTNode aggTree,
boolean isDistinct, boolean isAllColumns)
throws SemanticException {
@@ -4053,7 +4053,7 @@ public class SemanticAnalyzer extends Ba
* @throws SemanticException
* when the UDAF is not found or has problems.
*/
- static GenericUDAFInfo getGenericUDAFInfo(GenericUDAFEvaluator evaluator,
+ public static GenericUDAFInfo getGenericUDAFInfo(GenericUDAFEvaluator evaluator,
GenericUDAFEvaluator.Mode emode, ArrayList<ExprNodeDesc> aggParameters)
throws SemanticException {
@@ -4082,7 +4082,7 @@ public class SemanticAnalyzer extends Ba
return r;
}
- static GenericUDAFEvaluator.Mode groupByDescModeToUDAFMode(
+ public static GenericUDAFEvaluator.Mode groupByDescModeToUDAFMode(
GroupByDesc.Mode mode, boolean isDistinct) {
switch (mode) {
case COMPLETE:
@@ -4125,7 +4125,7 @@ public class SemanticAnalyzer extends Ba
* @return the ExprNodeDesc of the constant parameter if the given internalName represents
* a constant parameter; otherwise, return null
*/
- private ExprNodeDesc isConstantParameterInAggregationParameters(String internalName,
+ public static ExprNodeDesc isConstantParameterInAggregationParameters(String internalName,
List<ExprNodeDesc> reduceValues) {
// only the pattern of "VALUE._col([0-9]+)" should be handled.
@@ -5572,7 +5572,7 @@ public class SemanticAnalyzer extends Ba
return false;
}
- private void checkExpressionsForGroupingSet(List<ASTNode> grpByExprs,
+ void checkExpressionsForGroupingSet(List<ASTNode> grpByExprs,
List<ASTNode> distinctGrpByExprs,
Map<String, ASTNode> aggregationTrees,
RowResolver inputRowResolver) throws SemanticException {
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java?rev=1669206&r1=1669205&r2=1669206&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java Wed Mar 25 18:44:16 2015
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import org.apache.hadoop.hive.ql.exec.ColumnInfo;
import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator;
import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluatorFactory;
import org.apache.hadoop.hive.ql.exec.FunctionRegistry;
@@ -444,4 +445,42 @@ public class ExprNodeDescUtils {
// If the child is also decimal, no cast is needed (we hope - can target type be narrower?).
return HiveDecimalUtils.getDecimalTypeForPrimitiveCategory(childTi);
}
+
+ /**
+ * Build ExprNodeColumnDesc for the projections in the input operator from
+ * sartpos to endpos(both included). Operator must have an associated
+ * colExprMap.
+ *
+ * @param inputOp
+ * Input Hive Operator
+ * @param startPos
+ * starting position in the input operator schema; must be >=0 and <=
+ * endPos
+ * @param endPos
+ * end position in the input operator schema; must be >=0.
+ * @return List of ExprNodeDesc
+ */
+ public static ArrayList<ExprNodeDesc> genExprNodeDesc(Operator inputOp, int startPos, int endPos,
+ boolean addEmptyTabAlias, boolean setColToNonVirtual) {
+ ArrayList<ExprNodeDesc> exprColLst = new ArrayList<ExprNodeDesc>();
+ List<ColumnInfo> colInfoLst = inputOp.getSchema().getSignature();
+
+ String tabAlias;
+ boolean vc;
+ ColumnInfo ci;
+ for (int i = startPos; i <= endPos; i++) {
+ ci = colInfoLst.get(i);
+ tabAlias = ci.getTabAlias();
+ if (addEmptyTabAlias) {
+ tabAlias = "";
+ }
+ vc = ci.getIsVirtualCol();
+ if (setColToNonVirtual) {
+ vc = false;
+ }
+ exprColLst.add(new ExprNodeColumnDesc(ci.getType(), ci.getInternalName(), tabAlias, vc));
+ }
+
+ return exprColLst;
+ }
}