You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ha...@apache.org on 2013/03/02 23:38:08 UTC
svn commit: r1451954 [6/27] - in /hive/branches/ptf-windowing: ./
cli/src/java/org/apache/hadoop/hive/cli/ common/src/java/conf/
common/src/java/org/apache/hadoop/hive/conf/ conf/
contrib/src/test/results/clientpositive/ data/conf/ data/files/ hbase-ha...
Modified: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/AbstractBucketJoinProc.java
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/AbstractBucketJoinProc.java?rev=1451954&r1=1451953&r2=1451954&view=diff
==============================================================================
--- hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/AbstractBucketJoinProc.java (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/AbstractBucketJoinProc.java Sat Mar 2 22:37:59 2013
@@ -17,32 +17,438 @@
*/
package org.apache.hadoop.hive.ql.optimizer;
+import java.io.IOException;
+import java.net.URI;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.Stack;
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.fs.Path;
+import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
+import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.exec.TableScanOperator;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.lib.NodeProcessor;
import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
+import org.apache.hadoop.hive.ql.parse.ParseContext;
+import org.apache.hadoop.hive.ql.parse.PrunedPartitionList;
+import org.apache.hadoop.hive.ql.parse.QB;
+import org.apache.hadoop.hive.ql.parse.QBJoinTree;
import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.parse.TableAccessAnalyzer;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+import org.apache.hadoop.hive.ql.plan.MapJoinDesc;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
/**
* this transformation does bucket map join optimization.
*/
abstract public class AbstractBucketJoinProc implements NodeProcessor {
+ private static final Log LOG =
+ LogFactory.getLog(AbstractBucketJoinProc.class.getName());
- private static final Log LOG = LogFactory.getLog(AbstractBucketJoinProc.class.getName());
+ protected ParseContext pGraphContext;
+
+ public AbstractBucketJoinProc(ParseContext pGraphContext) {
+ this.pGraphContext = pGraphContext;
+ }
public AbstractBucketJoinProc() {
}
@Override
abstract public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
- Object... nodeOutputs) throws SemanticException;
+ Object... nodeOutputs) throws SemanticException;
+
+ private static List<String> getOnePartitionBucketFileNames(
+ URI location, ParseContext pGraphContext) throws SemanticException {
+ List<String> fileNames = new ArrayList<String>();
+ try {
+ FileSystem fs = FileSystem.get(location, pGraphContext.getConf());
+ FileStatus[] files = fs.listStatus(new Path(location.toString()));
+ if (files != null) {
+ for (FileStatus file : files) {
+ fileNames.add(file.getPath().toString());
+ }
+ }
+ } catch (IOException e) {
+ throw new SemanticException(e);
+ }
+ return fileNames;
+ }
+
+ private boolean checkBucketColumns(List<String> bucketColumns,
+ List<String> keys,
+ Integer[] orders) {
+ if (keys == null || bucketColumns == null || bucketColumns.isEmpty()) {
+ return false;
+ }
+ for (int i = 0; i < keys.size(); i++) {
+ int index = bucketColumns.indexOf(keys.get(i));
+ if (orders[i] != null && orders[i] != index) {
+ return false;
+ }
+ orders[i] = index;
+ }
+
+ // Check if the join columns contains all bucket columns.
+ // If a table is bucketized on column B, but the join key is A and B,
+ // it is easy to see joining on different buckets yield empty results.
+ return keys.containsAll(bucketColumns);
+ }
+
+ private boolean checkBucketNumberAgainstBigTable(
+ Map<String, List<Integer>> aliasToBucketNumber, int bucketNumberInPart) {
+ for (List<Integer> bucketNums : aliasToBucketNumber.values()) {
+ for (int nxt : bucketNums) {
+ boolean ok = (nxt >= bucketNumberInPart) ? nxt % bucketNumberInPart == 0
+ : bucketNumberInPart % nxt == 0;
+ if (!ok) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ protected boolean canConvertMapJoinToBucketMapJoin(
+ MapJoinOperator mapJoinOp,
+ ParseContext pGraphContext,
+ BucketJoinProcCtx context) throws SemanticException {
+
+ QBJoinTree joinCtx = this.pGraphContext.getMapJoinContext().get(mapJoinOp);
+ if (joinCtx == null) {
+ return false;
+ }
+
+ List<String> joinAliases = new ArrayList<String>();
+ String[] srcs = joinCtx.getBaseSrc();
+ String[] left = joinCtx.getLeftAliases();
+ List<String> mapAlias = joinCtx.getMapAliases();
+ String baseBigAlias = null;
+
+ for (String s : left) {
+ if (s != null) {
+ String subQueryAlias = QB.getAppendedAliasFromId(joinCtx.getId(), s);
+ if (!joinAliases.contains(subQueryAlias)) {
+ joinAliases.add(subQueryAlias);
+ if (!mapAlias.contains(s)) {
+ baseBigAlias = subQueryAlias;
+ }
+ }
+ }
+ }
+
+ for (String s : srcs) {
+ if (s != null) {
+ String subQueryAlias = QB.getAppendedAliasFromId(joinCtx.getId(), s);
+ if (!joinAliases.contains(subQueryAlias)) {
+ joinAliases.add(subQueryAlias);
+ if (!mapAlias.contains(s)) {
+ baseBigAlias = subQueryAlias;
+ }
+ }
+ }
+ }
+
+ Map<Byte, List<ExprNodeDesc>> keysMap = mapJoinOp.getConf().getKeys();
+
+ return checkConvertBucketMapJoin(
+ pGraphContext,
+ context,
+ joinCtx,
+ keysMap,
+ baseBigAlias,
+ joinAliases);
+ }
+
+ /*
+ * Can this mapjoin be converted to a bucketed mapjoin ?
+ * The following checks are performed:
+ * a. The join columns contains all the bucket columns.
+ * b. The join keys are not transformed in the sub-query.
+ * c. All partitions contain the expected number of files (number of buckets).
+ * d. The number of buckets in the big table can be divided by no of buckets in small tables.
+ */
+ protected boolean checkConvertBucketMapJoin(
+ ParseContext pGraphContext,
+ BucketJoinProcCtx context,
+ QBJoinTree joinCtx,
+ Map<Byte, List<ExprNodeDesc>> keysMap,
+ String baseBigAlias,
+ List<String> joinAliases) throws SemanticException {
+
+ LinkedHashMap<String, List<Integer>> aliasToPartitionBucketNumberMapping =
+ new LinkedHashMap<String, List<Integer>>();
+ LinkedHashMap<String, List<List<String>>> aliasToPartitionBucketFileNamesMapping =
+ new LinkedHashMap<String, List<List<String>>>();
+
+ HashMap<String, Operator<? extends OperatorDesc>> topOps = pGraphContext.getTopOps();
+ Map<TableScanOperator, Table> topToTable = pGraphContext.getTopToTable();
+
+ // (partition to bucket file names) and (partition to bucket number) for
+ // the big table;
+ LinkedHashMap<Partition, List<String>> bigTblPartsToBucketFileNames =
+ new LinkedHashMap<Partition, List<String>>();
+ LinkedHashMap<Partition, Integer> bigTblPartsToBucketNumber =
+ new LinkedHashMap<Partition, Integer>();
+
+ Integer[] orders = null; // accessing order of join cols to bucket cols, should be same
+ boolean bigTablePartitioned = true;
+ for (int index = 0; index < joinAliases.size(); index++) {
+ String alias = joinAliases.get(index);
+ Operator<? extends OperatorDesc> topOp = joinCtx.getAliasToOpInfo().get(alias);
+ if (topOp == null) {
+ return false;
+ }
+ List<String> keys = toColumns(keysMap.get((byte) index));
+ if (keys == null || keys.isEmpty()) {
+ return false;
+ }
+ int oldKeySize = keys.size();
+ TableScanOperator tso = TableAccessAnalyzer.genRootTableScan(topOp, keys);
+ if (tso == null) {
+ return false;
+ }
+
+ // For nested sub-queries, the alias mapping is not maintained in QB currently.
+ if (topOps.containsValue(tso)) {
+ for (Map.Entry<String, Operator<? extends OperatorDesc>> topOpEntry : topOps.entrySet()) {
+ if (topOpEntry.getValue() == tso) {
+ String newAlias = topOpEntry.getKey();
+ joinAliases.set(index, newAlias);
+ if (baseBigAlias.equals(alias)) {
+ baseBigAlias = newAlias;
+ }
+ alias = newAlias;
+ break;
+ }
+ }
+ }
+ else {
+ // Ideally, this should never happen, and this should be an assert.
+ return false;
+ }
+
+ // The join keys cannot be transformed in the sub-query currently.
+ // TableAccessAnalyzer.genRootTableScan will only return the base table scan
+ // if the join keys are constants or a column. Even a simple cast of the join keys
+ // will result in a null table scan operator. In case of constant join keys, they would
+ // be removed, and the size before and after the genRootTableScan will be different.
+ if (keys.size() != oldKeySize) {
+ return false;
+ }
+
+ if (orders == null) {
+ orders = new Integer[keys.size()];
+ }
+
+ Table tbl = topToTable.get(tso);
+ if (tbl.isPartitioned()) {
+ PrunedPartitionList prunedParts;
+ try {
+ prunedParts = pGraphContext.getOpToPartList().get(tso);
+ if (prunedParts == null) {
+ prunedParts =
+ PartitionPruner.prune(tbl, pGraphContext.getOpToPartPruner().get(tso),
+ pGraphContext.getConf(), alias,
+ pGraphContext.getPrunedPartitions());
+ pGraphContext.getOpToPartList().put(tso, prunedParts);
+ }
+ } catch (HiveException e) {
+ // Has to use full name to make sure it does not conflict with
+ // org.apache.commons.lang.StringUtils
+ LOG.error(org.apache.hadoop.util.StringUtils.stringifyException(e));
+ throw new SemanticException(e.getMessage(), e);
+ }
+ List<Partition> partitions = prunedParts.getNotDeniedPartns();
+ // construct a mapping of (Partition->bucket file names) and (Partition -> bucket number)
+ if (partitions.isEmpty()) {
+ if (!alias.equals(baseBigAlias)) {
+ aliasToPartitionBucketNumberMapping.put(alias, Arrays.<Integer>asList());
+ aliasToPartitionBucketFileNamesMapping.put(alias, new ArrayList<List<String>>());
+ }
+ } else {
+ List<Integer> buckets = new ArrayList<Integer>();
+ List<List<String>> files = new ArrayList<List<String>>();
+ for (Partition p : partitions) {
+ if (!checkBucketColumns(p.getBucketCols(), keys, orders)) {
+ return false;
+ }
+ List<String> fileNames =
+ getOnePartitionBucketFileNames(p.getDataLocation(), pGraphContext);
+ // The number of files for the table should be same as number of buckets.
+ int bucketCount = p.getBucketCount();
+
+ if (fileNames.size() != bucketCount) {
+ String msg = "The number of buckets for table " +
+ tbl.getTableName() + " partition " + p.getName() + " is " +
+ p.getBucketCount() + ", whereas the number of files is " + fileNames.size();
+ throw new SemanticException(
+ ErrorMsg.BUCKETED_TABLE_METADATA_INCORRECT.getMsg(msg));
+ }
+
+ if (alias.equals(baseBigAlias)) {
+ bigTblPartsToBucketFileNames.put(p, fileNames);
+ bigTblPartsToBucketNumber.put(p, bucketCount);
+ } else {
+ files.add(fileNames);
+ buckets.add(bucketCount);
+ }
+ }
+ if (!alias.equals(baseBigAlias)) {
+ aliasToPartitionBucketNumberMapping.put(alias, buckets);
+ aliasToPartitionBucketFileNamesMapping.put(alias, files);
+ }
+ }
+ } else {
+ if (!checkBucketColumns(tbl.getBucketCols(), keys, orders)) {
+ return false;
+ }
+ List<String> fileNames =
+ getOnePartitionBucketFileNames(tbl.getDataLocation(), pGraphContext);
+ Integer num = new Integer(tbl.getNumBuckets());
+
+ // The number of files for the table should be same as number of buckets.
+ if (fileNames.size() != num) {
+ String msg = "The number of buckets for table " +
+ tbl.getTableName() + " is " + tbl.getNumBuckets() +
+ ", whereas the number of files is " + fileNames.size();
+ throw new SemanticException(
+ ErrorMsg.BUCKETED_TABLE_METADATA_INCORRECT.getMsg(msg));
+ }
+
+ if (alias.equals(baseBigAlias)) {
+ bigTblPartsToBucketFileNames.put(null, fileNames);
+ bigTblPartsToBucketNumber.put(null, tbl.getNumBuckets());
+ bigTablePartitioned = false;
+ } else {
+ aliasToPartitionBucketNumberMapping.put(alias, Arrays.asList(num));
+ aliasToPartitionBucketFileNamesMapping.put(alias, Arrays.asList(fileNames));
+ }
+ }
+ }
+
+ // All tables or partitions are bucketed, and their bucket number is
+ // stored in 'bucketNumbers', we need to check if the number of buckets in
+ // the big table can be divided by no of buckets in small tables.
+ for (Integer bucketNumber : bigTblPartsToBucketNumber.values()) {
+ if (!checkBucketNumberAgainstBigTable(aliasToPartitionBucketNumberMapping, bucketNumber)) {
+ return false;
+ }
+ }
+
+ context.setAliasToPartitionBucketNumberMapping(aliasToPartitionBucketNumberMapping);
+ context.setAliasToPartitionBucketFileNamesMapping(aliasToPartitionBucketFileNamesMapping);
+ context.setBigTblPartsToBucketFileNames(bigTblPartsToBucketFileNames);
+ context.setBigTblPartsToBucketNumber(bigTblPartsToBucketNumber);
+ context.setJoinAliases(joinAliases);
+ context.setBaseBigAlias(baseBigAlias);
+ context.setBigTablePartitioned(bigTablePartitioned);
+
+ return true;
+ }
+
+ /*
+ * Convert mapjoin to a bucketed mapjoin.
+ * The operator tree is not changed, but the mapjoin descriptor in the big table is
+ * enhanced to keep the big table bucket -> small table buckets mapping.
+ */
+ protected void convertMapJoinToBucketMapJoin(
+ MapJoinOperator mapJoinOp,
+ BucketJoinProcCtx context) throws SemanticException {
+ MapJoinDesc desc = mapJoinOp.getConf();
+
+ Map<String, Map<String, List<String>>> aliasBucketFileNameMapping =
+ new LinkedHashMap<String, Map<String, List<String>>>();
+
+ Map<String, List<Integer>> aliasToPartitionBucketNumberMapping =
+ context.getAliasToPartitionBucketNumberMapping();
+
+ Map<String, List<List<String>>> aliasToPartitionBucketFileNamesMapping =
+ context.getAliasToPartitionBucketFileNamesMapping();
+
+ Map<Partition, List<String>> bigTblPartsToBucketFileNames =
+ context.getBigTblPartsToBucketFileNames();
+
+ Map<Partition, Integer> bigTblPartsToBucketNumber =
+ context.getBigTblPartsToBucketNumber();
+
+ List<String> joinAliases = context.getJoinAliases();
+ String baseBigAlias = context.getBaseBigAlias();
+
+ // sort bucket names for the big table
+ for (List<String> partBucketNames : bigTblPartsToBucketFileNames.values()) {
+ Collections.sort(partBucketNames);
+ }
+
+ // go through all small tables and get the mapping from bucket file name
+ // in the big table to bucket file names in small tables.
+ for (int j = 0; j < joinAliases.size(); j++) {
+ String alias = joinAliases.get(j);
+ if (alias.equals(baseBigAlias)) {
+ continue;
+ }
+ for (List<String> names : aliasToPartitionBucketFileNamesMapping.get(alias)) {
+ Collections.sort(names);
+ }
+ List<Integer> smallTblBucketNums = aliasToPartitionBucketNumberMapping.get(alias);
+ List<List<String>> smallTblFilesList = aliasToPartitionBucketFileNamesMapping.get(alias);
+
+ Map<String, List<String>> mapping = new LinkedHashMap<String, List<String>>();
+ aliasBucketFileNameMapping.put(alias, mapping);
+
+ // for each bucket file in big table, get the corresponding bucket file
+ // name in the small table.
+ // more than 1 partition in the big table, do the mapping for each partition
+ Iterator<Entry<Partition, List<String>>> bigTblPartToBucketNames =
+ bigTblPartsToBucketFileNames.entrySet().iterator();
+ Iterator<Entry<Partition, Integer>> bigTblPartToBucketNum = bigTblPartsToBucketNumber
+ .entrySet().iterator();
+ while (bigTblPartToBucketNames.hasNext()) {
+ assert bigTblPartToBucketNum.hasNext();
+ int bigTblBucketNum = bigTblPartToBucketNum.next().getValue();
+ List<String> bigTblBucketNameList = bigTblPartToBucketNames.next().getValue();
+ fillMapping(smallTblBucketNums, smallTblFilesList,
+ mapping, bigTblBucketNum, bigTblBucketNameList, desc.getBigTableBucketNumMapping());
+ }
+ }
+ desc.setAliasBucketFileNameMapping(aliasBucketFileNameMapping);
+ desc.setBigTableAlias(baseBigAlias);
+ boolean bigTablePartitioned = context.isBigTablePartitioned();
+ if (bigTablePartitioned) {
+ desc.setBigTablePartSpecToFileMapping(convert(bigTblPartsToBucketFileNames));
+ }
+
+ // successfully convert to bucket map join
+ desc.setBucketMapJoin(true);
+ }
+
+ // convert partition to partition spec string
+ private static Map<String, List<String>> convert(Map<Partition, List<String>> mapping) {
+ Map<String, List<String>> converted = new HashMap<String, List<String>>();
+ for (Map.Entry<Partition, List<String>> entry : mapping.entrySet()) {
+ converted.put(entry.getKey().getName(), entry.getValue());
+ }
+ return converted;
+ }
public List<String> toColumns(List<ExprNodeDesc> keys) {
List<String> columns = new ArrayList<String>();
@@ -54,4 +460,37 @@ abstract public class AbstractBucketJoin
}
return columns;
}
+
+ // called for each partition of big table and populates mapping for each file in the partition
+ private static void fillMapping(
+ List<Integer> smallTblBucketNums,
+ List<List<String>> smallTblFilesList,
+ Map<String, List<String>> mapping,
+ int bigTblBucketNum, List<String> bigTblBucketNameList,
+ Map<String, Integer> bucketFileNameMapping) {
+
+ for (int bindex = 0; bindex < bigTblBucketNameList.size(); bindex++) {
+ ArrayList<String> resultFileNames = new ArrayList<String>();
+ for (int sindex = 0 ; sindex < smallTblBucketNums.size(); sindex++) {
+ int smallTblBucketNum = smallTblBucketNums.get(sindex);
+ List<String> smallTblFileNames = smallTblFilesList.get(sindex);
+ if (bigTblBucketNum >= smallTblBucketNum) {
+ // if the big table has more buckets than the current small table,
+ // use "MOD" to get small table bucket names. For example, if the big
+ // table has 4 buckets and the small table has 2 buckets, then the
+ // mapping should be 0->0, 1->1, 2->0, 3->1.
+ int toAddSmallIndex = bindex % smallTblBucketNum;
+ resultFileNames.add(smallTblFileNames.get(toAddSmallIndex));
+ } else {
+ int jump = smallTblBucketNum / bigTblBucketNum;
+ for (int i = bindex; i < smallTblFileNames.size(); i = i + jump) {
+ resultFileNames.add(smallTblFileNames.get(i));
+ }
+ }
+ }
+ String inputBigTBLBucket = bigTblBucketNameList.get(bindex);
+ mapping.put(inputBigTBLBucket, resultFileNames);
+ bucketFileNameMapping.put(inputBigTBLBucket, bindex);
+ }
+ }
}
Modified: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/BucketMapJoinOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/BucketMapJoinOptimizer.java?rev=1451954&r1=1451953&r2=1451954&view=diff
==============================================================================
--- hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/BucketMapJoinOptimizer.java (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/BucketMapJoinOptimizer.java Sat Mar 2 22:37:59 2013
@@ -17,33 +17,15 @@
*/
package org.apache.hadoop.hive.ql.optimizer;
-import java.io.IOException;
-import java.net.URI;
import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
import java.util.Stack;
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.fs.Path;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.ql.ErrorMsg;
import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
-import org.apache.hadoop.hive.ql.exec.Operator;
-import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
-import org.apache.hadoop.hive.ql.exec.TableScanOperator;
-import org.apache.hadoop.hive.ql.exec.UnionOperator;
import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
@@ -53,18 +35,8 @@ import org.apache.hadoop.hive.ql.lib.Nod
import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
import org.apache.hadoop.hive.ql.lib.Rule;
import org.apache.hadoop.hive.ql.lib.RuleRegExp;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Partition;
-import org.apache.hadoop.hive.ql.metadata.Table;
-import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
import org.apache.hadoop.hive.ql.parse.ParseContext;
-import org.apache.hadoop.hive.ql.parse.PrunedPartitionList;
-import org.apache.hadoop.hive.ql.parse.QB;
-import org.apache.hadoop.hive.ql.parse.QBJoinTree;
import org.apache.hadoop.hive.ql.parse.SemanticException;
-import org.apache.hadoop.hive.ql.parse.TableAccessAnalyzer;
-import org.apache.hadoop.hive.ql.plan.MapJoinDesc;
-import org.apache.hadoop.hive.ql.plan.OperatorDesc;
/**
* this transformation does bucket map join optimization.
@@ -81,22 +53,13 @@ public class BucketMapJoinOptimizer impl
public ParseContext transform(ParseContext pctx) throws SemanticException {
Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
- BucketMapjoinOptProcCtx bucketMapJoinOptimizeCtx =
- new BucketMapjoinOptProcCtx(pctx.getConf());
+ BucketJoinProcCtx bucketMapJoinOptimizeCtx =
+ new BucketJoinProcCtx(pctx.getConf());
// process map joins with no reducers pattern
opRules.put(new RuleRegExp("R1",
MapJoinOperator.getOperatorName() + "%"),
getBucketMapjoinProc(pctx));
- opRules.put(new RuleRegExp("R2",
- ReduceSinkOperator.getOperatorName() + "%.*" + MapJoinOperator.getOperatorName()),
- getBucketMapjoinRejectProc(pctx));
- opRules.put(new RuleRegExp(new String("R3"),
- UnionOperator.getOperatorName() + "%.*" + MapJoinOperator.getOperatorName() + "%"),
- getBucketMapjoinRejectProc(pctx));
- opRules.put(new RuleRegExp(new String("R4"),
- MapJoinOperator.getOperatorName() + "%.*" + MapJoinOperator.getOperatorName() + "%"),
- getBucketMapjoinRejectProc(pctx));
// The dispatcher fires the processor corresponding to the closest matching
// rule and passes the context along
@@ -105,29 +68,15 @@ public class BucketMapJoinOptimizer impl
GraphWalker ogw = new DefaultGraphWalker(disp);
// Create a list of topop nodes
- ArrayList<Node> topNodes = new ArrayList<Node>();
+ List<Node> topNodes = new ArrayList<Node>();
topNodes.addAll(pctx.getTopOps().values());
ogw.startWalking(topNodes, null);
return pctx;
}
- private NodeProcessor getBucketMapjoinRejectProc(ParseContext pctx) {
- return new NodeProcessor() {
- @Override
- public Object process(Node nd, Stack<Node> stack,
- NodeProcessorCtx procCtx, Object... nodeOutputs)
- throws SemanticException {
- MapJoinOperator mapJoinOp = (MapJoinOperator) nd;
- BucketMapjoinOptProcCtx context = (BucketMapjoinOptProcCtx) procCtx;
- context.listOfRejectedMapjoins.add(mapJoinOp);
- return null;
- }
- };
- }
-
private NodeProcessor getBucketMapjoinProc(ParseContext pctx) {
- return new BucketMapjoinOptProc(pctx);
+ return new BucketMapjoinProc(pctx);
}
private NodeProcessor getDefaultProc() {
@@ -140,394 +89,4 @@ public class BucketMapJoinOptimizer impl
}
};
}
-
- class BucketMapjoinOptProc extends AbstractBucketJoinProc implements NodeProcessor {
-
- protected ParseContext pGraphContext;
-
- public BucketMapjoinOptProc(ParseContext pGraphContext) {
- super();
- this.pGraphContext = pGraphContext;
- }
-
- private boolean convertBucketMapJoin(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
- Object... nodeOutputs) throws SemanticException {
- MapJoinOperator mapJoinOp = (MapJoinOperator) nd;
- BucketMapjoinOptProcCtx context = (BucketMapjoinOptProcCtx) procCtx;
- HiveConf conf = context.getConf();
-
- if (context.getListOfRejectedMapjoins().contains(mapJoinOp)) {
- return false;
- }
-
- QBJoinTree joinCxt = this.pGraphContext.getMapJoinContext().get(mapJoinOp);
- if (joinCxt == null) {
- return false;
- }
-
- List<String> joinAliases = new ArrayList<String>();
- String[] srcs = joinCxt.getBaseSrc();
- String[] left = joinCxt.getLeftAliases();
- List<String> mapAlias = joinCxt.getMapAliases();
- String baseBigAlias = null;
-
- for (String s : left) {
- if (s != null) {
- String subQueryAlias = QB.getAppendedAliasFromId(joinCxt.getId(), s);
- if (!joinAliases.contains(subQueryAlias)) {
- joinAliases.add(subQueryAlias);
- if(!mapAlias.contains(s)) {
- baseBigAlias = subQueryAlias;
- }
- }
- }
- }
-
- for (String s : srcs) {
- if (s != null) {
- String subQueryAlias = QB.getAppendedAliasFromId(joinCxt.getId(), s);
- if (!joinAliases.contains(subQueryAlias)) {
- joinAliases.add(subQueryAlias);
- if(!mapAlias.contains(s)) {
- baseBigAlias = subQueryAlias;
- }
- }
- }
- }
-
- MapJoinDesc mjDesc = mapJoinOp.getConf();
- LinkedHashMap<String, List<Integer>> aliasToPartitionBucketNumberMapping =
- new LinkedHashMap<String, List<Integer>>();
- LinkedHashMap<String, List<List<String>>> aliasToPartitionBucketFileNamesMapping =
- new LinkedHashMap<String, List<List<String>>>();
-
- Map<String, Operator<? extends OperatorDesc>> topOps =
- this.pGraphContext.getTopOps();
- Map<TableScanOperator, Table> topToTable = this.pGraphContext.getTopToTable();
-
- // (partition to bucket file names) and (partition to bucket number) for
- // the big table;
- LinkedHashMap<Partition, List<String>> bigTblPartsToBucketFileNames = new LinkedHashMap<Partition, List<String>>();
- LinkedHashMap<Partition, Integer> bigTblPartsToBucketNumber = new LinkedHashMap<Partition, Integer>();
-
- Integer[] orders = null; // accessing order of join cols to bucket cols, should be same
- boolean bigTablePartitioned = true;
- for (int index = 0; index < joinAliases.size(); index++) {
- String alias = joinAliases.get(index);
- Operator<? extends OperatorDesc> topOp = joinCxt.getAliasToOpInfo().get(alias);
- if (topOp == null) {
- return false;
- }
- List<String> keys = toColumns(mjDesc.getKeys().get((byte) index));
- if (keys == null || keys.isEmpty()) {
- return false;
- }
- int oldKeySize = keys.size();
- TableScanOperator tso = TableAccessAnalyzer.genRootTableScan(topOp, keys);
- if (tso == null) {
- return false;
- }
-
- // For nested sub-queries, the alias mapping is not maintained in QB currently.
- if (topOps.containsValue(tso)) {
- for (Map.Entry<String, Operator<? extends OperatorDesc>> topOpEntry : topOps.entrySet()) {
- if (topOpEntry.getValue() == tso) {
- String newAlias = topOpEntry.getKey();
- joinAliases.set(index, newAlias);
- if (baseBigAlias.equals(alias)) {
- baseBigAlias = newAlias;
- }
- alias = newAlias;
- break;
- }
- }
- }
- else {
- // Ideally, this should never happen, and this should be an assert.
- return false;
- }
-
- // The join keys cannot be transformed in the sub-query currently.
- // TableAccessAnalyzer.genRootTableScan will only return the base table scan
- // if the join keys are constants or a column. Even a simple cast of the join keys
- // will result in a null table scan operator. In case of constant join keys, they would
- // be removed, and the size before and after the genRootTableScan will be different.
- if (keys.size() != oldKeySize) {
- return false;
- }
- if (orders == null) {
- orders = new Integer[keys.size()];
- }
-
- Table tbl = topToTable.get(tso);
- if (tbl.isPartitioned()) {
- PrunedPartitionList prunedParts;
- try {
- prunedParts = pGraphContext.getOpToPartList().get(tso);
- if (prunedParts == null) {
- prunedParts = PartitionPruner.prune(tbl, pGraphContext.getOpToPartPruner().get(tso),
- pGraphContext.getConf(), alias,
- pGraphContext.getPrunedPartitions());
- pGraphContext.getOpToPartList().put(tso, prunedParts);
- }
- } catch (HiveException e) {
- // Has to use full name to make sure it does not conflict with
- // org.apache.commons.lang.StringUtils
- LOG.error(org.apache.hadoop.util.StringUtils.stringifyException(e));
- throw new SemanticException(e.getMessage(), e);
- }
- List<Partition> partitions = prunedParts.getNotDeniedPartns();
- // construct a mapping of (Partition->bucket file names) and (Partition -> bucket number)
- if (partitions.isEmpty()) {
- if (!alias.equals(baseBigAlias)) {
- aliasToPartitionBucketNumberMapping.put(alias, Arrays.<Integer> asList());
- aliasToPartitionBucketFileNamesMapping.put(alias, new ArrayList<List<String>>());
- }
- } else {
- List<Integer> buckets = new ArrayList<Integer>();
- List<List<String>> files = new ArrayList<List<String>>();
- for (Partition p : partitions) {
- if (!checkBucketColumns(p.getBucketCols(), keys, orders)) {
- return false;
- }
- List<String> fileNames = getOnePartitionBucketFileNames(p.getDataLocation());
- // The number of files for the table should be same as number of buckets.
- int bucketCount = p.getBucketCount();
- if (fileNames.size() != bucketCount) {
- String msg = "The number of buckets for table " +
- tbl.getTableName() + " partition " + p.getName() + " is " +
- p.getBucketCount() + ", whereas the number of files is " + fileNames.size();
- throw new SemanticException(
- ErrorMsg.BUCKETED_TABLE_METADATA_INCORRECT.getMsg(msg));
- }
- if (alias.equals(baseBigAlias)) {
- bigTblPartsToBucketFileNames.put(p, fileNames);
- bigTblPartsToBucketNumber.put(p, bucketCount);
- } else {
- files.add(fileNames);
- buckets.add(bucketCount);
- }
- }
- if (!alias.equals(baseBigAlias)) {
- aliasToPartitionBucketNumberMapping.put(alias, buckets);
- aliasToPartitionBucketFileNamesMapping.put(alias, files);
- }
- }
- } else {
- if (!checkBucketColumns(tbl.getBucketCols(), keys, orders)) {
- return false;
- }
- List<String> fileNames = getOnePartitionBucketFileNames(tbl.getDataLocation());
- Integer num = new Integer(tbl.getNumBuckets());
- // The number of files for the table should be same as number of buckets.
- if (fileNames.size() != num) {
- String msg = "The number of buckets for table " +
- tbl.getTableName() + " is " + tbl.getNumBuckets() +
- ", whereas the number of files is " + fileNames.size();
- throw new SemanticException(
- ErrorMsg.BUCKETED_TABLE_METADATA_INCORRECT.getMsg(msg));
- }
- if (alias.equals(baseBigAlias)) {
- bigTblPartsToBucketFileNames.put(null, fileNames);
- bigTblPartsToBucketNumber.put(null, tbl.getNumBuckets());
- bigTablePartitioned = false;
- } else {
- aliasToPartitionBucketNumberMapping.put(alias, Arrays.asList(num));
- aliasToPartitionBucketFileNamesMapping.put(alias, Arrays.asList(fileNames));
- }
- }
- }
-
- // All tables or partitions are bucketed, and their bucket number is
- // stored in 'bucketNumbers', we need to check if the number of buckets in
- // the big table can be divided by no of buckets in small tables.
- for (Integer bucketNumber : bigTblPartsToBucketNumber.values()) {
- if (!checkBucketNumberAgainstBigTable(aliasToPartitionBucketNumberMapping, bucketNumber)) {
- return false;
- }
- }
-
- MapJoinDesc desc = mapJoinOp.getConf();
-
- Map<String, Map<String, List<String>>> aliasBucketFileNameMapping =
- new LinkedHashMap<String, Map<String, List<String>>>();
-
- // sort bucket names for the big table
- for (List<String> partBucketNames : bigTblPartsToBucketFileNames.values()) {
- Collections.sort(partBucketNames);
- }
-
- // go through all small tables and get the mapping from bucket file name
- // in the big table to bucket file names in small tables.
- for (int j = 0; j < joinAliases.size(); j++) {
- String alias = joinAliases.get(j);
- if (alias.equals(baseBigAlias)) {
- continue;
- }
- for (List<String> names : aliasToPartitionBucketFileNamesMapping.get(alias)) {
- Collections.sort(names);
- }
- List<Integer> smallTblBucketNums = aliasToPartitionBucketNumberMapping.get(alias);
- List<List<String>> smallTblFilesList = aliasToPartitionBucketFileNamesMapping.get(alias);
-
- Map<String, List<String>> mapping = new LinkedHashMap<String, List<String>>();
- aliasBucketFileNameMapping.put(alias, mapping);
-
- // for each bucket file in big table, get the corresponding bucket file
- // name in the small table.
- // more than 1 partition in the big table, do the mapping for each partition
- Iterator<Entry<Partition, List<String>>> bigTblPartToBucketNames =
- bigTblPartsToBucketFileNames.entrySet().iterator();
- Iterator<Entry<Partition, Integer>> bigTblPartToBucketNum = bigTblPartsToBucketNumber
- .entrySet().iterator();
- while (bigTblPartToBucketNames.hasNext()) {
- assert bigTblPartToBucketNum.hasNext();
- int bigTblBucketNum = bigTblPartToBucketNum.next().getValue();
- List<String> bigTblBucketNameList = bigTblPartToBucketNames.next().getValue();
- fillMapping(smallTblBucketNums, smallTblFilesList,
- mapping, bigTblBucketNum, bigTblBucketNameList, desc.getBigTableBucketNumMapping());
- }
- }
- desc.setAliasBucketFileNameMapping(aliasBucketFileNameMapping);
- desc.setBigTableAlias(baseBigAlias);
- if (bigTablePartitioned) {
- desc.setBigTablePartSpecToFileMapping(convert(bigTblPartsToBucketFileNames));
- }
- // successfully convert to bucket map join
- desc.setBucketMapJoin(true);
-
- return true;
- }
-
-
- @Override
- @SuppressWarnings("unchecked")
- public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
- Object... nodeOutputs) throws SemanticException {
-
- boolean convert = convertBucketMapJoin(nd, stack, procCtx, nodeOutputs);
- BucketMapjoinOptProcCtx context = (BucketMapjoinOptProcCtx) procCtx;
- HiveConf conf = context.getConf();
-
- // Throw an error if the user asked for bucketed mapjoin to be enforced and
- // bucketed mapjoin cannot be performed
- if (!convert && conf.getBoolVar(HiveConf.ConfVars.HIVEENFORCEBUCKETMAPJOIN)) {
- throw new SemanticException(ErrorMsg.BUCKET_MAPJOIN_NOT_POSSIBLE.getMsg());
- }
-
- return null;
- }
-
- // convert partition to partition spec string
- private Map<String, List<String>> convert(Map<Partition, List<String>> mapping) {
- Map<String, List<String>> converted = new HashMap<String, List<String>>();
- for (Map.Entry<Partition, List<String>> entry : mapping.entrySet()) {
- converted.put(entry.getKey().getName(), entry.getValue());
- }
- return converted;
- }
-
- // called for each partition of big table and populates mapping for each file in the partition
- private void fillMapping(
- List<Integer> smallTblBucketNums,
- List<List<String>> smallTblFilesList,
- Map<String, List<String>> mapping,
- int bigTblBucketNum, List<String> bigTblBucketNameList,
- Map<String, Integer> bucketFileNameMapping) {
-
- for (int bindex = 0; bindex < bigTblBucketNameList.size(); bindex++) {
- ArrayList<String> resultFileNames = new ArrayList<String>();
- for (int sindex = 0; sindex < smallTblBucketNums.size(); sindex++) {
- int smallTblBucketNum = smallTblBucketNums.get(sindex);
- List<String> smallTblFileNames = smallTblFilesList.get(sindex);
- if (bigTblBucketNum >= smallTblBucketNum) {
- // if the big table has more buckets than the current small table,
- // use "MOD" to get small table bucket names. For example, if the big
- // table has 4 buckets and the small table has 2 buckets, then the
- // mapping should be 0->0, 1->1, 2->0, 3->1.
- int toAddSmallIndex = bindex % smallTblBucketNum;
- resultFileNames.add(smallTblFileNames.get(toAddSmallIndex));
- } else {
- int jump = smallTblBucketNum / bigTblBucketNum;
- for (int i = bindex; i < smallTblFileNames.size(); i = i + jump) {
- resultFileNames.add(smallTblFileNames.get(i));
- }
- }
- }
- String inputBigTBLBucket = bigTblBucketNameList.get(bindex);
- mapping.put(inputBigTBLBucket, resultFileNames);
- bucketFileNameMapping.put(inputBigTBLBucket, bindex);
- }
- }
-
- private boolean checkBucketNumberAgainstBigTable(
- Map<String, List<Integer>> aliasToBucketNumber, int bucketNumberInPart) {
- for (List<Integer> bucketNums : aliasToBucketNumber.values()) {
- for (int nxt : bucketNums) {
- boolean ok = (nxt >= bucketNumberInPart) ? nxt % bucketNumberInPart == 0
- : bucketNumberInPart % nxt == 0;
- if (!ok) {
- return false;
- }
- }
- }
- return true;
- }
-
- private List<String> getOnePartitionBucketFileNames(URI location)
- throws SemanticException {
- List<String> fileNames = new ArrayList<String>();
- try {
- FileSystem fs = FileSystem.get(location, this.pGraphContext.getConf());
- FileStatus[] files = fs.listStatus(new Path(location.toString()));
- if (files != null) {
- for (FileStatus file : files) {
- fileNames.add(file.getPath().toString());
- }
- }
- } catch (IOException e) {
- throw new SemanticException(e);
- }
- return fileNames;
- }
-
- private boolean checkBucketColumns(List<String> bucketColumns, List<String> keys,
- Integer[] orders) {
- if (keys == null || bucketColumns == null || bucketColumns.isEmpty()) {
- return false;
- }
- for (int i = 0; i < keys.size(); i++) {
- int index = bucketColumns.indexOf(keys.get(i));
- if (orders[i] != null && orders[i] != index) {
- return false;
- }
- orders[i] = index;
- }
- // Check if the join columns contains all bucket columns.
- // If a table is bucketized on column B, but the join key is A and B,
- // it is easy to see joining on different buckets yield empty results.
- return keys.containsAll(bucketColumns);
- }
- }
-
- class BucketMapjoinOptProcCtx implements NodeProcessorCtx {
- private final HiveConf conf;
-
- // we only convert map joins that follows a root table scan in the same
- // mapper. That means there is no reducer between the root table scan and
- // mapjoin.
- Set<MapJoinOperator> listOfRejectedMapjoins = new HashSet<MapJoinOperator>();
-
- public BucketMapjoinOptProcCtx(HiveConf conf) {
- this.conf = conf;
- }
-
- public HiveConf getConf() {
- return conf;
- }
-
- public Set<MapJoinOperator> getListOfRejectedMapjoins() {
- return listOfRejectedMapjoins;
- }
- }
}
Modified: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java?rev=1451954&r1=1451953&r2=1451954&view=diff
==============================================================================
--- hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java Sat Mar 2 22:37:59 2013
@@ -74,17 +74,28 @@ public class Optimizer {
}
transformations.add(new SamplePruner());
transformations.add(new MapJoinProcessor());
+ boolean bucketMapJoinOptimizer = false;
if (HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTBUCKETMAPJOIN)) {
transformations.add(new BucketMapJoinOptimizer());
- if(HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTSORTMERGEBUCKETMAPJOIN)) {
- transformations.add(new SortedMergeBucketMapJoinOptimizer());
+ bucketMapJoinOptimizer = true;
+ }
+
+ // If optimize hive.optimize.bucketmapjoin.sortedmerge is set, add both
+ // BucketMapJoinOptimizer and SortedMergeBucketMapJoinOptimizer
+ if (HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTSORTMERGEBUCKETMAPJOIN)) {
+ if (!bucketMapJoinOptimizer) {
+ // No need to add BucketMapJoinOptimizer twice
+ transformations.add(new BucketMapJoinOptimizer());
}
+ transformations.add(new SortedMergeBucketMapJoinOptimizer());
}
+
transformations.add(new UnionProcessor());
transformations.add(new JoinReorder());
if(HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTREDUCEDEDUPLICATION)) {
transformations.add(new ReduceSinkDeDuplication());
}
+ transformations.add(new NonBlockingOpDeDupProc());
if (HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVELIMITOPTENABLE)) {
transformations.add(new GlobalLimitOptimizer());
}
Modified: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeBucketMapJoinOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeBucketMapJoinOptimizer.java?rev=1451954&r1=1451953&r2=1451954&view=diff
==============================================================================
--- hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeBucketMapJoinOptimizer.java (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedMergeBucketMapJoinOptimizer.java Sat Mar 2 22:37:59 2013
@@ -19,22 +19,17 @@
package org.apache.hadoop.hive.ql.optimizer;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Map;
import java.util.Stack;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.api.Order;
-import org.apache.hadoop.hive.ql.ErrorMsg;
-import org.apache.hadoop.hive.ql.exec.DummyStoreOperator;
+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.SMBMapJoinOperator;
-import org.apache.hadoop.hive.ql.exec.TableScanOperator;
+import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
@@ -44,18 +39,9 @@ import org.apache.hadoop.hive.ql.lib.Nod
import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
import org.apache.hadoop.hive.ql.lib.Rule;
import org.apache.hadoop.hive.ql.lib.RuleRegExp;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Partition;
-import org.apache.hadoop.hive.ql.metadata.Table;
-import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
import org.apache.hadoop.hive.ql.parse.ParseContext;
-import org.apache.hadoop.hive.ql.parse.PrunedPartitionList;
-import org.apache.hadoop.hive.ql.parse.QB;
-import org.apache.hadoop.hive.ql.parse.QBJoinTree;
import org.apache.hadoop.hive.ql.parse.SemanticException;
-import org.apache.hadoop.hive.ql.parse.TableAccessAnalyzer;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
-import org.apache.hadoop.hive.ql.plan.SMBJoinDesc;
//try to replace a bucket map join with a sorted merge map join
public class SortedMergeBucketMapJoinOptimizer implements Transform {
@@ -66,8 +52,38 @@ public class SortedMergeBucketMapJoinOpt
public SortedMergeBucketMapJoinOptimizer() {
}
+ private void getListOfRejectedJoins(
+ ParseContext pctx, SortBucketJoinProcCtx smbJoinContext)
+ throws SemanticException {
+
+ // Go through all joins - it should only contain selects and filters between
+ // tablescan and join operators.
+ Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
+ opRules.put(new RuleRegExp("R1", JoinOperator.getOperatorName() + "%"),
+ getCheckCandidateJoin());
+
+ // The dispatcher fires the processor corresponding to the closest matching
+ // rule and passes the context along
+ Dispatcher disp = new DefaultRuleDispatcher(getDefaultProc(), opRules, smbJoinContext);
+ GraphWalker ogw = new DefaultGraphWalker(disp);
+
+ // Create a list of topop nodes
+ ArrayList<Node> topNodes = new ArrayList<Node>();
+ topNodes.addAll(pctx.getTopOps().values());
+ ogw.startWalking(topNodes, null);
+ }
+
@Override
public ParseContext transform(ParseContext pctx) throws SemanticException {
+ HiveConf conf = pctx.getConf();
+ SortBucketJoinProcCtx smbJoinContext =
+ new SortBucketJoinProcCtx(conf);
+
+ // Get a list of joins which cannot be converted to a sort merge join
+ // Only selects and filters operators are allowed between the table scan and
+ // join currently. More operators can be added - the method supportAutomaticSortMergeJoin
+ // dictates which operator is allowed
+ getListOfRejectedJoins(pctx, smbJoinContext);
Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
// go through all map joins and find out all which have enabled bucket map
@@ -76,7 +92,15 @@ public class SortedMergeBucketMapJoinOpt
getSortedMergeBucketMapjoinProc(pctx));
// The dispatcher fires the processor corresponding to the closest matching
// rule and passes the context along
- Dispatcher disp = new DefaultRuleDispatcher(getDefaultProc(), opRules, null);
+
+ // There is no need for the user to specify mapjoin for it to be
+ // converted to sort-merge join
+ if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_AUTO_SORTMERGE_JOIN)) {
+ opRules.put(new RuleRegExp("R2", "JOIN%"),
+ getSortedMergeJoinProc(pctx));
+ }
+
+ Dispatcher disp = new DefaultRuleDispatcher(getDefaultProc(), opRules, smbJoinContext);
GraphWalker ogw = new DefaultGraphWalker(disp);
// Create a list of topop nodes
@@ -91,6 +115,10 @@ public class SortedMergeBucketMapJoinOpt
return new SortedMergeBucketMapjoinProc(pctx);
}
+ private NodeProcessor getSortedMergeJoinProc(ParseContext pctx) {
+ return new SortedMergeJoinProc(pctx);
+ }
+
private NodeProcessor getDefaultProc() {
return new NodeProcessor() {
@Override
@@ -102,318 +130,34 @@ public class SortedMergeBucketMapJoinOpt
};
}
- class SortedMergeBucketMapjoinProc extends AbstractBucketJoinProc implements NodeProcessor {
- private ParseContext pGraphContext;
-
- public SortedMergeBucketMapjoinProc(ParseContext pctx) {
- this.pGraphContext = pctx;
- }
-
- public SortedMergeBucketMapjoinProc() {
- }
-
- // Return true or false based on whether the mapjoin was converted successfully to
- // a sort-merge map join operator.
- private boolean convertSMBJoin(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
- Object... nodeOutputs) throws SemanticException {
- if (nd instanceof SMBMapJoinOperator) {
- return false;
- }
- MapJoinOperator mapJoinOp = (MapJoinOperator) nd;
- if (mapJoinOp.getConf().getAliasBucketFileNameMapping() == null
- || mapJoinOp.getConf().getAliasBucketFileNameMapping().size() == 0) {
- return false;
- }
-
- boolean tableSorted = true;
- QBJoinTree joinCxt = this.pGraphContext.getMapJoinContext()
- .get(mapJoinOp);
- if (joinCxt == null) {
- return false;
- }
- String[] srcs = joinCxt.getBaseSrc();
- for (int srcPos = 0; srcPos < srcs.length; srcPos++) {
- srcs[srcPos] = QB.getAppendedAliasFromId(joinCxt.getId(), srcs[srcPos]);
- }
-
- // All the tables/partitions columns should be sorted in the same order
- // For example, if tables A and B are being joined on columns c1, c2 and c3
- // which are the sorted and bucketed columns. The join would work, as long
- // c1, c2 and c3 are sorted in the same order.
- List<Order> sortColumnsFirstTable = new ArrayList<Order>();
-
- for (int pos = 0; pos < srcs.length; pos++) {
- tableSorted = tableSorted
- && isTableSorted(this.pGraphContext,
- mapJoinOp,
- joinCxt,
- pos,
- sortColumnsFirstTable,
- srcs);
- }
- if (!tableSorted) {
- //this is a mapjoin but not suit for a sort merge bucket map join. check outer joins
- MapJoinProcessor.checkMapJoin(((MapJoinOperator) nd).getConf().getPosBigTable(),
- ((MapJoinOperator) nd).getConf().getConds());
- return false;
- }
- // convert a bucket map join operator to a sorted merge bucket map join
- // operator
- convertToSMBJoin(mapJoinOp, srcs);
- return true;
- }
-
- @Override
- public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ // check if the join operator encountered is a candidate for being converted
+ // to a sort-merge join
+ private NodeProcessor getCheckCandidateJoin() {
+ return new NodeProcessor() {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
Object... nodeOutputs) throws SemanticException {
- boolean convert = convertSMBJoin(nd, stack, procCtx, nodeOutputs);
- // Throw an error if the user asked for sort merge bucketed mapjoin to be enforced
- // and sort merge bucketed mapjoin cannot be performed
- if (!convert &&
- pGraphContext.getConf().getBoolVar(
- HiveConf.ConfVars.HIVEENFORCESORTMERGEBUCKETMAPJOIN)) {
- throw new SemanticException(ErrorMsg.SORTMERGE_MAPJOIN_FAILED.getMsg());
- }
-
- return null;
- }
-
- private SMBMapJoinOperator convertToSMBJoin(MapJoinOperator mapJoinOp,
- String[] srcs) {
- SMBMapJoinOperator smbJop = new SMBMapJoinOperator(mapJoinOp);
- SMBJoinDesc smbJoinDesc = new SMBJoinDesc(mapJoinOp.getConf());
- smbJop.setConf(smbJoinDesc);
- HashMap<Byte, String> tagToAlias = new HashMap<Byte, String>();
- for (int i = 0; i < srcs.length; i++) {
- tagToAlias.put((byte) i, srcs[i]);
- }
- smbJoinDesc.setTagToAlias(tagToAlias);
-
- int indexInListMapJoinNoReducer = this.pGraphContext.getListMapJoinOpsNoReducer().indexOf(mapJoinOp);
- if(indexInListMapJoinNoReducer >= 0 ) {
- this.pGraphContext.getListMapJoinOpsNoReducer().remove(indexInListMapJoinNoReducer);
- this.pGraphContext.getListMapJoinOpsNoReducer().add(indexInListMapJoinNoReducer, smbJop);
- }
-
- Map<String, DummyStoreOperator> aliasToSink =
- new HashMap<String, DummyStoreOperator>();
- // For all parents (other than the big table), insert a dummy store operator
- /* Consider a query like:
- *
- * select * from
- * (subq1 --> has a filter)
- * join
- * (subq2 --> has a filter)
- * on some key
- *
- * Let us assume that subq1 is the small table (either specified by the user or inferred
- * automatically). The following operator tree will be created:
- *
- * TableScan (subq1) --> Select --> Filter --> DummyStore
- * \
- * \ SMBJoin
- * /
- * /
- * TableScan (subq2) --> Select --> Filter
- */
- List<? extends Operator> parentOperators = mapJoinOp.getParentOperators();
- for (int i = 0; i < parentOperators.size(); i++) {
- Operator par = parentOperators.get(i);
- int index = par.getChildOperators().indexOf(mapJoinOp);
- par.getChildOperators().remove(index);
- if (i == smbJoinDesc.getPosBigTable()) {
- par.getChildOperators().add(index, smbJop);
- }
- else {
- DummyStoreOperator dummyStoreOp = new DummyStoreOperator();
- par.getChildOperators().add(index, dummyStoreOp);
-
- List<Operator<? extends OperatorDesc>> childrenOps =
- new ArrayList<Operator<? extends OperatorDesc>>();
- childrenOps.add(smbJop);
- dummyStoreOp.setChildOperators(childrenOps);
-
- List<Operator<? extends OperatorDesc>> parentOps =
- new ArrayList<Operator<? extends OperatorDesc>>();
- parentOps.add(par);
- dummyStoreOp.setParentOperators(parentOps);
-
- aliasToSink.put(srcs[i], dummyStoreOp);
- smbJop.getParentOperators().remove(i);
- smbJop.getParentOperators().add(i, dummyStoreOp);
- }
- }
- smbJoinDesc.setAliasToSink(aliasToSink);
- List<? extends Operator> childOps = mapJoinOp.getChildOperators();
- for (int i = 0; i < childOps.size(); i++) {
- Operator child = childOps.get(i);
- int index = child.getParentOperators().indexOf(mapJoinOp);
- child.getParentOperators().remove(index);
- child.getParentOperators().add(index, smbJop);
- }
- return smbJop;
- }
-
- /**
- * Whether this table is eligible for a sort-merge join.
- *
- * @param pctx parse context
- * @param op map join operator being considered
- * @param joinTree join tree being considered
- * @param alias table alias in the join tree being checked
- * @param pos position of the table
- * @param sortColumnsFirstTable The names and order of the sorted columns for the first table.
- * It is not initialized when pos = 0.
- * @return
- * @throws SemanticException
- */
- private boolean isTableSorted(ParseContext pctx,
- MapJoinOperator op,
- QBJoinTree joinTree,
- int pos,
- List<Order> sortColumnsFirstTable,
- String[] aliases)
- throws SemanticException {
- String alias = aliases[pos];
- Map<TableScanOperator, Table> topToTable = this.pGraphContext
- .getTopToTable();
-
- /*
- * Consider a query like:
- *
- * select -- mapjoin(subq1) -- * from
- * (select a.key, a.value from tbl1 a) subq1
- * join
- * (select a.key, a.value from tbl2 a) subq2
- * on subq1.key = subq2.key;
- *
- * aliasToOpInfo contains the SelectOperator for subq1 and subq2.
- * We need to traverse the tree (using TableAccessAnalyzer) to get to the base
- * table. If the object being map-joined is a base table, then aliasToOpInfo
- * contains the TableScanOperator, and TableAccessAnalyzer is a no-op.
- */
- Operator<? extends OperatorDesc> topOp = joinTree.getAliasToOpInfo().get(alias);
- if (topOp == null) {
- return false;
- }
- List<String> joinCols = toColumns(op.getConf().getKeys().get((byte) pos));
- if (joinCols == null || joinCols.isEmpty()) {
- return false;
- }
- TableScanOperator tso = TableAccessAnalyzer.genRootTableScan(topOp, joinCols);
- if (tso == null) {
- return false;
- }
-
- // For nested sub-queries, the alias mapping is not maintained in QB currently.
- /*
- * Consider a query like:
- *
- * select count(*) from
- * (
- * select key, count(*) from
- * (
- * select --mapjoin(a)-- a.key as key, a.value as val1, b.value as val2
- * from tbl1 a join tbl2 b on a.key = b.key
- * ) subq1
- * group by key
- * ) subq2;
- *
- * The table alias should be subq2:subq1:a which needs to be fetched from topOps.
- */
- if (pGraphContext.getTopOps().containsValue(tso)) {
- for (Map.Entry<String, Operator<? extends OperatorDesc>> topOpEntry :
- this.pGraphContext.getTopOps().entrySet()) {
- if (topOpEntry.getValue() == tso) {
- alias = topOpEntry.getKey();
- aliases[pos] = alias;
- break;
+ SortBucketJoinProcCtx smbJoinContext = (SortBucketJoinProcCtx)procCtx;
+ JoinOperator joinOperator = (JoinOperator)nd;
+ int size = stack.size();
+ if (!(stack.get(size-1) instanceof JoinOperator) ||
+ !(stack.get(size-2) instanceof ReduceSinkOperator)) {
+ smbJoinContext.getRejectedJoinOps().add(joinOperator);
+ return null;
+ }
+
+ // If any operator in the stack does not support a auto-conversion, this join should
+ // not be converted.
+ for (int pos = size -3; pos >= 0; pos--) {
+ Operator<? extends OperatorDesc> op = (Operator<? extends OperatorDesc>)stack.get(pos);
+ if (!op.supportAutomaticSortMergeJoin()) {
+ smbJoinContext.getRejectedJoinOps().add(joinOperator);
+ return null;
}
}
- }
- else {
- // Ideally, this should never happen, and this should be an assert.
- return false;
- }
-
- Table tbl = topToTable.get(tso);
- if (tbl.isPartitioned()) {
- PrunedPartitionList prunedParts = null;
- try {
- prunedParts = pGraphContext.getOpToPartList().get(tso);
- if (prunedParts == null) {
- prunedParts = PartitionPruner.prune(tbl, pGraphContext
- .getOpToPartPruner().get(tso), pGraphContext.getConf(), alias,
- pGraphContext.getPrunedPartitions());
- pGraphContext.getOpToPartList().put(tso, prunedParts);
- }
- } catch (HiveException e) {
- LOG.error(org.apache.hadoop.util.StringUtils.stringifyException(e));
- throw new SemanticException(e.getMessage(), e);
- }
- List<Partition> partitions = prunedParts.getNotDeniedPartns();
- // Populate the names and order of columns for the first partition of the
- // first table
- if ((pos == 0) && (partitions != null) && (!partitions.isEmpty())) {
- Partition firstPartition = partitions.get(0);
- sortColumnsFirstTable.addAll(firstPartition.getSortCols());
- }
-
- for (Partition partition : prunedParts.getNotDeniedPartns()) {
- if (!checkSortColsAndJoinCols(partition.getSortCols(),
- joinCols,
- sortColumnsFirstTable)) {
- return false;
- }
- }
- return true;
- }
-
- // Populate the names and order of columns for the first table
- if (pos == 0) {
- sortColumnsFirstTable.addAll(tbl.getSortCols());
- }
-
- return checkSortColsAndJoinCols(tbl.getSortCols(),
- joinCols,
- sortColumnsFirstTable);
- }
-
- private boolean checkSortColsAndJoinCols(List<Order> sortCols,
- List<String> joinCols,
- List<Order> sortColumnsFirstPartition) {
-
- if (sortCols == null || sortCols.size() < joinCols.size()) {
- return false;
- }
-
- // A join is eligible for a sort-merge join, only if it is eligible for
- // a bucketized map join. So, we dont need to check for bucketized map
- // join here. We are guaranteed that the join keys contain all the
- // bucketized keys (note that the order need not be the same).
- List<String> sortColNames = new ArrayList<String>();
-
- // The join columns should contain all the sort columns
- // The sort columns of all the tables should be in the same order
- // compare the column names and the order with the first table/partition.
- for (int pos = 0; pos < sortCols.size(); pos++) {
- Order o = sortCols.get(pos);
- if (o.getOrder() != sortColumnsFirstPartition.get(pos).getOrder()) {
- return false;
- }
- sortColNames.add(o.getCol());
+ return null;
}
-
- // The column names and order (ascending/descending) matched
- // The first 'n' sorted columns should be the same as the joinCols, where
- // 'n' is the size of join columns.
- // For eg: if the table is sorted by (a,b,c), it is OK to convert if the join is
- // on (a), (a,b), or any combination of (a,b,c):
- // (a,b,c), (a,c,b), (c,a,b), (c,b,a), (b,c,a), (b,a,c)
- // but it is not OK to convert if the join is on (a,c)
- return sortColNames.subList(0, joinCols.size()).containsAll(joinCols);
- }
+ };
}
-
}
Modified: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinResolver.java
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinResolver.java?rev=1451954&r1=1451953&r2=1451954&view=diff
==============================================================================
--- hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinResolver.java (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinResolver.java Sat Mar 2 22:37:59 2013
@@ -50,8 +50,7 @@ import org.apache.hadoop.hive.ql.parse.P
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;
+import org.apache.hadoop.hive.ql.plan.ConditionalResolverCommonJoin.ConditionalResolverCommonJoinCtx;
import org.apache.hadoop.hive.ql.plan.ConditionalWork;
import org.apache.hadoop.hive.ql.plan.JoinDesc;
import org.apache.hadoop.hive.ql.plan.MapredLocalWork;
@@ -554,12 +553,42 @@ public class CommonJoinResolver implemen
return null;
}
+ /*
+ * If any operator which does not allow map-side conversion is present in the mapper, dont
+ * convert it into a conditional task.
+ */
+ private boolean checkOperatorOKMapJoinConversion(Operator<? extends OperatorDesc> op) {
+ if (!op.opAllowedConvertMapJoin()) {
+ return false;
+ }
+
+ if (op.getChildOperators() == null) {
+ return true;
+ }
+
+ for (Operator<? extends OperatorDesc> childOp : op.getChildOperators()) {
+ if (!checkOperatorOKMapJoinConversion(childOp)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
private JoinOperator getJoinOp(MapRedTask task) throws SemanticException {
- if (task.getWork() == null) {
+ MapredWork work = task.getWork();
+ if (work == null) {
return null;
}
- Operator<? extends OperatorDesc> reducerOp = task.getWork().getReducer();
+ Operator<? extends OperatorDesc> reducerOp = work.getReducer();
if (reducerOp instanceof JoinOperator) {
+ /* Is any operator present, which prevents the conversion */
+ Map<String, Operator<? extends OperatorDesc>> aliasToWork = work.getAliasToWork();
+ for (Operator<? extends OperatorDesc> op : aliasToWork.values()) {
+ if (!checkOperatorOKMapJoinConversion(op)) {
+ return null;
+ }
+ }
return (JoinOperator) reducerOp;
} else {
return null;
Modified: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java?rev=1451954&r1=1451953&r2=1451954&view=diff
==============================================================================
--- hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java Sat Mar 2 22:37:59 2013
@@ -126,6 +126,7 @@ import org.apache.hadoop.hive.serde.serd
import org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.hive.ql.plan.AlterTableAlterPartDesc;
/**
* DDLSemanticAnalyzer.
@@ -330,6 +331,9 @@ public class DDLSemanticAnalyzer extends
case HiveParser.TOK_ALTERTABLE_DROPPARTS:
analyzeAlterTableDropParts(ast, false);
break;
+ case HiveParser.TOK_ALTERTABLE_ALTERPARTS:
+ analyzeAlterTableAlterParts(ast);
+ break;
case HiveParser.TOK_ALTERTABLE_PROPERTIES:
analyzeAlterTableProps(ast, false, false);
break;
@@ -2277,7 +2281,60 @@ public class DDLSemanticAnalyzer extends
dropTblDesc), conf));
}
- /**
+ private void analyzeAlterTableAlterParts(ASTNode ast)
+ throws SemanticException {
+ // get table name
+ String tblName = getUnescapedName((ASTNode)ast.getChild(0));
+
+ Table tab = null;
+
+ // check if table exists.
+ try {
+ tab = db.getTable(db.getCurrentDatabase(), tblName, true);
+ inputs.add(new ReadEntity(tab));
+ } catch (HiveException e) {
+ throw new SemanticException(ErrorMsg.INVALID_TABLE.getMsg(tblName));
+ }
+
+ // validate the DDL is a valid operation on the table.
+ validateAlterTableType(tab, AlterTableTypes.ALTERPARTITION, false);
+
+ // Alter table ... partition column ( column newtype) only takes one column at a time.
+ // It must have a column name followed with type.
+ ASTNode colAst = (ASTNode) ast.getChild(1);
+ assert(colAst.getChildCount() == 2);
+
+ FieldSchema newCol = new FieldSchema();
+
+ // get column name
+ String name = colAst.getChild(0).getText().toLowerCase();
+ newCol.setName(unescapeIdentifier(name));
+
+ // get column type
+ ASTNode typeChild = (ASTNode) (colAst.getChild(1));
+ newCol.setType(getTypeStringFromAST(typeChild));
+
+ // check if column is defined or not
+ boolean fFoundColumn = false;
+ for( FieldSchema col : tab.getTTable().getPartitionKeys()) {
+ if (col.getName().compareTo(newCol.getName()) == 0) {
+ fFoundColumn = true;
+ }
+ }
+
+ // raise error if we could not find the column
+ if (!fFoundColumn) {
+ throw new SemanticException(ErrorMsg.INVALID_COLUMN.getMsg(newCol.getName()));
+ }
+
+ AlterTableAlterPartDesc alterTblAlterPartDesc =
+ new AlterTableAlterPartDesc(db.getCurrentDatabase(), tblName, newCol);
+
+ rootTasks.add(TaskFactory.get(new DDLWork(getInputs(), getOutputs(),
+ alterTblAlterPartDesc), conf));
+ }
+
+ /**
* Add one or more partitions to a table. Useful when the data has been copied
* to the right location by some other process.
*
Copied: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g (from r1451443, hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g)
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g?p2=hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g&p1=hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g&r1=1451443&r2=1451954&rev=1451954&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g Sat Mar 2 22:37:59 2013
@@ -139,7 +139,7 @@ fromSource
@init { gParent.msgs.push("from source"); }
@after { gParent.msgs.pop(); }
:
- (tableSource | subQuerySource) (lateralView^)*
+ ((Identifier LPAREN)=> partitionedTableFunction | tableSource | subQuerySource) (lateralView^)*
;
tableBucketSample
@@ -202,6 +202,98 @@ subQuerySource
LPAREN queryStatementExpression RPAREN identifier -> ^(TOK_SUBQUERY queryStatementExpression identifier)
;
+//---------------------- Rules for parsing PTF clauses -----------------------------
+partitioningSpec
+@init { gParent.msgs.push("partitioningSpec clause"); }
+@after { gParent.msgs.pop(); }
+ :
+ partitionByClause orderByClause? -> ^(TOK_PARTITIONINGSPEC partitionByClause orderByClause?) |
+ distributeByClause sortByClause? -> ^(TOK_PARTITIONINGSPEC distributeByClause sortByClause?) |
+ clusterByClause -> ^(TOK_PARTITIONINGSPEC clusterByClause)
+ ;
+
+partitionTableFunctionSource
+@init { gParent.msgs.push("partitionTableFunctionSource clause"); }
+@after { gParent.msgs.pop(); }
+ :
+ subQuerySource |
+ tableSource |
+ partitionedTableFunction
+ ;
+
+partitionedTableFunction
+@init { gParent.msgs.push("ptf clause"); }
+@after { gParent.msgs.pop(); }
+ :
+ name=Identifier
+ LPAREN KW_ON ptfsrc=partitionTableFunctionSource partitioningSpec?
+ ((Identifier LPAREN expression RPAREN ) => Identifier LPAREN expression RPAREN ( COMMA Identifier LPAREN expression RPAREN)*)?
+ RPAREN alias=Identifier?
+ -> ^(TOK_PTBLFUNCTION $name $alias? partitionTableFunctionSource partitioningSpec? expression*)
+ ;
+
+//---------------------- Rules for windowing clauses -------------------------------
+window_clause
+@init { gParent.msgs.push("window_clause"); }
+@after { gParent.msgs.pop(); }
+:
+ KW_WINDOW window_defn (COMMA window_defn)* -> ^(KW_WINDOW window_defn+)
+;
+
+window_defn
+@init { gParent.msgs.push("window_defn"); }
+@after { gParent.msgs.pop(); }
+:
+ Identifier KW_AS window_specification -> ^(TOK_WINDOWDEF Identifier window_specification)
+;
+
+window_specification
+@init { gParent.msgs.push("window_specification"); }
+@after { gParent.msgs.pop(); }
+:
+ (Identifier | ( LPAREN Identifier? partitioningSpec? window_frame? RPAREN)) -> ^(TOK_WINDOWSPEC Identifier? partitioningSpec? window_frame?)
+;
+
+window_frame :
+ window_range_expression |
+ window_value_expression
+;
+
+window_range_expression
+@init { gParent.msgs.push("window_range_expression"); }
+@after { gParent.msgs.pop(); }
+:
+ KW_ROWS KW_UNBOUNDED KW_PRECEDING -> ^(TOK_WINDOWRANGE ^(KW_PRECEDING KW_UNBOUNDED) ^(KW_CURRENT)) |
+ KW_ROWS KW_BETWEEN s=rowsboundary KW_AND end=rowsboundary -> ^(TOK_WINDOWRANGE $s $end)
+;
+
+rowsboundary
+@init { gParent.msgs.push("rowsboundary"); }
+@after { gParent.msgs.pop(); }
+:
+ KW_UNBOUNDED (r=KW_PRECEDING|r=KW_FOLLOWING) -> ^($r KW_UNBOUNDED) |
+ KW_CURRENT KW_ROW -> ^(KW_CURRENT) |
+ Number (d=KW_PRECEDING | d=KW_FOLLOWING ) -> ^($d Number)
+;
+
+window_value_expression
+@init { gParent.msgs.push("window_value_expression"); }
+@after { gParent.msgs.pop(); }
+:
+ KW_RANGE KW_UNBOUNDED KW_PRECEDING -> ^(TOK_WINDOWVALUES ^(KW_PRECEDING KW_UNBOUNDED) ^(KW_CURRENT)) |
+ KW_RANGE KW_BETWEEN s=valuesboundary KW_AND end=valuesboundary -> ^(TOK_WINDOWVALUES $s $end)
+;
+
+valuesboundary
+@init { gParent.msgs.push("valuesboundary"); }
+@after { gParent.msgs.pop(); }
+:
+ KW_UNBOUNDED (r=KW_PRECEDING|r=KW_FOLLOWING) -> ^($r KW_UNBOUNDED) |
+ KW_CURRENT KW_ROW -> ^(KW_CURRENT) |
+ rowExp=expression rngExp=Number (d=KW_LESS | d=KW_MORE ) -> ^($d $rowExp $rngExp)
+;
+
+
//----------------------- Rules for parsing whereClause -----------------------------
// where a=b and ...
whereClause
Copied: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveLexer.g (from r1451443, hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveLexer.g)
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveLexer.g?p2=hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveLexer.g&p1=hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveLexer.g&r1=1451443&r2=1451954&rev=1451954&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveLexer.g (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveLexer.g Sat Mar 2 22:37:59 2013
@@ -241,6 +241,14 @@ KW_ROLLUP: 'ROLLUP';
KW_CUBE: 'CUBE';
KW_DIRECTORIES: 'DIRECTORIES';
KW_FOR: 'FOR';
+KW_WINDOW: 'WINDOW';
+KW_UNBOUNDED: 'UNBOUNDED';
+KW_PRECEDING: 'PRECEDING';
+KW_FOLLOWING: 'FOLLOWING';
+KW_CURRENT: 'CURRENT';
+KW_LESS: 'LESS';
+KW_MORE: 'MORE';
+KW_OVER: 'OVER';
KW_GROUPING: 'GROUPING';
KW_SETS: 'SETS';
KW_TRUNCATE: 'TRUNCATE';
Copied: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g (from r1451443, hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g)
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g?p2=hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g&p1=hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g&r1=1451443&r2=1451954&rev=1451954&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g Sat Mar 2 22:37:59 2013
@@ -287,6 +287,12 @@ TOK_SKEWED_LOCATIONS;
TOK_SKEWED_LOCATION_LIST;
TOK_SKEWED_LOCATION_MAP;
TOK_STOREDASDIRS;
+TOK_PARTITIONINGSPEC;
+TOK_PTBLFUNCTION;
+TOK_WINDOWDEF;
+TOK_WINDOWSPEC;
+TOK_WINDOWVALUES;
+TOK_WINDOWRANGE;
TOK_IGNOREPROTECTION;
}
@@ -1789,9 +1795,10 @@ regular_body
clusterByClause?
distributeByClause?
sortByClause?
+ window_clause?
limitClause? -> ^(TOK_QUERY fromClause ^(TOK_INSERT insertClause
selectClause whereClause? groupByClause? havingClause? orderByClause? clusterByClause?
- distributeByClause? sortByClause? limitClause?))
+ distributeByClause? sortByClause? window_clause? limitClause?))
|
selectStatement
;
@@ -1807,9 +1814,10 @@ selectStatement
clusterByClause?
distributeByClause?
sortByClause?
+ window_clause?
limitClause? -> ^(TOK_QUERY fromClause ^(TOK_INSERT ^(TOK_DESTINATION ^(TOK_DIR TOK_TMP_FILE))
selectClause whereClause? groupByClause? havingClause? orderByClause? clusterByClause?
- distributeByClause? sortByClause? limitClause?))
+ distributeByClause? sortByClause? window_clause? limitClause?))
;
@@ -1824,9 +1832,10 @@ body
clusterByClause?
distributeByClause?
sortByClause?
+ window_clause?
limitClause? -> ^(TOK_INSERT insertClause?
selectClause whereClause? groupByClause? havingClause? orderByClause? clusterByClause?
- distributeByClause? sortByClause? limitClause?)
+ distributeByClause? sortByClause? window_clause? limitClause?)
|
selectClause
whereClause?
@@ -1836,9 +1845,10 @@ body
clusterByClause?
distributeByClause?
sortByClause?
+ window_clause?
limitClause? -> ^(TOK_INSERT ^(TOK_DESTINATION ^(TOK_DIR TOK_TMP_FILE))
selectClause whereClause? groupByClause? havingClause? orderByClause? clusterByClause?
- distributeByClause? sortByClause? limitClause?)
+ distributeByClause? sortByClause? window_clause? limitClause?)
;
insertClause
Copied: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g (from r1451443, hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g)
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g?p2=hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g&p1=hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g&r1=1451443&r2=1451954&rev=1451954&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g Sat Mar 2 22:37:59 2013
@@ -124,7 +124,18 @@ clusterByClause
|
KW_CLUSTER KW_BY
expression
- ( COMMA expression )* -> ^(TOK_CLUSTERBY expression+)
+ ( (COMMA)=>COMMA expression )* -> ^(TOK_CLUSTERBY expression+)
+ ;
+
+partitionByClause
+@init { gParent.msgs.push("partition by clause"); }
+@after { gParent.msgs.pop(); }
+ :
+ KW_PARTITION KW_BY
+ LPAREN expression (COMMA expression)* RPAREN -> ^(TOK_DISTRIBUTEBY expression+)
+ |
+ KW_PARTITION KW_BY
+ expression ((COMMA)=> COMMA expression)* -> ^(TOK_DISTRIBUTEBY expression+)
;
distributeByClause
@@ -135,7 +146,7 @@ distributeByClause
LPAREN expression (COMMA expression)* RPAREN -> ^(TOK_DISTRIBUTEBY expression+)
|
KW_DISTRIBUTE KW_BY
- expression (COMMA expression)* -> ^(TOK_DISTRIBUTEBY expression+)
+ expression ((COMMA)=> COMMA expression)* -> ^(TOK_DISTRIBUTEBY expression+)
;
sortByClause
@@ -148,7 +159,7 @@ sortByClause
|
KW_SORT KW_BY
columnRefOrder
- ( COMMA columnRefOrder)* -> ^(TOK_SORTBY columnRefOrder+)
+ ( (COMMA)=> COMMA columnRefOrder)* -> ^(TOK_SORTBY columnRefOrder+)
;
// fun(par1, par2, par3)
Modified: hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java
URL: http://svn.apache.org/viewvc/hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java?rev=1451954&r1=1451953&r2=1451954&view=diff
==============================================================================
--- hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java (original)
+++ hive/branches/ptf-windowing/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java Sat Mar 2 22:37:59 2013
@@ -34,6 +34,7 @@ import org.apache.hadoop.hive.ql.exec.Gr
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.SMBMapJoinOperator;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
import org.apache.hadoop.hive.ql.exec.Task;
import org.apache.hadoop.hive.ql.hooks.LineageInfo;
@@ -71,6 +72,7 @@ public class ParseContext {
private LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtx;
private Map<JoinOperator, QBJoinTree> joinContext;
private Map<MapJoinOperator, QBJoinTree> mapJoinContext;
+ private Map<SMBMapJoinOperator, QBJoinTree> smbMapJoinContext;
private HashMap<TableScanOperator, Table> topToTable;
private HashMap<String, SplitSample> nameToSplitSample;
private List<LoadTableDesc> loadTableWork;
@@ -160,6 +162,7 @@ public class ParseContext {
HashMap<String, Operator<? extends OperatorDesc>> topSelOps,
LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtx,
Map<JoinOperator, QBJoinTree> joinContext,
+ Map<SMBMapJoinOperator, QBJoinTree> smbMapJoinContext,
HashMap<TableScanOperator, Table> topToTable,
List<LoadTableDesc> loadTableWork, List<LoadFileDesc> loadFileWork,
Context ctx, HashMap<String, String> idToTableNameMap, int destTableId,
@@ -178,6 +181,7 @@ public class ParseContext {
this.opToPartPruner = opToPartPruner;
this.opToPartList = opToPartList;
this.joinContext = joinContext;
+ this.smbMapJoinContext = smbMapJoinContext;
this.topToTable = topToTable;
this.loadFileWork = loadFileWork;
this.loadTableWork = loadTableWork;
@@ -528,6 +532,14 @@ public class ParseContext {
this.mapJoinContext = mapJoinContext;
}
+ public Map<SMBMapJoinOperator, QBJoinTree> getSmbMapJoinContext() {
+ return smbMapJoinContext;
+ }
+
+ public void setSmbMapJoinContext(Map<SMBMapJoinOperator, QBJoinTree> smbMapJoinContext) {
+ this.smbMapJoinContext = smbMapJoinContext;
+ }
+
public GlobalLimitCtx getGlobalLimitCtx() {
return globalLimitCtx;
}