You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2015/01/23 20:59:24 UTC
svn commit: r1654355 [14/27] - in /hive/branches/llap: ./
beeline/src/java/org/apache/hive/beeline/
cli/src/java/org/apache/hadoop/hive/cli/
common/src/java/org/apache/hadoop/hive/common/
common/src/java/org/apache/hadoop/hive/conf/ data/conf/ data/con...
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GlobalLimitOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GlobalLimitOptimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GlobalLimitOptimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GlobalLimitOptimizer.java Fri Jan 23 19:59:11 2015
@@ -62,10 +62,8 @@ public class GlobalLimitOptimizer implem
GlobalLimitCtx globalLimitCtx = pctx.getGlobalLimitCtx();
Map<TableScanOperator, ExprNodeDesc> opToPartPruner = pctx.getOpToPartPruner();
Map<String, SplitSample> nameToSplitSample = pctx.getNameToSplitSample();
- Map<TableScanOperator, Table> topToTable = pctx.getTopToTable();
QB qb = pctx.getQB();
- HiveConf conf = pctx.getConf();
QBParseInfo qbParseInfo = qb.getParseInfo();
// determine the query qualifies reduce input size for LIMIT
@@ -93,7 +91,7 @@ public class GlobalLimitOptimizer implem
// query qualify for the optimization
if (tempGlobalLimit != null && tempGlobalLimit != 0) {
TableScanOperator ts = (TableScanOperator) topOps.values().toArray()[0];
- Table tab = topToTable.get(ts);
+ Table tab = ts.getConf().getTableMetadata();
if (!tab.isPartitioned()) {
if (qbParseInfo.getDestToWhereExpr().isEmpty()) {
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GroupByOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GroupByOptimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GroupByOptimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GroupByOptimizer.java Fri Jan 23 19:59:11 2015
@@ -304,7 +304,7 @@ public class GroupByOptimizer implements
// Create a mapping from the group by columns to the table columns
Map<String, String> tableColsMapping = new HashMap<String, String>();
Set<String> constantCols = new HashSet<String>();
- Table table = pGraphContext.getTopToTable().get(currOp);
+ Table table = tableScanOp.getConf().getTableMetadata();
for (FieldSchema col : table.getAllCols()) {
tableColsMapping.put(col.getName(), col.getName());
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/IndexUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/IndexUtils.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/IndexUtils.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/IndexUtils.java Fri Jan 23 19:59:11 2015
@@ -29,6 +29,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.Index;
import org.apache.hadoop.hive.ql.Driver;
@@ -123,22 +124,23 @@ public final class IndexUtils {
Partition part) throws HiveException {
LOG.info("checking index staleness...");
try {
- FileSystem partFs = part.getDataLocation().getFileSystem(hive.getConf());
- FileStatus partFss = partFs.getFileStatus(part.getDataLocation());
- String ts = index.getParameters().get(part.getSpec().toString());
- if (ts == null) {
+ String indexTs = index.getParameters().get(part.getSpec().toString());
+ if (indexTs == null) {
return false;
}
- long indexTs = Long.parseLong(ts);
- LOG.info(partFss.getModificationTime());
- LOG.info(ts);
- if (partFss.getModificationTime() > indexTs) {
- LOG.info("index is stale on the partitions that matched " + part.getSpec());
- return false;
+
+ FileSystem partFs = part.getDataLocation().getFileSystem(hive.getConf());
+ FileStatus[] parts = partFs.listStatus(part.getDataLocation(), FileUtils.HIDDEN_FILES_PATH_FILTER);
+ for (FileStatus status : parts) {
+ if (status.getModificationTime() > Long.parseLong(indexTs)) {
+ LOG.info("Index is stale on partition '" + part.getName()
+ + "'. Modified time (" + status.getModificationTime() + ") for '" + status.getPath()
+ + "' is higher than index creation time (" + indexTs + ").");
+ return false;
+ }
}
} catch (IOException e) {
- LOG.info("failed to grab timestamp info");
- throw new HiveException(e);
+ throw new HiveException("Failed to grab timestamp information from partition '" + part.getName() + "': " + e.getMessage(), e);
}
return true;
}
@@ -156,22 +158,23 @@ public final class IndexUtils {
for (Index index : indexes) {
LOG.info("checking index staleness...");
try {
- FileSystem srcFs = src.getPath().getFileSystem(hive.getConf());
- FileStatus srcFss= srcFs.getFileStatus(src.getPath());
- String ts = index.getParameters().get("base_timestamp");
- if (ts == null) {
+ String indexTs = index.getParameters().get("base_timestamp");
+ if (indexTs == null) {
return false;
}
- long indexTs = Long.parseLong(ts);
- LOG.info(srcFss.getModificationTime());
- LOG.info(ts);
- if (srcFss.getModificationTime() > indexTs) {
- LOG.info("index is stale ");
- return false;
+
+ FileSystem srcFs = src.getPath().getFileSystem(hive.getConf());
+ FileStatus[] srcs = srcFs.listStatus(src.getPath(), FileUtils.HIDDEN_FILES_PATH_FILTER);
+ for (FileStatus status : srcs) {
+ if (status.getModificationTime() > Long.parseLong(indexTs)) {
+ LOG.info("Index is stale on table '" + src.getTableName()
+ + "'. Modified time (" + status.getModificationTime() + ") for '" + status.getPath()
+ + "' is higher than index creation time (" + indexTs + ").");
+ return false;
+ }
}
} catch (IOException e) {
- LOG.info("failed to grab timestamp info");
- throw new HiveException(e);
+ throw new HiveException("Failed to grab timestamp information from table '" + src.getTableName() + "': " + e.getMessage(), e);
}
}
return true;
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/JoinReorder.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/JoinReorder.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/JoinReorder.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/JoinReorder.java Fri Jan 23 19:59:11 2015
@@ -26,7 +26,6 @@ import org.apache.hadoop.hive.ql.exec.Op
import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
import org.apache.hadoop.hive.ql.parse.ParseContext;
-import org.apache.hadoop.hive.ql.parse.QBJoinTree;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
@@ -95,9 +94,9 @@ public class JoinReorder implements Tran
private Set<String> getBigTables(ParseContext joinCtx) {
Set<String> bigTables = new HashSet<String>();
- for (QBJoinTree qbJoin : joinCtx.getJoinContext().values()) {
- if (qbJoin.getStreamAliases() != null) {
- bigTables.addAll(qbJoin.getStreamAliases());
+ for (JoinOperator joinOp : joinCtx.getJoinOps()) {
+ if (joinOp.getConf().getStreamAliases() != null) {
+ bigTables.addAll(joinOp.getConf().getStreamAliases());
}
}
@@ -155,7 +154,7 @@ public class JoinReorder implements Tran
public ParseContext transform(ParseContext pactx) throws SemanticException {
Set<String> bigTables = getBigTables(pactx);
- for (JoinOperator joinOp : pactx.getJoinContext().keySet()) {
+ for (JoinOperator joinOp : pactx.getJoinOps()) {
reorder(joinOp, bigTables);
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java Fri Jan 23 19:59:11 2015
@@ -61,7 +61,6 @@ import org.apache.hadoop.hive.ql.lib.Rul
import org.apache.hadoop.hive.ql.parse.GenMapRedWalker;
import org.apache.hadoop.hive.ql.parse.OpParseContext;
import org.apache.hadoop.hive.ql.parse.ParseContext;
-import org.apache.hadoop.hive.ql.parse.QBJoinTree;
import org.apache.hadoop.hive.ql.parse.RowResolver;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
@@ -96,18 +95,13 @@ public class MapJoinProcessor implements
// needs to be passed. Use the string defined below for that.
private static final String MAPJOINKEY_FIELDPREFIX = "mapjoinkey";
- private ParseContext pGraphContext;
-
- /**
- * empty constructor.
- */
public MapJoinProcessor() {
- pGraphContext = null;
}
@SuppressWarnings("nls")
- private Operator<? extends OperatorDesc>
- putOpInsertMap(Operator<? extends OperatorDesc> op, RowResolver rr) {
+ private static Operator<? extends OperatorDesc> putOpInsertMap (
+ ParseContext pGraphContext, Operator<? extends OperatorDesc> op,
+ RowResolver rr) {
OpParseContext ctx = new OpParseContext(rr);
pGraphContext.getOpParseCtx().put(op, ctx);
return op;
@@ -232,10 +226,10 @@ public class MapJoinProcessor implements
throws SemanticException {
LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtxMap =
newWork.getMapWork().getOpParseCtxMap();
- QBJoinTree newJoinTree = newWork.getMapWork().getJoinTree();
// generate the map join operator; already checked the map join
- MapJoinOperator newMapJoinOp = MapJoinProcessor.convertMapJoin(conf, opParseCtxMap, op,
- newJoinTree, mapJoinPos, true, false);
+ MapJoinOperator newMapJoinOp = new MapJoinProcessor().convertMapJoin(conf, opParseCtxMap, op,
+ newWork.getMapWork().isLeftInputJoin(), newWork.getMapWork().getBaseSrc(), newWork.getMapWork().getMapAliases(),
+ mapJoinPos, true, false);
genLocalWorkForMapJoin(newWork, newMapJoinOp, mapJoinPos);
}
@@ -247,7 +241,9 @@ public class MapJoinProcessor implements
MapJoinProcessor.genMapJoinLocalWork(newWork, newMapJoinOp, mapJoinPos);
// clean up the mapred work
newWork.getMapWork().setOpParseCtxMap(null);
- newWork.getMapWork().setJoinTree(null);
+ newWork.getMapWork().setLeftInputJoin(false);
+ newWork.getMapWork().setBaseSrc(null);
+ newWork.getMapWork().setMapAliases(null);
} catch (Exception e) {
e.printStackTrace();
@@ -303,12 +299,12 @@ public class MapJoinProcessor implements
* position of the source to be read as part of map-reduce framework. All other sources
* are cached in memory
* @param noCheckOuterJoin
+ * @param validateMapJoinTree
*/
- public static MapJoinOperator convertMapJoin(HiveConf conf,
+ public MapJoinOperator convertMapJoin(HiveConf conf,
LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtxMap,
- JoinOperator op, QBJoinTree joinTree, int mapJoinPos, boolean noCheckOuterJoin,
- boolean validateMapJoinTree)
- throws SemanticException {
+ JoinOperator op, boolean leftInputJoin, String[] baseSrc, List<String> mapAliases,
+ int mapJoinPos, boolean noCheckOuterJoin, boolean validateMapJoinTree) throws SemanticException {
// outer join cannot be performed on a table which is being cached
JoinDesc desc = op.getConf();
@@ -323,8 +319,6 @@ public class MapJoinProcessor implements
// Walk over all the sources (which are guaranteed to be reduce sink
// operators).
// The join outputs a concatenation of all the inputs.
- QBJoinTree leftSrc = joinTree.getJoinSrc();
-
List<Operator<? extends OperatorDesc>> parentOps = op.getParentOperators();
List<Operator<? extends OperatorDesc>> newParentOps =
new ArrayList<Operator<? extends OperatorDesc>>();
@@ -332,7 +326,7 @@ public class MapJoinProcessor implements
new ArrayList<Operator<? extends OperatorDesc>>();
// found a source which is not to be stored in memory
- if (leftSrc != null) {
+ if (leftInputJoin) {
// assert mapJoinPos == 0;
Operator<? extends OperatorDesc> parentOp = parentOps.get(0);
assert parentOp.getParentOperators().size() == 1;
@@ -344,7 +338,7 @@ public class MapJoinProcessor implements
byte pos = 0;
// Remove parent reduce-sink operators
- for (String src : joinTree.getBaseSrc()) {
+ for (String src : baseSrc) {
if (src != null) {
Operator<? extends OperatorDesc> parentOp = parentOps.get(pos);
assert parentOp.getParentOperators().size() == 1;
@@ -359,7 +353,7 @@ public class MapJoinProcessor implements
// create the map-join operator
MapJoinOperator mapJoinOp = convertJoinOpMapJoinOp(conf, opParseCtxMap,
- op, joinTree, mapJoinPos, noCheckOuterJoin);
+ op, leftInputJoin, baseSrc, mapAliases, mapJoinPos, noCheckOuterJoin);
// remove old parents
@@ -381,13 +375,14 @@ public class MapJoinProcessor implements
return mapJoinOp;
}
- static MapJoinOperator convertJoinOpMapJoinOp(HiveConf hconf,
+ public static MapJoinOperator convertJoinOpMapJoinOp(HiveConf hconf,
LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtxMap,
- JoinOperator op, QBJoinTree joinTree, int mapJoinPos, boolean noCheckOuterJoin)
- throws SemanticException {
+ JoinOperator op, boolean leftInputJoin, String[] baseSrc, List<String> mapAliases,
+ int mapJoinPos, boolean noCheckOuterJoin) throws SemanticException {
MapJoinDesc mapJoinDescriptor =
- getMapJoinDesc(hconf, opParseCtxMap, op, joinTree, mapJoinPos, noCheckOuterJoin);
+ getMapJoinDesc(hconf, opParseCtxMap, op, leftInputJoin, baseSrc, mapAliases,
+ mapJoinPos, noCheckOuterJoin);
// reduce sink row resolver used to generate map join op
RowResolver outputRS = opParseCtxMap.get(op).getRowResolver();
@@ -407,6 +402,7 @@ public class MapJoinProcessor implements
childOp.replaceParent(op, mapJoinOp);
}
+ mapJoinOp.setPosToAliasMap(op.getPosToAliasMap());
mapJoinOp.setChildOperators(childOps);
op.setChildOperators(null);
op.setParentOperators(null);
@@ -439,7 +435,7 @@ public class MapJoinProcessor implements
*/
public static MapJoinOperator convertSMBJoinToMapJoin(HiveConf hconf,
Map<Operator<? extends OperatorDesc>, OpParseContext> opParseCtxMap,
- SMBMapJoinOperator smbJoinOp, QBJoinTree joinTree, int bigTablePos, boolean noCheckOuterJoin)
+ SMBMapJoinOperator smbJoinOp, int bigTablePos, boolean noCheckOuterJoin)
throws SemanticException {
// Create a new map join operator
SMBJoinDesc smbJoinDesc = smbJoinOp.getConf();
@@ -486,7 +482,7 @@ public class MapJoinProcessor implements
}
public MapJoinOperator generateMapJoinOperator(ParseContext pctx, JoinOperator op,
- QBJoinTree joinTree, int mapJoinPos) throws SemanticException {
+ int mapJoinPos) throws SemanticException {
HiveConf hiveConf = pctx.getConf();
boolean noCheckOuterJoin = HiveConf.getBoolVar(hiveConf,
HiveConf.ConfVars.HIVEOPTSORTMERGEBUCKETMAPJOIN)
@@ -495,7 +491,8 @@ public class MapJoinProcessor implements
LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtxMap = pctx
.getOpParseCtx();
MapJoinOperator mapJoinOp = convertMapJoin(pctx.getConf(), opParseCtxMap, op,
- joinTree, mapJoinPos, noCheckOuterJoin, true);
+ op.getConf().isLeftInputJoin(), op.getConf().getBaseSrc(), op.getConf().getMapAliases(),
+ mapJoinPos, noCheckOuterJoin, true);
// create a dummy select to select all columns
genSelectPlan(pctx, mapJoinOp);
return mapJoinOp;
@@ -594,7 +591,7 @@ public class MapJoinProcessor implements
return mapJoinPos;
}
- private void genSelectPlan(ParseContext pctx, MapJoinOperator input) throws SemanticException {
+ protected void genSelectPlan(ParseContext pctx, MapJoinOperator input) throws SemanticException {
List<Operator<? extends OperatorDesc>> childOps = input.getChildOperators();
input.setChildOperators(null);
@@ -624,7 +621,7 @@ public class MapJoinProcessor implements
SelectDesc select = new SelectDesc(exprs, outputs, false);
- SelectOperator sel = (SelectOperator) putOpInsertMap(OperatorFactory.getAndMakeChild(select,
+ SelectOperator sel = (SelectOperator) putOpInsertMap(pctx, OperatorFactory.getAndMakeChild(select,
new RowSchema(inputRR.getColumnInfos()), input), inputRR);
sel.setColumnExprMap(colExprMap);
@@ -641,24 +638,22 @@ public class MapJoinProcessor implements
*
* @param op
* join operator
- * @param joinTree
- * qb join tree
* @return -1 if it cannot be converted to a map-side join, position of the map join node
* otherwise
*/
- private int mapSideJoin(JoinOperator op, QBJoinTree joinTree) throws SemanticException {
+ private int mapSideJoin(JoinOperator op) throws SemanticException {
int mapJoinPos = -1;
- if (joinTree.isMapSideJoin()) {
+ if (op.getConf().isMapSideJoin()) {
int pos = 0;
// In a map-side join, exactly one table is not present in memory.
// The client provides the list of tables which can be cached in memory
// via a hint.
- if (joinTree.getJoinSrc() != null) {
+ if (op.getConf().isLeftInputJoin()) {
mapJoinPos = pos;
}
- for (String src : joinTree.getBaseSrc()) {
+ for (String src : op.getConf().getBaseSrc()) {
if (src != null) {
- if (!joinTree.getMapAliases().contains(src)) {
+ if (!op.getConf().getMapAliases().contains(src)) {
if (mapJoinPos >= 0) {
return -1;
}
@@ -673,7 +668,7 @@ public class MapJoinProcessor implements
// leaving some table from the list of tables to be cached
if (mapJoinPos == -1) {
throw new SemanticException(ErrorMsg.INVALID_MAPJOIN_HINT.getMsg(
- Arrays.toString(joinTree.getBaseSrc())));
+ Arrays.toString(op.getConf().getBaseSrc())));
}
}
@@ -689,36 +684,34 @@ public class MapJoinProcessor implements
*/
@Override
public ParseContext transform(ParseContext pactx) throws SemanticException {
- pGraphContext = pactx;
List<MapJoinOperator> listMapJoinOps = new ArrayList<MapJoinOperator>();
// traverse all the joins and convert them if necessary
- if (pGraphContext.getJoinContext() != null) {
- Map<JoinOperator, QBJoinTree> joinMap = new HashMap<JoinOperator, QBJoinTree>();
- Map<MapJoinOperator, QBJoinTree> mapJoinMap = pGraphContext.getMapJoinContext();
+ if (pactx.getJoinOps() != null) {
+ Set<JoinOperator> joinMap = new HashSet<JoinOperator>();
+ Set<MapJoinOperator> mapJoinMap = pactx.getMapJoinOps();
if (mapJoinMap == null) {
- mapJoinMap = new HashMap<MapJoinOperator, QBJoinTree>();
- pGraphContext.setMapJoinContext(mapJoinMap);
+ mapJoinMap = new HashSet<MapJoinOperator>();
+ pactx.setMapJoinOps(mapJoinMap);
}
- Set<Map.Entry<JoinOperator, QBJoinTree>> joinCtx = pGraphContext.getJoinContext().entrySet();
- Iterator<Map.Entry<JoinOperator, QBJoinTree>> joinCtxIter = joinCtx.iterator();
+ Iterator<JoinOperator> joinCtxIter = pactx.getJoinOps().iterator();
while (joinCtxIter.hasNext()) {
- Map.Entry<JoinOperator, QBJoinTree> joinEntry = joinCtxIter.next();
- JoinOperator joinOp = joinEntry.getKey();
- QBJoinTree qbJoin = joinEntry.getValue();
- int mapJoinPos = mapSideJoin(joinOp, qbJoin);
+ JoinOperator joinOp = joinCtxIter.next();
+ int mapJoinPos = mapSideJoin(joinOp);
if (mapJoinPos >= 0) {
- MapJoinOperator mapJoinOp = generateMapJoinOperator(pactx, joinOp, qbJoin, mapJoinPos);
+ MapJoinOperator mapJoinOp = generateMapJoinOperator(pactx, joinOp, mapJoinPos);
listMapJoinOps.add(mapJoinOp);
- mapJoinMap.put(mapJoinOp, qbJoin);
+ mapJoinOp.getConf().setQBJoinTreeProps(joinOp.getConf());
+ mapJoinMap.add(mapJoinOp);
} else {
- joinMap.put(joinOp, qbJoin);
+ joinOp.getConf().setQBJoinTreeProps(joinOp.getConf());
+ joinMap.add(joinOp);
}
}
// store the new joinContext
- pGraphContext.setJoinContext(joinMap);
+ pactx.setJoinOps(joinMap);
}
// Go over the list and find if a reducer is not needed
@@ -744,15 +737,15 @@ public class MapJoinProcessor implements
// The dispatcher fires the processor corresponding to the closest matching
// rule and passes the context along
Dispatcher disp = new DefaultRuleDispatcher(getDefault(), opRules, new MapJoinWalkerCtx(
- listMapJoinOpsNoRed, pGraphContext));
+ listMapJoinOpsNoRed, pactx));
GraphWalker ogw = new GenMapRedWalker(disp);
ArrayList<Node> topNodes = new ArrayList<Node>();
topNodes.addAll(listMapJoinOps);
ogw.startWalking(topNodes, null);
- pGraphContext.setListMapJoinOpsNoReducer(listMapJoinOpsNoRed);
- return pGraphContext;
+ pactx.setListMapJoinOpsNoReducer(listMapJoinOpsNoRed);
+ return pactx;
}
/**
@@ -798,7 +791,7 @@ public class MapJoinProcessor implements
}
Operator<? extends OperatorDesc> ch = parent.getChildOperators().get(0);
if (ch instanceof MapJoinOperator) {
- if (!nonSubqueryMapJoin(ctx.getpGraphContext(), (MapJoinOperator) ch, mapJoin)) {
+ if (!nonSubqueryMapJoin((MapJoinOperator) ch, mapJoin)) {
if (ch.getParentOperators().indexOf(parent) == ((MapJoinOperator) ch).getConf()
.getPosBigTable()) {
// not come from the local branch
@@ -818,11 +811,8 @@ public class MapJoinProcessor implements
}
}
- private boolean nonSubqueryMapJoin(ParseContext pGraphContext, MapJoinOperator mapJoin,
- MapJoinOperator parentMapJoin) {
- QBJoinTree joinTree = pGraphContext.getMapJoinContext().get(mapJoin);
- QBJoinTree parentJoinTree = pGraphContext.getMapJoinContext().get(parentMapJoin);
- if (joinTree.getJoinSrc() != null && joinTree.getJoinSrc().equals(parentJoinTree)) {
+ private boolean nonSubqueryMapJoin(MapJoinOperator mapJoin, MapJoinOperator parentMapJoin) {
+ if (mapJoin.getParentOperators().contains(parentMapJoin)) {
return true;
}
return false;
@@ -1026,15 +1016,15 @@ public class MapJoinProcessor implements
}
- public static ObjectPair<List<ReduceSinkOperator>, Map<Byte, List<ExprNodeDesc>>> getKeys(QBJoinTree joinTree, JoinOperator op) {
+ public static ObjectPair<List<ReduceSinkOperator>, Map<Byte, List<ExprNodeDesc>>> getKeys(
+ boolean leftInputJoin, String[] baseSrc, JoinOperator op) {
// Walk over all the sources (which are guaranteed to be reduce sink
// operators).
// The join outputs a concatenation of all the inputs.
- QBJoinTree leftSrc = joinTree.getJoinSrc();
List<ReduceSinkOperator> oldReduceSinkParentOps =
new ArrayList<ReduceSinkOperator>(op.getNumParent());
- if (leftSrc != null) {
+ if (leftInputJoin) {
// assert mapJoinPos == 0;
Operator<? extends OperatorDesc> parentOp = op.getParentOperators().get(0);
assert parentOp.getParentOperators().size() == 1;
@@ -1042,7 +1032,7 @@ public class MapJoinProcessor implements
}
byte pos = 0;
- for (String src : joinTree.getBaseSrc()) {
+ for (String src : baseSrc) {
if (src != null) {
Operator<? extends OperatorDesc> parentOp = op.getParentOperators().get(pos);
assert parentOp.getParentOperators().size() == 1;
@@ -1060,12 +1050,14 @@ public class MapJoinProcessor implements
keyExprMap.put(pos, keyCols);
}
- return new ObjectPair<List<ReduceSinkOperator>, Map<Byte,List<ExprNodeDesc>>>(oldReduceSinkParentOps, keyExprMap);
+ return new ObjectPair<List<ReduceSinkOperator>, Map<Byte,List<ExprNodeDesc>>>(
+ oldReduceSinkParentOps, keyExprMap);
}
public static MapJoinDesc getMapJoinDesc(HiveConf hconf,
LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtxMap,
- JoinOperator op, QBJoinTree joinTree, int mapJoinPos, boolean noCheckOuterJoin) throws SemanticException {
+ JoinOperator op, boolean leftInputJoin, String[] baseSrc, List<String> mapAliases,
+ int mapJoinPos, boolean noCheckOuterJoin) throws SemanticException {
JoinDesc desc = op.getConf();
JoinCondDesc[] condns = desc.getConds();
Byte[] tagOrder = desc.getTagOrder();
@@ -1082,7 +1074,8 @@ public class MapJoinProcessor implements
Map<Byte, List<ExprNodeDesc>> valueExprs = op.getConf().getExprs();
Map<Byte, List<ExprNodeDesc>> newValueExprs = new HashMap<Byte, List<ExprNodeDesc>>();
- ObjectPair<List<ReduceSinkOperator>, Map<Byte,List<ExprNodeDesc>>> pair = getKeys(joinTree, op);
+ ObjectPair<List<ReduceSinkOperator>, Map<Byte,List<ExprNodeDesc>>> pair =
+ getKeys(leftInputJoin, baseSrc, op);
List<ReduceSinkOperator> oldReduceSinkParentOps = pair.getFirst();
for (Map.Entry<Byte, List<ExprNodeDesc>> entry : valueExprs.entrySet()) {
byte tag = entry.getKey();
@@ -1172,8 +1165,8 @@ public class MapJoinProcessor implements
// create dumpfile prefix needed to create descriptor
String dumpFilePrefix = "";
- if (joinTree.getMapAliases() != null) {
- for (String mapAlias : joinTree.getMapAliases()) {
+ if (mapAliases != null) {
+ for (String mapAlias : mapAliases) {
dumpFilePrefix = dumpFilePrefix + mapAlias;
}
dumpFilePrefix = dumpFilePrefix + "-" + PlanUtils.getCountForMapJoinDumpFilePrefix();
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java Fri Jan 23 19:59:11 2015
@@ -30,6 +30,7 @@ import java.util.Stack;
import org.apache.hadoop.hive.ql.exec.FilterOperator;
import org.apache.hadoop.hive.ql.exec.JoinOperator;
+import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.SelectOperator;
import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
@@ -42,7 +43,6 @@ import org.apache.hadoop.hive.ql.lib.Nod
import org.apache.hadoop.hive.ql.lib.Rule;
import org.apache.hadoop.hive.ql.lib.RuleRegExp;
import org.apache.hadoop.hive.ql.parse.ParseContext;
-import org.apache.hadoop.hive.ql.parse.QBJoinTree;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
@@ -55,15 +55,12 @@ import org.apache.hadoop.hive.ql.plan.Op
*/
public class NonBlockingOpDeDupProc implements Transform {
- private ParseContext pctx;
-
@Override
public ParseContext transform(ParseContext pctx) throws SemanticException {
- this.pctx = pctx;
String SEL = SelectOperator.getOperatorName();
String FIL = FilterOperator.getOperatorName();
Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
- opRules.put(new RuleRegExp("R1", SEL + "%" + SEL + "%"), new SelectDedup());
+ opRules.put(new RuleRegExp("R1", SEL + "%" + SEL + "%"), new SelectDedup(pctx));
opRules.put(new RuleRegExp("R2", FIL + "%" + FIL + "%"), new FilterDedup());
Dispatcher disp = new DefaultRuleDispatcher(null, opRules, null);
@@ -76,6 +73,13 @@ public class NonBlockingOpDeDupProc impl
}
private class SelectDedup implements NodeProcessor {
+
+ private ParseContext pctx;
+
+ public SelectDedup (ParseContext pctx) {
+ this.pctx = pctx;
+ }
+
@Override
public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
Object... nodeOutputs) throws SemanticException {
@@ -178,28 +182,37 @@ public class NonBlockingOpDeDupProc impl
}
return true;
}
- }
- /**
- * Change existing references in the context to point from child to parent operator.
- * @param cSEL child operator (to be removed, and merged into parent)
- * @param pSEL parent operator
- */
- private void fixContextReferences(SelectOperator cSEL, SelectOperator pSEL) {
- Collection<QBJoinTree> qbJoinTrees = new ArrayList<QBJoinTree>();
- qbJoinTrees.addAll(pctx.getJoinContext().values());
- qbJoinTrees.addAll(pctx.getMapJoinContext().values());
- for (QBJoinTree qbJoinTree : qbJoinTrees) {
- Map<String, Operator<? extends OperatorDesc>> aliasToOpInfo = qbJoinTree.getAliasToOpInfo();
- for (Map.Entry<String, Operator<? extends OperatorDesc>> entry : aliasToOpInfo.entrySet()) {
- if (entry.getValue() == cSEL) {
- aliasToOpInfo.put(entry.getKey(), pSEL);
+ /**
+ * Change existing references in the context to point from child to parent operator.
+ * @param cSEL child operator (to be removed, and merged into parent)
+ * @param pSEL parent operator
+ */
+ private void fixContextReferences(SelectOperator cSEL, SelectOperator pSEL) {
+ Collection<Map<String, Operator<? extends OperatorDesc>>> mapsAliasToOpInfo =
+ new ArrayList<Map<String, Operator<? extends OperatorDesc>>>();
+ for (JoinOperator joinOp : pctx.getJoinOps()) {
+ if (joinOp.getConf().getAliasToOpInfo() != null) {
+ mapsAliasToOpInfo.add(joinOp.getConf().getAliasToOpInfo());
+ }
+ }
+ for (MapJoinOperator mapJoinOp : pctx.getMapJoinOps()) {
+ if (mapJoinOp.getConf().getAliasToOpInfo() != null) {
+ mapsAliasToOpInfo.add(mapJoinOp.getConf().getAliasToOpInfo());
+ }
+ }
+ for (Map<String, Operator<? extends OperatorDesc>> aliasToOpInfo : mapsAliasToOpInfo) {
+ for (Map.Entry<String, Operator<? extends OperatorDesc>> entry : aliasToOpInfo.entrySet()) {
+ if (entry.getValue() == cSEL) {
+ aliasToOpInfo.put(entry.getKey(), pSEL);
+ }
}
}
}
}
private class FilterDedup implements NodeProcessor {
+
@Override
public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
Object... nodeOutputs) throws SemanticException {
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java Fri Jan 23 19:59:11 2015
@@ -56,6 +56,7 @@ public class Optimizer {
public void initialize(HiveConf hiveConf) {
boolean isTezExecEngine = HiveConf.getVar(hiveConf, HiveConf.ConfVars.HIVE_EXECUTION_ENGINE).equals("tez");
+ boolean isSparkExecEngine = HiveConf.getVar(hiveConf, HiveConf.ConfVars.HIVE_EXECUTION_ENGINE).equals("spark");
boolean bucketMapJoinOptimizer = false;
transformations = new ArrayList<Transform>();
@@ -100,9 +101,13 @@ public class Optimizer {
transformations.add(new RewriteGBUsingIndex());
}
transformations.add(new SamplePruner());
- transformations.add(new MapJoinProcessor());
- if ((HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTBUCKETMAPJOIN)) && !isTezExecEngine) {
+ MapJoinProcessor mapJoinProcessor = isSparkExecEngine ? new SparkMapJoinProcessor()
+ : new MapJoinProcessor();
+ transformations.add(mapJoinProcessor);
+
+ if ((HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTBUCKETMAPJOIN))
+ && !isTezExecEngine && !isSparkExecEngine) {
transformations.add(new BucketMapJoinOptimizer());
bucketMapJoinOptimizer = true;
}
@@ -110,7 +115,7 @@ public class Optimizer {
// If optimize hive.optimize.bucketmapjoin.sortedmerge is set, add both
// BucketMapJoinOptimizer and SortedMergeBucketMapJoinOptimizer
if ((HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTSORTMERGEBUCKETMAPJOIN))
- && !isTezExecEngine) {
+ && !isTezExecEngine && !isSparkExecEngine) {
if (!bucketMapJoinOptimizer) {
// No need to add BucketMapJoinOptimizer twice
transformations.add(new BucketMapJoinOptimizer());
@@ -152,7 +157,7 @@ public class Optimizer {
if(HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTIMIZEMETADATAQUERIES)) {
transformations.add(new StatsOptimizer());
}
- if (pctx.getContext().getExplain() && !isTezExecEngine) {
+ if (isSparkExecEngine || (pctx.getContext().getExplain() && !isTezExecEngine)) {
transformations.add(new AnnotateWithStatistics());
transformations.add(new AnnotateWithOpTraits());
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchOptimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchOptimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchOptimizer.java Fri Jan 23 19:59:11 2015
@@ -143,8 +143,15 @@ public class SimpleFetchOptimizer implem
}
private boolean checkThreshold(FetchData data, int limit, ParseContext pctx) throws Exception {
- if (limit > 0 && data.hasOnlyPruningFilter()) {
- return true;
+ if (limit > 0) {
+ if (data.hasOnlyPruningFilter()) {
+ /* partitioned table + query has only pruning filters */
+ return true;
+ } else if (data.isPartitioned() == false && data.isFiltered() == false) {
+ /* unpartitioned table + no filters */
+ return true;
+ }
+ /* fall through */
}
long threshold = HiveConf.getLongVar(pctx.getConf(),
HiveConf.ConfVars.HIVEFETCHTASKCONVERSIONTHRESHOLD);
@@ -176,7 +183,7 @@ public class SimpleFetchOptimizer implem
if (!aggressive && qb.hasTableSample(alias)) {
return null;
}
- Table table = pctx.getTopToTable().get(ts);
+ Table table = ts.getConf().getTableMetadata();
if (table == null) {
return null;
}
@@ -228,6 +235,10 @@ public class SimpleFetchOptimizer implem
if (op.getChildOperators() == null || op.getChildOperators().size() != 1) {
return null;
}
+
+ if (op instanceof FilterOperator) {
+ fetch.setFiltered(true);
+ }
}
if (op instanceof FileSinkOperator) {
@@ -279,6 +290,11 @@ public class SimpleFetchOptimizer implem
|| operator instanceof ScriptOperator) {
return false;
}
+
+ if (operator instanceof FilterOperator) {
+ fetch.setFiltered(true);
+ }
+
if (!traversed.add(operator)) {
return true;
}
@@ -314,6 +330,7 @@ public class SimpleFetchOptimizer implem
// this is always non-null when conversion is completed
private Operator<?> fileSink;
+ private boolean filtered;
private FetchData(TableScanOperator scanOp, ReadEntity parent, Table table, SplitSample splitSample) {
this.scanOp = scanOp;
@@ -337,10 +354,23 @@ public class SimpleFetchOptimizer implem
/*
* all filters were executed during partition pruning
*/
- public boolean hasOnlyPruningFilter() {
+ public final boolean hasOnlyPruningFilter() {
return this.onlyPruningFilter;
}
+ public final boolean isPartitioned() {
+ return this.table.isPartitioned();
+ }
+
+ /* there are filter operators in the pipeline */
+ public final boolean isFiltered() {
+ return this.filtered;
+ }
+
+ public final void setFiltered(boolean filtered) {
+ this.filtered = filtered;
+ }
+
private FetchWork convertToWork() throws HiveException {
inputs.clear();
if (!table.isPartitioned()) {
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SkewJoinOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SkewJoinOptimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SkewJoinOptimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SkewJoinOptimizer.java Fri Jan 23 19:59:11 2015
@@ -79,11 +79,13 @@ import org.apache.hadoop.hive.serde2.typ
public class SkewJoinOptimizer implements Transform {
private static final Log LOG = LogFactory.getLog(SkewJoinOptimizer.class.getName());
- private static ParseContext parseContext;
public static class SkewJoinProc implements NodeProcessor {
- public SkewJoinProc() {
+ private ParseContext parseContext;
+
+ public SkewJoinProc(ParseContext parseContext) {
super();
+ this.parseContext = parseContext;
}
@Override
@@ -165,23 +167,14 @@ public class SkewJoinOptimizer implement
return null;
}
- // have to create a QBJoinTree for the cloned join operator
- QBJoinTree originJoinTree = parseContext.getJoinContext().get(joinOp);
- QBJoinTree newJoinTree;
- try {
- newJoinTree = originJoinTree.clone();
- } catch (CloneNotSupportedException e) {
- LOG.debug("QBJoinTree could not be cloned: ", e);
- return null;
- }
-
JoinOperator joinOpClone;
if (processSelect) {
joinOpClone = (JoinOperator)(currOpClone.getParentOperators().get(0));
} else {
joinOpClone = (JoinOperator)currOpClone;
}
- parseContext.getJoinContext().put(joinOpClone, newJoinTree);
+ joinOpClone.getConf().cloneQBJoinTreeProps(joinOp.getConf());
+ parseContext.getJoinOps().add(joinOpClone);
List<TableScanOperator> tableScanCloneOpsForJoin =
new ArrayList<TableScanOperator>();
@@ -202,9 +195,7 @@ public class SkewJoinOptimizer implement
for (Entry<String, Operator<? extends OperatorDesc>> topOp : topOps.entrySet()) {
TableScanOperator tso = (TableScanOperator) topOp.getValue();
- Table origTable = parseContext.getTopToTable().get(ctx.getCloneTSOpMap().get(tso));
String tabAlias = tso.getConf().getAlias();
- parseContext.getTopToTable().put(tso, origTable);
int initCnt = 1;
String newAlias = "subquery" + initCnt + ":" + tabAlias;
while (origTopOps.containsKey(newAlias)) {
@@ -213,7 +204,7 @@ public class SkewJoinOptimizer implement
}
parseContext.getTopOps().put(newAlias, tso);
- setUpAlias(originJoinTree, newJoinTree, tabAlias, newAlias, tso);
+ setUpAlias(joinOp, joinOpClone, tabAlias, newAlias, tso);
}
// Now do a union of the select operators: selectOp and selectOpClone
@@ -410,7 +401,7 @@ public class SkewJoinOptimizer implement
if (op instanceof TableScanOperator) {
TableScanOperator tsOp = (TableScanOperator)op;
if (tableScanOpsForJoin.contains(tsOp)) {
- return parseContext.getTopToTable().get(tsOp);
+ return tsOp.getConf().getTableMetadata();
}
}
if ((op.getParentOperators() == null) || (op.getParentOperators().isEmpty()) ||
@@ -629,19 +620,19 @@ public class SkewJoinOptimizer implement
/**
* Set alias in the cloned join tree
*/
- private static void setUpAlias(QBJoinTree origin, QBJoinTree cloned, String origAlias,
+ private static void setUpAlias(JoinOperator origin, JoinOperator cloned, String origAlias,
String newAlias, Operator<? extends OperatorDesc> topOp) {
- cloned.getAliasToOpInfo().remove(origAlias);
- cloned.getAliasToOpInfo().put(newAlias, topOp);
- if (origin.getLeftAlias().equals(origAlias)) {
- cloned.setLeftAlias(null);
- cloned.setLeftAlias(newAlias);
- }
- replaceAlias(origin.getLeftAliases(), cloned.getLeftAliases(), origAlias, newAlias);
- replaceAlias(origin.getRightAliases(), cloned.getRightAliases(), origAlias, newAlias);
- replaceAlias(origin.getBaseSrc(), cloned.getBaseSrc(), origAlias, newAlias);
- replaceAlias(origin.getMapAliases(), cloned.getMapAliases(), origAlias, newAlias);
- replaceAlias(origin.getStreamAliases(), cloned.getStreamAliases(), origAlias, newAlias);
+ cloned.getConf().getAliasToOpInfo().remove(origAlias);
+ cloned.getConf().getAliasToOpInfo().put(newAlias, topOp);
+ if (origin.getConf().getLeftAlias().equals(origAlias)) {
+ cloned.getConf().setLeftAlias(null);
+ cloned.getConf().setLeftAlias(newAlias);
+ }
+ replaceAlias(origin.getConf().getLeftAliases(), cloned.getConf().getLeftAliases(), origAlias, newAlias);
+ replaceAlias(origin.getConf().getRightAliases(), cloned.getConf().getRightAliases(), origAlias, newAlias);
+ replaceAlias(origin.getConf().getBaseSrc(), cloned.getConf().getBaseSrc(), origAlias, newAlias);
+ replaceAlias(origin.getConf().getMapAliases(), cloned.getConf().getMapAliases(), origAlias, newAlias);
+ replaceAlias(origin.getConf().getStreamAliases(), cloned.getConf().getStreamAliases(), origAlias, newAlias);
}
private static void replaceAlias(String[] origin, String[] cloned,
@@ -677,7 +668,7 @@ public class SkewJoinOptimizer implement
public ParseContext transform(ParseContext pctx) throws SemanticException {
Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
- opRules.put(new RuleRegExp("R1", "TS%.*RS%JOIN%"), getSkewJoinProc());
+ opRules.put(new RuleRegExp("R1", "TS%.*RS%JOIN%"), getSkewJoinProc(pctx));
SkewJoinOptProcCtx skewJoinOptProcCtx = new SkewJoinOptProcCtx(pctx);
// The dispatcher fires the processor corresponding to the closest matching
// rule and passes the context along
@@ -692,8 +683,8 @@ public class SkewJoinOptimizer implement
return pctx;
}
- private NodeProcessor getSkewJoinProc() {
- return new SkewJoinProc();
+ private NodeProcessor getSkewJoinProc(ParseContext parseContext) {
+ return new SkewJoinProc(parseContext);
}
/**
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionOptimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionOptimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionOptimizer.java Fri Jan 23 19:59:11 2015
@@ -143,7 +143,7 @@ public class SortedDynPartitionOptimizer
return null;
}
- Table destTable = parseCtx.getFsopToTable().get(fsOp);
+ Table destTable = fsOp.getConf().getTable();
if (destTable == null) {
LOG.debug("Bailing out of sort dynamic partition optimization as destination table is null");
return null;
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeBucketMapjoinProc.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeBucketMapjoinProc.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeBucketMapjoinProc.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeBucketMapjoinProc.java Fri Jan 23 19:59:11 2015
@@ -60,7 +60,7 @@ public class SortedMergeBucketMapjoinPro
}
if (convert) {
- convertBucketMapJoinToSMBJoin(mapJoinOp, smbJoinContext, pGraphContext);
+ convertBucketMapJoinToSMBJoin(mapJoinOp, smbJoinContext);
}
return null;
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeJoinProc.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeJoinProc.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeJoinProc.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeJoinProc.java Fri Jan 23 19:59:11 2015
@@ -44,10 +44,10 @@ public class SortedMergeJoinProc extends
SortBucketJoinProcCtx smbJoinContext = (SortBucketJoinProcCtx) procCtx;
boolean convert =
canConvertJoinToSMBJoin(
- joinOp, smbJoinContext, pGraphContext);
+ joinOp, smbJoinContext);
if (convert) {
- convertJoinToSMBJoin(joinOp, smbJoinContext, pGraphContext);
+ convertJoinToSMBJoin(joinOp, smbJoinContext);
}
return null;
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java Fri Jan 23 19:59:11 2015
@@ -228,7 +228,7 @@ public class StatsOptimizer implements T
return null;
}
- Table tbl = pctx.getTopToTable().get(tsOp);
+ Table tbl = tsOp.getConf().getTableMetadata();
List<Object> oneRow = new ArrayList<Object>();
List<ObjectInspector> ois = new ArrayList<ObjectInspector>();
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/TableSizeBasedBigTableSelectorForAutoSMJ.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/TableSizeBasedBigTableSelectorForAutoSMJ.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/TableSizeBasedBigTableSelectorForAutoSMJ.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/TableSizeBasedBigTableSelectorForAutoSMJ.java Fri Jan 23 19:59:11 2015
@@ -60,7 +60,7 @@ implements BigTableSelectorForAutoSMJ {
currentPos++;
continue;
}
- Table table = parseCtx.getTopToTable().get(topOp);
+ Table table = topOp.getConf().getTableMetadata();
long currentSize = 0;
if (!table.isPartitioned()) {
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java Fri Jan 23 19:59:11 2015
@@ -273,8 +273,8 @@ public class RelOptHiveTable extends Rel
}
} catch (HiveException e) {
String logMsg = "Collecting stats failed.";
- LOG.error(logMsg);
- throw new RuntimeException(logMsg);
+ LOG.error(logMsg, e);
+ throw new RuntimeException(logMsg, e);
}
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ASTBuilder.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ASTBuilder.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ASTBuilder.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ASTBuilder.java Fri Jan 23 19:59:11 2015
@@ -21,6 +21,7 @@ import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Calendar;
+import org.apache.calcite.avatica.util.ByteString;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rex.RexLiteral;
@@ -144,7 +145,8 @@ class ASTBuilder {
switch (sqlType) {
case BINARY:
- val = literal.byteValue();
+ ByteString bs = (ByteString) literal.getValue();
+ val = bs.byteAt(0);
type = HiveParser.BigintLiteral;
break;
case TINYINT:
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ASTConverter.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ASTConverter.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ASTConverter.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ASTConverter.java Fri Jan 23 19:59:11 2015
@@ -27,6 +27,7 @@ import org.apache.calcite.rel.RelFieldCo
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelVisitor;
import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.Aggregate.Group;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.Filter;
import org.apache.calcite.rel.core.Join;
@@ -49,12 +50,14 @@ import org.apache.calcite.rex.RexWindowB
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql.type.SqlTypeName;
-import org.apache.calcite.util.BitSets;
+import org.apache.calcite.util.ImmutableBitSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
import org.apache.hadoop.hive.ql.optimizer.calcite.CalciteSemanticException;
import org.apache.hadoop.hive.ql.optimizer.calcite.RelOptHiveTable;
+import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveGroupingID;
import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveSort;
import org.apache.hadoop.hive.ql.optimizer.calcite.translator.SqlFunctionConverter.HiveToken;
import org.apache.hadoop.hive.ql.parse.ASTNode;
@@ -93,7 +96,7 @@ public class ASTConverter {
return c.convert();
}
- private ASTNode convert() {
+ private ASTNode convert() throws CalciteSemanticException {
/*
* 1. Walk RelNode Graph; note from, where, gBy.. nodes.
*/
@@ -118,15 +121,50 @@ public class ASTConverter {
* 4. GBy
*/
if (groupBy != null) {
- ASTBuilder b = ASTBuilder.construct(HiveParser.TOK_GROUPBY, "TOK_GROUPBY");
- for (int i : BitSets.toIter(groupBy.getGroupSet())) {
+ ASTBuilder b;
+ boolean groupingSetsExpression = false;
+ if (groupBy.indicator) {
+ Group aggregateType = Aggregate.Group.induce(groupBy.getGroupSet(),
+ groupBy.getGroupSets());
+ if (aggregateType == Group.ROLLUP) {
+ b = ASTBuilder.construct(HiveParser.TOK_ROLLUP_GROUPBY, "TOK_ROLLUP_GROUPBY");
+ }
+ else if (aggregateType == Group.CUBE) {
+ b = ASTBuilder.construct(HiveParser.TOK_CUBE_GROUPBY, "TOK_CUBE_GROUPBY");
+ }
+ else {
+ b = ASTBuilder.construct(HiveParser.TOK_GROUPING_SETS, "TOK_GROUPING_SETS");
+ groupingSetsExpression = true;
+ }
+ }
+ else {
+ b = ASTBuilder.construct(HiveParser.TOK_GROUPBY, "TOK_GROUPBY");
+ }
+
+ for (int i : groupBy.getGroupSet()) {
RexInputRef iRef = new RexInputRef(i, groupBy.getCluster().getTypeFactory()
.createSqlType(SqlTypeName.ANY));
b.add(iRef.accept(new RexVisitor(schema)));
}
- if (!groupBy.getGroupSet().isEmpty())
+ //Grouping sets expressions
+ if(groupingSetsExpression) {
+ for(ImmutableBitSet groupSet: groupBy.getGroupSets()) {
+ ASTBuilder expression = ASTBuilder.construct(
+ HiveParser.TOK_GROUPING_SETS_EXPRESSION, "TOK_GROUPING_SETS_EXPRESSION");
+ for (int i : groupSet) {
+ RexInputRef iRef = new RexInputRef(i, groupBy.getCluster().getTypeFactory()
+ .createSqlType(SqlTypeName.ANY));
+ expression.add(iRef.accept(new RexVisitor(schema)));
+ }
+ b.add(expression);
+ }
+ }
+
+ if (!groupBy.getGroupSet().isEmpty()) {
hiveAST.groupBy = b.node();
+ }
+
schema = new Schema(schema, groupBy);
}
@@ -151,9 +189,33 @@ public class ASTConverter {
int i = 0;
for (RexNode r : select.getChildExps()) {
- ASTNode selectExpr = ASTBuilder.selectExpr(r.accept(
- new RexVisitor(schema, r instanceof RexLiteral)),
- select.getRowType().getFieldNames().get(i++));
+ // If it is a GroupBy with grouping sets and grouping__id column
+ // is selected, we reformulate to project that column from
+ // the output of the GroupBy operator
+ boolean reformulate = false;
+ if (groupBy != null && groupBy.indicator) {
+ RexNode expr = select.getChildExps().get(i);
+ if (expr instanceof RexCall) {
+ if ( ((RexCall) expr).getOperator().
+ equals(HiveGroupingID.GROUPING__ID)) {
+ reformulate = true;
+ }
+ }
+ }
+ ASTNode expr;
+ if(reformulate) {
+ RexInputRef iRef = new RexInputRef(
+ groupBy.getGroupCount() * 2 + groupBy.getAggCallList().size(),
+ TypeConverter.convert(
+ VirtualColumn.GROUPINGID.getTypeInfo(),
+ groupBy.getCluster().getTypeFactory()));
+ expr = iRef.accept(new RexVisitor(schema));
+ }
+ else {
+ expr = r.accept(new RexVisitor(schema, r instanceof RexLiteral));
+ }
+ String alias = select.getRowType().getFieldNames().get(i++);
+ ASTNode selectExpr = ASTBuilder.selectExpr(expr, alias);
b.add(selectExpr);
}
}
@@ -232,7 +294,7 @@ public class ASTConverter {
return new Schema(select, tblAlias);
}
- private QueryBlockInfo convertSource(RelNode r) {
+ private QueryBlockInfo convertSource(RelNode r) throws CalciteSemanticException {
Schema s;
ASTNode ast;
@@ -554,10 +616,19 @@ public class ASTConverter {
}
Schema(Schema src, Aggregate gBy) {
- for (int i : BitSets.toIter(gBy.getGroupSet())) {
+ for (int i : gBy.getGroupSet()) {
ColumnInfo cI = src.get(i);
add(cI);
}
+ // If we are using grouping sets, we add the
+ // fields again, these correspond to the boolean
+ // grouping in Calcite. They are not used by Hive.
+ if(gBy.indicator) {
+ for (int i : gBy.getGroupSet()) {
+ ColumnInfo cI = src.get(i);
+ add(cI);
+ }
+ }
List<AggregateCall> aggs = gBy.getAggCallList();
for (AggregateCall agg : aggs) {
int argCount = agg.getArgList().size();
@@ -572,6 +643,9 @@ public class ASTConverter {
}
add(new ColumnInfo(null, b.node()));
}
+ if(gBy.indicator) {
+ add(new ColumnInfo(null,VirtualColumn.GROUPINGID.getName()));
+ }
}
/**
@@ -665,4 +739,5 @@ public class ASTConverter {
return flat;
}
+
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/CorrelationOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/CorrelationOptimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/CorrelationOptimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/CorrelationOptimizer.java Fri Jan 23 19:59:11 2015
@@ -107,7 +107,7 @@ public class CorrelationOptimizer implem
// that has both intermediate tables and query input tables as input tables,
// we should be able to guess if this JoinOperator will be converted to a MapJoin
// based on hive.auto.convert.join.noconditionaltask.size.
- for (JoinOperator joinOp: pCtx.getJoinContext().keySet()) {
+ for (JoinOperator joinOp: pCtx.getJoinOps()) {
boolean isAbleToGuess = true;
boolean mayConvert = false;
// Get total size and individual alias's size
@@ -124,7 +124,7 @@ public class CorrelationOptimizer implem
Set<String> aliases = new LinkedHashSet<String>();
for (TableScanOperator tsop : topOps) {
- Table table = pCtx.getTopToTable().get(tsop);
+ Table table = tsop.getConf().getTableMetadata();
if (table == null) {
// table should not be null.
throw new SemanticException("The table of " +
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/ReduceSinkDeDuplication.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/ReduceSinkDeDuplication.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/ReduceSinkDeDuplication.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/ReduceSinkDeDuplication.java Fri Jan 23 19:59:11 2015
@@ -489,6 +489,7 @@ public class ReduceSinkDeDuplication imp
if (pRS != null && merge(cRS, pRS, dedupCtx.minReducer())) {
CorrelationUtilities.replaceReduceSinkWithSelectOperator(
cRS, dedupCtx.getPctx(), dedupCtx);
+ pRS.getConf().setEnforceSort(true);
return true;
}
return false;
@@ -511,6 +512,7 @@ public class ReduceSinkDeDuplication imp
if (pRS != null && merge(cRS, pRS, dedupCtx.minReducer())) {
CorrelationUtilities.removeReduceSinkForGroupBy(
cRS, cGBY, dedupCtx.getPctx(), dedupCtx);
+ pRS.getConf().setEnforceSort(true);
return true;
}
return false;
@@ -529,6 +531,12 @@ public class ReduceSinkDeDuplication imp
pJoin.getConf().setFixedAsSorted(true);
CorrelationUtilities.replaceReduceSinkWithSelectOperator(
cRS, dedupCtx.getPctx(), dedupCtx);
+ ReduceSinkOperator pRS =
+ CorrelationUtilities.findPossibleParent(
+ pJoin, ReduceSinkOperator.class, dedupCtx.trustScript());
+ if (pRS != null) {
+ pRS.getConf().setEnforceSort(true);
+ }
return true;
}
return false;
@@ -547,6 +555,12 @@ public class ReduceSinkDeDuplication imp
pJoin.getConf().setFixedAsSorted(true);
CorrelationUtilities.removeReduceSinkForGroupBy(
cRS, cGBY, dedupCtx.getPctx(), dedupCtx);
+ ReduceSinkOperator pRS =
+ CorrelationUtilities.findPossibleParent(
+ pJoin, ReduceSinkOperator.class, dedupCtx.trustScript());
+ if (pRS != null) {
+ pRS.getConf().setEnforceSort(true);
+ }
return true;
}
return false;
@@ -565,6 +579,7 @@ public class ReduceSinkDeDuplication imp
if (pRS != null && merge(cRS, pRS, dedupCtx.minReducer())) {
CorrelationUtilities.replaceReduceSinkWithSelectOperator(
cRS, dedupCtx.getPctx(), dedupCtx);
+ pRS.getConf().setEnforceSort(true);
return true;
}
return false;
@@ -581,6 +596,7 @@ public class ReduceSinkDeDuplication imp
start, ReduceSinkOperator.class, dedupCtx.trustScript());
if (pRS != null && merge(cRS, pRS, dedupCtx.minReducer())) {
CorrelationUtilities.removeReduceSinkForGroupBy(cRS, cGBY, dedupCtx.getPctx(), dedupCtx);
+ pRS.getConf().setEnforceSort(true);
return true;
}
return false;
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/index/RewriteGBUsingIndex.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/index/RewriteGBUsingIndex.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/index/RewriteGBUsingIndex.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/index/RewriteGBUsingIndex.java Fri Jan 23 19:59:11 2015
@@ -161,11 +161,10 @@ public class RewriteGBUsingIndex impleme
* if the optimization can be applied. If yes, we add the name of the top table to
* the tsOpToProcess to apply rewrite later on.
* */
- Map<TableScanOperator, Table> topToTable = parseContext.getTopToTable();
for (Map.Entry<String, Operator<?>> entry : parseContext.getTopOps().entrySet()) {
String alias = entry.getKey();
TableScanOperator topOp = (TableScanOperator) entry.getValue();
- Table table = topToTable.get(topOp);
+ Table table = topOp.getConf().getTableMetadata();
List<Index> indexes = tableToIndex.get(table);
if (indexes.isEmpty()) {
continue;
@@ -232,12 +231,16 @@ public class RewriteGBUsingIndex impleme
supportedIndexes.add(AggregateIndexHandler.class.getName());
// query the metastore to know what columns we have indexed
- Collection<Table> topTables = parseContext.getTopToTable().values();
+ Collection<Operator<? extends OperatorDesc>> topTables = parseContext.getTopOps().values();
Map<Table, List<Index>> indexes = new HashMap<Table, List<Index>>();
- for (Table tbl : topTables){
- List<Index> tblIndexes = IndexUtils.getIndexes(tbl, supportedIndexes);
- if (tblIndexes.size() > 0) {
- indexes.put(tbl, tblIndexes);
+ for (Operator<? extends OperatorDesc> op : topTables) {
+ if (op instanceof TableScanOperator) {
+ TableScanOperator tsOP = (TableScanOperator) op;
+ List<Index> tblIndexes = IndexUtils.getIndexes(tsOP.getConf().getTableMetadata(),
+ supportedIndexes);
+ if (tblIndexes.size() > 0) {
+ indexes.put(tsOP.getConf().getTableMetadata(), tblIndexes);
+ }
}
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/index/RewriteQueryUsingAggregateIndexCtx.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/index/RewriteQueryUsingAggregateIndexCtx.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/index/RewriteQueryUsingAggregateIndexCtx.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/index/RewriteQueryUsingAggregateIndexCtx.java Fri Jan 23 19:59:11 2015
@@ -170,7 +170,6 @@ public final class RewriteQueryUsingAggr
// Need to remove the original TableScanOperators from these data structures
// and add new ones
- Map<TableScanOperator, Table> topToTable = rewriteQueryCtx.getParseContext().getTopToTable();
Map<String, Operator<? extends OperatorDesc>> topOps = rewriteQueryCtx.getParseContext()
.getTopOps();
Map<Operator<? extends OperatorDesc>, OpParseContext> opParseContext = rewriteQueryCtx
@@ -181,13 +180,8 @@ public final class RewriteQueryUsingAggr
// remove original TableScanOperator
topOps.remove(alias);
- topToTable.remove(scanOperator);
opParseContext.remove(scanOperator);
- // construct a new descriptor for the index table scan
- TableScanDesc indexTableScanDesc = new TableScanDesc();
- indexTableScanDesc.setGatherStats(false);
-
String indexTableName = rewriteQueryCtx.getIndexName();
Table indexTableHandle = null;
try {
@@ -198,6 +192,10 @@ public final class RewriteQueryUsingAggr
throw new SemanticException(e.getMessage(), e);
}
+ // construct a new descriptor for the index table scan
+ TableScanDesc indexTableScanDesc = new TableScanDesc(indexTableHandle);
+ indexTableScanDesc.setGatherStats(false);
+
String k = indexTableName + Path.SEPARATOR;
indexTableScanDesc.setStatsAggPrefix(k);
scanOperator.setConf(indexTableScanDesc);
@@ -227,12 +225,10 @@ public final class RewriteQueryUsingAggr
}
// Scan operator now points to other table
- topToTable.put(scanOperator, indexTableHandle);
scanOperator.getConf().setAlias(newAlias);
scanOperator.setAlias(indexTableName);
topOps.put(newAlias, scanOperator);
opParseContext.put(scanOperator, operatorContext);
- rewriteQueryCtx.getParseContext().setTopToTable((HashMap<TableScanOperator, Table>) topToTable);
rewriteQueryCtx.getParseContext().setTopOps(
(HashMap<String, Operator<? extends OperatorDesc>>) topOps);
rewriteQueryCtx.getParseContext().setOpParseCtx(
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/OpProcFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/OpProcFactory.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/OpProcFactory.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/OpProcFactory.java Fri Jan 23 19:59:11 2015
@@ -139,7 +139,7 @@ public class OpProcFactory {
// Table scan operator.
TableScanOperator top = (TableScanOperator)nd;
- org.apache.hadoop.hive.ql.metadata.Table t = pctx.getTopToTable().get(top);
+ org.apache.hadoop.hive.ql.metadata.Table t = top.getConf().getTableMetadata();
Table tab = t.getTTable();
// Generate the mappings
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/listbucketingpruner/LBPartitionProcFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/listbucketingpruner/LBPartitionProcFactory.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/listbucketingpruner/LBPartitionProcFactory.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/listbucketingpruner/LBPartitionProcFactory.java Fri Jan 23 19:59:11 2015
@@ -52,7 +52,7 @@ public class LBPartitionProcFactory exte
protected void generatePredicate(NodeProcessorCtx procCtx, FilterOperator fop,
TableScanOperator top) throws SemanticException, UDFArgumentException {
LBOpPartitionWalkerCtx owc = (LBOpPartitionWalkerCtx) procCtx;
- Table tbl = owc.getParseContext().getTopToTable().get(top);
+ Table tbl = top.getConf().getTableMetadata();
if (tbl.isPartitioned()) {
// Run partition pruner to get partitions
ParseContext parseCtx = owc.getParseContext();
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/metainfo/annotation/OpTraitsRulesProcFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/metainfo/annotation/OpTraitsRulesProcFactory.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/metainfo/annotation/OpTraitsRulesProcFactory.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/metainfo/annotation/OpTraitsRulesProcFactory.java Fri Jan 23 19:59:11 2015
@@ -161,7 +161,7 @@ public class OpTraitsRulesProcFactory {
Object... nodeOutputs) throws SemanticException {
TableScanOperator ts = (TableScanOperator)nd;
AnnotateOpTraitsProcCtx opTraitsCtx = (AnnotateOpTraitsProcCtx)procCtx;
- Table table = opTraitsCtx.getParseContext().getTopToTable().get(ts);
+ Table table = ts.getConf().getTableMetadata();
PrunedPartitionList prunedPartList = null;
try {
prunedPartList =
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/AbstractJoinTaskDispatcher.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/AbstractJoinTaskDispatcher.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/AbstractJoinTaskDispatcher.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/AbstractJoinTaskDispatcher.java Fri Jan 23 19:59:11 2015
@@ -34,7 +34,6 @@ import org.apache.hadoop.hive.ql.lib.Dis
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.lib.TaskGraphWalker.TaskGraphWalkerContext;
import org.apache.hadoop.hive.ql.parse.SemanticException;
-import org.apache.hadoop.hive.ql.plan.MapredWork;
import org.apache.hadoop.hive.ql.plan.MapWork;
/**
@@ -53,8 +52,7 @@ public abstract class AbstractJoinTaskDi
throws SemanticException;
protected void replaceTaskWithConditionalTask(
- Task<? extends Serializable> currTask, ConditionalTask cndTsk,
- PhysicalContext physicalContext) {
+ Task<? extends Serializable> currTask, ConditionalTask cndTsk) {
// add this task into task tree
// set all parent tasks
List<Task<? extends Serializable>> parentTasks = currTask.getParentTasks();
@@ -88,8 +86,7 @@ public abstract class AbstractJoinTaskDi
// Replace the task with the new task. Copy the children and parents of the old
// task to the new task.
protected void replaceTask(
- Task<? extends Serializable> currTask, Task<? extends Serializable> newTask,
- PhysicalContext physicalContext) {
+ Task<? extends Serializable> currTask, Task<? extends Serializable> newTask) {
// add this task into task tree
// set all parent tasks
List<Task<? extends Serializable>> parentTasks = currTask.getParentTasks();
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java Fri Jan 23 19:59:11 2015
@@ -103,7 +103,7 @@ public class BucketingSortingInferenceOp
Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
opRules.put(new RuleRegExp("R1", SelectOperator.getOperatorName() + "%"),
BucketingSortingOpProcFactory.getSelProc());
- // Matches only GroupByOpeartors which are reducers, rather than map group by operators,
+ // Matches only GroupByOperators which are reducers, rather than map group by operators,
// or multi group by optimization specific operators
opRules.put(new RuleExactMatch("R2", GroupByOperator.getOperatorName() + "%"),
BucketingSortingOpProcFactory.getGroupByProc());
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java Fri Jan 23 19:59:11 2015
@@ -45,6 +45,7 @@ import org.apache.hadoop.hive.ql.parse.S
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDescUtils;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
import org.apache.hadoop.hive.ql.plan.JoinDesc;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc;
@@ -669,7 +670,7 @@ public class BucketingSortingOpProcFacto
processGroupByReduceSink((ReduceSinkOperator) rop, gop, bctx);
- return processGroupBy((ReduceSinkOperator)rop , gop, bctx);
+ return processGroupBy(rop , gop, bctx);
}
/**
@@ -683,12 +684,16 @@ public class BucketingSortingOpProcFacto
protected void processGroupByReduceSink(ReduceSinkOperator rop, GroupByOperator gop,
BucketingSortingCtx bctx){
+ GroupByDesc groupByDesc = gop.getConf();
String sortOrder = rop.getConf().getOrder();
List<BucketCol> bucketCols = new ArrayList<BucketCol>();
List<SortCol> sortCols = new ArrayList<SortCol>();
assert rop.getConf().getKeyCols().size() <= rop.getSchema().getSignature().size();
// Group by operators select the key cols, so no need to find them in the values
for (int i = 0; i < rop.getConf().getKeyCols().size(); i++) {
+ if (groupByDesc.pruneGroupingSetId() && groupByDesc.getGroupingSetPosition() == i) {
+ continue;
+ }
String colName = rop.getSchema().getSignature().get(i).getInternalName();
bucketCols.add(new BucketCol(colName, i));
sortCols.add(new SortCol(colName, i, sortOrder.charAt(i)));
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java Fri Jan 23 19:59:11 2015
@@ -48,7 +48,6 @@ import org.apache.hadoop.hive.ql.lib.Dis
import org.apache.hadoop.hive.ql.optimizer.GenMapRedUtils;
import org.apache.hadoop.hive.ql.optimizer.MapJoinProcessor;
import org.apache.hadoop.hive.ql.parse.ParseContext;
-import org.apache.hadoop.hive.ql.parse.QBJoinTree;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.ConditionalResolverCommonJoin;
import org.apache.hadoop.hive.ql.plan.ConditionalResolverCommonJoin.ConditionalResolverCommonJoinCtx;
@@ -400,7 +399,6 @@ public class CommonJoinTaskDispatcher ex
// get parseCtx for this Join Operator
ParseContext parseCtx = physicalContext.getParseContext();
- QBJoinTree joinTree = parseCtx.getJoinContext().get(joinOp);
// start to generate multiple map join tasks
JoinDesc joinDesc = joinOp.getConf();
@@ -458,7 +456,9 @@ public class CommonJoinTaskDispatcher ex
}
currWork.setOpParseCtxMap(parseCtx.getOpParseCtx());
- currWork.setJoinTree(joinTree);
+ currWork.setLeftInputJoin(joinOp.getConf().isLeftInputJoin());
+ currWork.setBaseSrc(joinOp.getConf().getBaseSrc());
+ currWork.setMapAliases(joinOp.getConf().getMapAliases());
if (bigTablePosition >= 0) {
// create map join task and set big table as bigTablePosition
@@ -466,7 +466,7 @@ public class CommonJoinTaskDispatcher ex
newTask.setTaskTag(Task.MAPJOIN_ONLY_NOBACKUP);
newTask.setFetchSource(currTask.isFetchSource());
- replaceTask(currTask, newTask, physicalContext);
+ replaceTask(currTask, newTask);
// Can this task be merged with the child task. This can happen if a big table is being
// joined with multiple small tables on different keys
@@ -522,7 +522,9 @@ public class CommonJoinTaskDispatcher ex
listTasks.add(currTask);
// clear JoinTree and OP Parse Context
currWork.setOpParseCtxMap(null);
- currWork.setJoinTree(null);
+ currWork.setLeftInputJoin(false);
+ currWork.setBaseSrc(null);
+ currWork.setMapAliases(null);
// create conditional task and insert conditional task into task tree
ConditionalWork cndWork = new ConditionalWork(listWorks);
@@ -541,7 +543,7 @@ public class CommonJoinTaskDispatcher ex
cndTsk.setResolverCtx(resolverCtx);
// replace the current task with the new generated conditional task
- replaceTaskWithConditionalTask(currTask, cndTsk, physicalContext);
+ replaceTaskWithConditionalTask(currTask, cndTsk);
return cndTsk;
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CrossProductCheck.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CrossProductCheck.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CrossProductCheck.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CrossProductCheck.java Fri Jan 23 19:59:11 2015
@@ -241,7 +241,7 @@ public class CrossProductCheck implement
* <p>
* For MR the taskname is the StageName, for Tez it is the vertex name.
*/
- class MapJoinCheck implements NodeProcessor, NodeProcessorCtx {
+ public static class MapJoinCheck implements NodeProcessor, NodeProcessorCtx {
final List<String> warnings;
final String taskName;
@@ -302,7 +302,7 @@ public class CrossProductCheck implement
* in the Work. For Tez, you can restrict it to ReduceSinks for a particular output
* vertex.
*/
- static class ExtractReduceSinkInfo implements NodeProcessor, NodeProcessorCtx {
+ public static class ExtractReduceSinkInfo implements NodeProcessor, NodeProcessorCtx {
static class Info {
List<ExprNodeDesc> keyCols;
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java?rev=1654355&r1=1654354&r2=1654355&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java Fri Jan 23 19:59:11 2015
@@ -21,6 +21,8 @@ package org.apache.hadoop.hive.ql.optimi
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -58,9 +60,10 @@ import org.apache.hadoop.hive.serde2.Nul
*/
public class NullScanTaskDispatcher implements Dispatcher {
- private final PhysicalContext physicalContext;
- private final Map<Rule, NodeProcessor> rules;
static final Log LOG = LogFactory.getLog(NullScanTaskDispatcher.class.getName());
+
+ private final PhysicalContext physicalContext;
+ private final Map<Rule, NodeProcessor> rules;
public NullScanTaskDispatcher(PhysicalContext context, Map<Rule, NodeProcessor> rules) {
super();
@@ -91,18 +94,6 @@ public class NullScanTaskDispatcher impl
return desc;
}
- private List<String> getPathsForAlias(MapWork work, String alias) {
- List<String> paths = new ArrayList<String>();
-
- for (Map.Entry<String, ArrayList<String>> entry : work.getPathToAliases().entrySet()) {
- if (entry.getValue().contains(alias)) {
- paths.add(entry.getKey());
- }
- }
-
- return paths;
- }
-
private void processAlias(MapWork work, String path, ArrayList<String> aliasesAffected,
ArrayList<String> aliases) {
// the aliases that are allowed to map to a null scan.
@@ -164,7 +155,15 @@ public class NullScanTaskDispatcher impl
ParseContext parseContext = physicalContext.getParseContext();
WalkerCtx walkerCtx = new WalkerCtx();
- for (MapWork mapWork: task.getMapWork()) {
+ List<MapWork> mapWorks = new ArrayList<MapWork>(task.getMapWork());
+ Collections.sort(mapWorks, new Comparator<MapWork>() {
+ @Override
+ public int compare(MapWork o1, MapWork o2) {
+ return o1.getName().compareTo(o2.getName());
+ }
+ });
+
+ for (MapWork mapWork : mapWorks) {
LOG.debug("Looking at: "+mapWork.getName());
Collection<Operator<? extends OperatorDesc>> topOperators
= mapWork.getAliasToWork().values();