You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by gu...@apache.org on 2013/08/30 01:31:38 UTC
svn commit: r1518855 - in
/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql: exec/ optimizer/
optimizer/ppr/ parse/ plan/
Author: gunther
Date: Thu Aug 29 23:31:38 2013
New Revision: 1518855
URL: http://svn.apache.org/r1518855
Log:
HIVE-5095: Hive needs new operator walker for parallelization/optimization for tez (Gunther Hagleitner)
Added:
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/OptimizeTezProcContext.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetReducerParallelism.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetStatistics.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezWalker.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/Statistics.java
Removed:
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezTaskWalker.java
Modified:
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/TableScanOperator.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/PrunerOperatorFactory.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionPruner.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/AbstractOperatorDesc.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/OperatorDesc.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java Thu Aug 29 23:31:38 2013
@@ -34,6 +34,8 @@ import org.apache.hadoop.hive.ql.exec.pe
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.JoinCondDesc;
import org.apache.hadoop.hive.ql.plan.JoinDesc;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+import org.apache.hadoop.hive.ql.plan.Statistics;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.serde2.io.ShortWritable;
import org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe;
@@ -817,6 +819,25 @@ public abstract class CommonJoinOperator
}
@Override
+ public Statistics getStatistics(HiveConf conf) throws HiveException {
+ Statistics stats = this.getConf().getStatistics();
+ if (stats == null) {
+ long maxSize = 0;
+ stats = new Statistics();
+ for (Operator<? extends OperatorDesc> parent: this.getParentOperators()) {
+ long size = parent.getStatistics(conf).getNumberOfBytes();
+ if (maxSize < size) {
+ maxSize = size;
+ }
+ }
+
+ stats.setNumberOfBytes(maxSize*this.getParentOperators().size());
+ this.getConf().setStatistics(stats);
+ }
+ return stats;
+ }
+
+ @Override
public boolean opAllowedBeforeMapJoin() {
return false;
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/DemuxOperator.java Thu Aug 29 23:31:38 2013
@@ -28,9 +28,11 @@ import java.util.Map.Entry;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.DemuxDesc;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+import org.apache.hadoop.hive.ql.plan.Statistics;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.ql.plan.api.OperatorType;
import org.apache.hadoop.hive.serde2.Deserializer;
@@ -306,6 +308,16 @@ public class DemuxOperator extends Opera
}
@Override
+ public Statistics getStatistics(HiveConf conf) throws HiveException {
+ Statistics stats = this.getConf().getStatistics();
+ if (stats == null) {
+ super.getStatistics(conf);
+ stats.setNumberOfBytes(stats.getNumberOfBytes()/this.getParentOperators().size());
+ }
+ return stats;
+ }
+
+ @Override
public void startGroup() throws HiveException {
lastChildIndex = 0;
super.startGroup();
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java Thu Aug 29 23:31:38 2013
@@ -32,6 +32,7 @@ import java.util.concurrent.atomic.Atomi
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.mr.ExecMapperContext;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.metadata.HiveException;
@@ -39,6 +40,7 @@ import org.apache.hadoop.hive.ql.parse.S
import org.apache.hadoop.hive.ql.plan.Explain;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+import org.apache.hadoop.hive.ql.plan.Statistics;
import org.apache.hadoop.hive.ql.plan.api.OperatorType;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
@@ -1504,6 +1506,25 @@ public abstract class Operator<T extends
return true;
}
+ /**
+ * Computes and retrieves the stats for this operator. Default implementation assumes same
+ * input/output size for operator.
+ *
+ * @return Statistics for this operator
+ */
+ public Statistics getStatistics(HiveConf conf) throws HiveException {
+ Statistics stats = this.getConf().getStatistics();
+
+ if (stats == null) {
+ stats = new Statistics();
+ for (Operator<? extends OperatorDesc> parent: this.getParentOperators()) {
+ stats.addNumberOfBytes(parent.getStatistics(conf).getNumberOfBytes());
+ }
+ this.getConf().setStatistics(stats);
+ }
+ return stats;
+ }
+
@Override
public String toString() {
return getName() + "[" + getIdentifier() + "]";
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/TableScanOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/TableScanOperator.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/TableScanOperator.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/TableScanOperator.java Thu Aug 29 23:31:38 2013
@@ -18,7 +18,6 @@
package org.apache.hadoop.hive.ql.exec;
-import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
@@ -28,9 +27,11 @@ import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.FileUtils;
+import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.ErrorMsg;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
+import org.apache.hadoop.hive.ql.plan.Statistics;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.ql.plan.TableScanDesc;
import org.apache.hadoop.hive.ql.plan.api.OperatorType;
@@ -316,6 +317,18 @@ public class TableScanOperator extends O
}
@Override
+ public Statistics getStatistics(HiveConf conf) throws HiveException {
+ Statistics stats = this.getConf().getStatistics();
+ if (stats == null) {
+ stats = new Statistics();
+ stats.addNumberOfBytes(Utilities.getSize(alias, getConf().getTable(), conf,
+ this, getConf().getPruningPredicate()));
+ this.getConf().setStatistics(stats);
+ }
+ return stats;
+ }
+
+ @Override
public boolean supportSkewJoinOptimization() {
return true;
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java Thu Aug 29 23:31:38 2013
@@ -116,6 +116,8 @@ import org.apache.hadoop.hive.ql.io.Rewo
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.PrunedPartitionList;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.BaseWork;
import org.apache.hadoop.hive.ql.plan.DynamicPartitionCtx;
@@ -2588,22 +2590,30 @@ public final class Utilities {
+ maxReducers + " totalInputFileSize=" + totalInputFileSize);
}
+ // If this map reduce job writes final data to a table and bucketing is being inferred,
+ // and the user has configured Hive to do this, make sure the number of reducers is a
+ // power of two
+ boolean powersOfTwo = conf.getBoolVar(HiveConf.ConfVars.HIVE_INFER_BUCKET_SORT_NUM_BUCKETS_POWER_TWO) &&
+ finalMapRed && !work.getBucketedColsByDirectory().isEmpty();
+
+ return estimateReducers(totalInputFileSize, bytesPerReducer, maxReducers, powersOfTwo);
+ }
+
+ public static int estimateReducers(long totalInputFileSize, long bytesPerReducer,
+ int maxReducers, boolean powersOfTwo) {
+
int reducers = (int) ((totalInputFileSize + bytesPerReducer - 1) / bytesPerReducer);
reducers = Math.max(1, reducers);
reducers = Math.min(maxReducers, reducers);
- // If this map reduce job writes final data to a table and bucketing is being inferred,
- // and the user has configured Hive to do this, make sure the number of reducers is a
- // power of two
- if (conf.getBoolVar(HiveConf.ConfVars.HIVE_INFER_BUCKET_SORT_NUM_BUCKETS_POWER_TWO) &&
- finalMapRed && !work.getBucketedColsByDirectory().isEmpty()) {
- int reducersLog = (int)(Math.log(reducers) / Math.log(2)) + 1;
- int reducersPowerTwo = (int)Math.pow(2, reducersLog);
+ int reducersLog = (int)(Math.log(reducers) / Math.log(2)) + 1;
+ int reducersPowerTwo = (int)Math.pow(2, reducersLog);
+ if (powersOfTwo) {
// If the original number of reducers was a power of two, use that
if (reducersPowerTwo / 2 == reducers) {
- return reducers;
+ // nothing to do
} else if (reducersPowerTwo > maxReducers) {
// If the next power of two greater than the original number of reducers is greater
// than the max number of reducers, use the preceding power of two, which is strictly
@@ -2614,7 +2624,6 @@ public final class Utilities {
reducers = reducersPowerTwo;
}
}
-
return reducers;
}
@@ -2970,5 +2979,58 @@ public final class Utilities {
}
}
}
+
+ public static long getSize(String alias, Table table, HiveConf conf,
+ TableScanOperator topOp, ExprNodeDesc expr) throws HiveException {
+ long result = 0;
+ int numPartitions = 0;
+ Map<String, PrunedPartitionList> prunedPartitionsMap
+ = new HashMap<String, PrunedPartitionList>();
+
+ if (!table.isPartitioned()) {
+ result = getSize(conf, table);
+ }
+ else {
+ // For partitioned tables, get the size of all the partitions
+ PrunedPartitionList partsList = PartitionPruner.prune(table, expr, conf,
+ alias, prunedPartitionsMap);
+ numPartitions = partsList.getNotDeniedPartns().size();
+ for (Partition part : partsList.getNotDeniedPartns()) {
+ result += getSize(conf, part);
+ }
+ }
+ return result;
+ }
+
+ private static long getSize(HiveConf conf, String size, Path path) {
+ // If the size is present in the metastore, use it
+ if (size != null) {
+ try {
+ return Long.valueOf(size);
+ } catch (NumberFormatException e) {
+ return -1;
+ }
+ }
+
+ try {
+ FileSystem fs = path.getFileSystem(conf);
+ return fs.getContentSummary(path).getLength();
+ } catch (Exception e) {
+ return -1;
+ }
+ }
+
+ private static long getSize(HiveConf conf, Table table) {
+ Path path = table.getPath();
+ String size = table.getProperty("totalSize");
+ return getSize(conf, size, path);
+ }
+
+ private static long getSize(HiveConf conf, Partition partition) {
+ Path path = partition.getPartitionPath();
+ String size = partition.getParameters().get("totalSize");
+
+ return getSize(conf, size, path);
+ }
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/PrunerOperatorFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/PrunerOperatorFactory.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/PrunerOperatorFactory.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/PrunerOperatorFactory.java Thu Aug 29 23:31:38 2013
@@ -125,6 +125,9 @@ public abstract class PrunerOperatorFact
// Put the mapping from table scan operator to pruner_pred
opToPrunner.put(top, pruner_pred);
+ // Set the predicate in the table directly
+ top.getConf().setPruningPredicate(pruner_pred);
+
return;
}
@@ -165,6 +168,9 @@ public abstract class PrunerOperatorFact
// Put the mapping from table scan operator to part-pruner map
opToPrunner.put(top, partToPruner);
+ // Set the predicate in the table directly
+ top.getConf().setPruningPredicate(pruner_pred);
+
return;
}
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionPruner.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionPruner.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionPruner.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionPruner.java Thu Aug 29 23:31:38 2013
@@ -154,7 +154,7 @@ public class PartitionPruner implements
* pruner condition.
* @throws HiveException
*/
- private static PrunedPartitionList prune(Table tab, ExprNodeDesc prunerExpr,
+ public static PrunedPartitionList prune(Table tab, ExprNodeDesc prunerExpr,
HiveConf conf, String alias, Map<String, PrunedPartitionList> prunedPartitionsMap)
throws HiveException {
LOG.trace("Started pruning partiton");
Added: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/OptimizeTezProcContext.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/OptimizeTezProcContext.java?rev=1518855&view=auto
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/OptimizeTezProcContext.java (added)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/OptimizeTezProcContext.java Thu Aug 29 23:31:38 2013
@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.ql.parse;
+
+import java.util.Deque;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
+import org.apache.hadoop.hive.ql.hooks.ReadEntity;
+import org.apache.hadoop.hive.ql.hooks.WriteEntity;
+import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+
+/**
+ * OptimizeTezProcContext. OptimizeTezProcContext maintains information
+ * about the current operator plan as we walk the operator tree
+ * to do some additional optimizations on it.
+ *
+ */
+public class OptimizeTezProcContext implements NodeProcessorCtx{
+
+ public final ParseContext parseContext;
+ public final HiveConf conf;
+
+ public final Set<ReadEntity> inputs;
+ public final Set<WriteEntity> outputs;
+
+ public final Set<ReduceSinkOperator> visitedReduceSinks
+ = new HashSet<ReduceSinkOperator>();
+
+ // rootOperators are all the table scan operators in sequence
+ // of traversal
+ public final Deque<Operator<? extends OperatorDesc>> rootOperators;
+
+ @SuppressWarnings("unchecked")
+ public OptimizeTezProcContext(HiveConf conf, ParseContext parseContext,
+ Set<ReadEntity> inputs, Set<WriteEntity> outputs,
+ Deque<Operator<?>> rootOperators) {
+
+ this.conf = conf;
+ this.parseContext = parseContext;
+ this.inputs = inputs;
+ this.outputs = outputs;
+ this.rootOperators = rootOperators;
+ }
+}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java Thu Aug 29 23:31:38 2013
@@ -7756,6 +7756,10 @@ public class SemanticAnalyzer extends Ba
// Add a mapping from the table scan operator to Table
topToTable.put((TableScanOperator) top, tab);
+
+ // set the table in the tablescan descriptor directly
+ ((TableScanOperator) top).getConf().setTable(tab);
+
Map<String, String> props = qb.getTabPropsForAlias(alias);
if (props != null) {
topToTableProps.put((TableScanOperator) top, props);
Added: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetReducerParallelism.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetReducerParallelism.java?rev=1518855&view=auto
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetReducerParallelism.java (added)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetReducerParallelism.java Thu Aug 29 23:31:38 2013
@@ -0,0 +1,96 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.ql.parse;
+
+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.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+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.plan.OperatorDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc;
+
+/**
+ * SetReducerParallelism determines how many reducers should
+ * be run for a given reduce sink.
+ */
+public class SetReducerParallelism implements NodeProcessor {
+
+ static final private Log LOG = LogFactory.getLog(SetReducerParallelism.class.getName());
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Object process(Node nd, Stack<Node> stack,
+ NodeProcessorCtx procContext, Object... nodeOutputs)
+ throws SemanticException {
+
+ OptimizeTezProcContext context = (OptimizeTezProcContext) procContext;
+
+ ReduceSinkOperator sink = (ReduceSinkOperator) nd;
+ ReduceSinkDesc desc = sink.getConf();
+
+ long bytesPerReducer = context.conf.getLongVar(HiveConf.ConfVars.BYTESPERREDUCER);
+ int maxReducers = context.conf.getIntVar(HiveConf.ConfVars.MAXREDUCERS);
+ int constantReducers = context.conf.getIntVar(HiveConf.ConfVars.HADOOPNUMREDUCERS);
+
+ if (context.visitedReduceSinks.contains(sink)) {
+ // skip walking the children
+ LOG.debug("Already processed reduce sink: " + sink.getName());
+ return true;
+ }
+
+ context.visitedReduceSinks.add(sink);
+
+ try {
+ if (desc.getNumReducers() <= 0) {
+ if (constantReducers > 0) {
+ LOG.info("Parallelism for reduce sink "+sink+" set by user to "+constantReducers);
+ desc.setNumReducers(constantReducers);
+ } else {
+ long numberOfBytes = 0;
+
+ // we need to add up all the estimates from the siblings of this reduce sink
+ for (Operator<? extends OperatorDesc> sibling:
+ sink.getChildOperators().get(0).getParentOperators()) {
+ numberOfBytes += sibling.getStatistics(context.conf).getNumberOfBytes();
+ }
+
+ int numReducers = Utilities.estimateReducers(numberOfBytes, bytesPerReducer,
+ maxReducers, false);
+ LOG.info("Set parallelism for reduce sink "+sink+" to: "+numReducers);
+ desc.setNumReducers(numReducers);
+ }
+ } else {
+ LOG.info("Number of reducers determined to be: "+desc.getNumReducers());
+ }
+ } catch (HiveException e) {
+ throw new SemanticException(e);
+ }
+
+ return false;
+ }
+
+}
Added: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetStatistics.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetStatistics.java?rev=1518855&view=auto
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetStatistics.java (added)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SetStatistics.java Thu Aug 29 23:31:38 2013
@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.ql.parse;
+
+import java.util.Stack;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
+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;
+
+/**
+ * SetStatistics just invokes getStatistics on FileSinkOperators
+ * to populate the stats in the operator tree
+ */
+public class SetStatistics implements NodeProcessor {
+
+ static final private Log LOG = LogFactory.getLog(SetStatistics.class.getName());
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Object process(Node nd, Stack<Node> stack,
+ NodeProcessorCtx procContext, Object... nodeOutputs)
+ throws SemanticException {
+
+ OptimizeTezProcContext context = (OptimizeTezProcContext)procContext;
+ FileSinkOperator sink = (FileSinkOperator) nd;
+ try {
+ sink.getStatistics(context.conf);
+ } catch (HiveException e) {
+ throw new SemanticException(e);
+ }
+ return false;
+ }
+}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java Thu Aug 29 23:31:38 2013
@@ -62,13 +62,49 @@ public class TezCompiler extends TaskCom
}
@Override
+ protected void optimizeOperatorPlan(ParseContext pCtx, Set<ReadEntity> inputs,
+ Set<WriteEntity> outputs) throws SemanticException {
+
+ // Sequence of TableScan operators to be walked
+ Deque<Operator<?>> deque = new LinkedList<Operator<?>>();
+ deque.addAll(pCtx.getTopOps().values());
+
+ // Create the context for the walker
+ OptimizeTezProcContext procCtx
+ = new OptimizeTezProcContext(conf, pCtx, inputs, outputs, deque);
+
+ // create a walker which walks the tree in a DFS manner while maintaining
+ // the operator stack.
+ Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
+ opRules.put(new RuleRegExp(new String("Set parallelism - ReduceSink"),
+ ReduceSinkOperator.getOperatorName() + "%"),
+ new SetReducerParallelism());
+
+ // if this is an explain statement add rule to generate statistics for
+ // the whole tree.
+ if (pCtx.getContext().getExplain()) {
+ opRules.put(new RuleRegExp(new String("Set statistics - FileSink"),
+ FileSinkOperator.getOperatorName() + "%"),
+ new SetStatistics());
+ }
+
+ // The dispatcher fires the processor corresponding to the closest matching
+ // rule and passes the context along
+ Dispatcher disp = new DefaultRuleDispatcher(null, opRules, procCtx);
+ List<Node> topNodes = new ArrayList<Node>();
+ topNodes.addAll(pCtx.getTopOps().values());
+ GraphWalker ogw = new TezWalker(disp);
+ ogw.startWalking(topNodes, null);
+ }
+
+ @Override
protected void generateTaskTree(List<Task<? extends Serializable>> rootTasks, ParseContext pCtx,
List<Task<MoveWork>> mvTask, Set<ReadEntity> inputs, Set<WriteEntity> outputs)
throws SemanticException {
- // generate map reduce plans
ParseContext tempParseContext = getParseContext(pCtx, rootTasks);
+ // Sequence of TableScan operators to be walked
Deque<Operator<?>> deque = new LinkedList<Operator<?>>();
deque.addAll(pCtx.getTopOps().values());
@@ -92,7 +128,7 @@ public class TezCompiler extends TaskCom
Dispatcher disp = new DefaultRuleDispatcher(null, opRules, procCtx);
List<Node> topNodes = new ArrayList<Node>();
topNodes.addAll(pCtx.getTopOps().values());
- GraphWalker ogw = new GenTezTaskWalker(disp);
+ GraphWalker ogw = new TezWalker(disp);
ogw.startWalking(topNodes, null);
}
Added: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezWalker.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezWalker.java?rev=1518855&view=auto
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezWalker.java (added)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/TezWalker.java Thu Aug 29 23:31:38 2013
@@ -0,0 +1,66 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.ql.parse;
+
+import java.util.List;
+
+import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
+import org.apache.hadoop.hive.ql.lib.Dispatcher;
+import org.apache.hadoop.hive.ql.lib.Node;
+
+/**
+ * Walks the operator tree in DFS fashion.
+ */
+public class TezWalker extends DefaultGraphWalker {
+
+ /**
+ * constructor of the walker - the dispatcher is passed.
+ *
+ * @param disp
+ * the dispatcher to be called for each node visited
+ */
+ public TezWalker(Dispatcher disp) {
+ super(disp);
+ }
+
+ /**
+ * Walk the given operator.
+ *
+ * @param nd
+ * operator being walked
+ */
+ @Override
+ public void walk(Node nd) throws SemanticException {
+ List<? extends Node> children = nd.getChildren();
+
+ // maintain the stack of operators encountered
+ opStack.push(nd);
+ Boolean skip = dispatchAndReturn(nd, opStack);
+
+ if (skip == null || !skip) {
+ // move all the children to the front of queue
+ for (Node ch : children) {
+ walk(ch);
+ }
+ }
+
+ // done with this operator
+ opStack.pop();
+ }
+}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/AbstractOperatorDesc.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/AbstractOperatorDesc.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/AbstractOperatorDesc.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/AbstractOperatorDesc.java Thu Aug 29 23:31:38 2013
@@ -19,6 +19,19 @@
package org.apache.hadoop.hive.ql.plan;
public class AbstractOperatorDesc implements OperatorDesc {
+ protected transient Statistics statistics;
+
+ @Override
+ @Explain(displayName = "Statistics", normalExplain = false)
+ public Statistics getStatistics() {
+ return statistics;
+ }
+
+ @Override
+ public void setStatistics(Statistics statistics) {
+ this.statistics = statistics;
+ }
+
@Override
public Object clone() throws CloneNotSupportedException {
throw new CloneNotSupportedException("clone not supported");
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/OperatorDesc.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/OperatorDesc.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/OperatorDesc.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/OperatorDesc.java Thu Aug 29 23:31:38 2013
@@ -22,4 +22,6 @@ import java.io.Serializable;
public interface OperatorDesc extends Serializable, Cloneable {
public Object clone() throws CloneNotSupportedException;
+ public Statistics getStatistics();
+ public void setStatistics(Statistics statistics);
}
Added: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/Statistics.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/Statistics.java?rev=1518855&view=auto
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/Statistics.java (added)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/Statistics.java Thu Aug 29 23:31:38 2013
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.ql.plan;
+
+import java.io.Serializable;
+
+/**
+ * Statistics. Describes the output of an operator in terms of size, rows, etc
+ * based on estimates.
+ *
+ */
+@SuppressWarnings("serial")
+public class Statistics implements Serializable {
+
+ // only available stat right now is the amount of data flowing out of an
+ // operator;
+ private long numberOfBytes = -1;
+
+ @Explain(displayName = "Estimated data size in bytes")
+ public long getNumberOfBytes() {
+ return numberOfBytes;
+ }
+
+ public void addNumberOfBytes(long numberOfBytes) {
+ if (this.numberOfBytes < 0) {
+ this.numberOfBytes = 0;
+ }
+
+ this.numberOfBytes += numberOfBytes;
+ }
+
+ public void setNumberOfBytes(long l) {
+ this.numberOfBytes = l;
+ }
+}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java?rev=1518855&r1=1518854&r2=1518855&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java Thu Aug 29 23:31:38 2013
@@ -22,6 +22,8 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import org.apache.hadoop.hive.ql.exec.PTFUtils;
+import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
/**
@@ -52,6 +54,10 @@ public class TableScanDesc extends Abstr
*/
private int rowLimit = -1;
+ private transient Table table;
+
+ private transient ExprNodeDesc pruningPredicate;
+
/**
* A boolean variable set to true by the semantic analyzer only in case of the analyze command.
*
@@ -71,10 +77,32 @@ public class TableScanDesc extends Abstr
// input file name (big) to bucket number
private Map<String, Integer> bucketFileNameMapping;
+ static{
+ PTFUtils.makeTransient(TableScanDesc.class, "pruningPredicate");
+ PTFUtils.makeTransient(TableScanDesc.class, "table");
+ }
+
+
@SuppressWarnings("nls")
public TableScanDesc() {
}
+ public Table getTable() {
+ return table;
+ }
+
+ public void setTable(Table t) {
+ table = t;
+ }
+
+ public ExprNodeDesc getPruningPredicate() {
+ return pruningPredicate;
+ }
+
+ public void setPruningPredicate(ExprNodeDesc expr) {
+ pruningPredicate = expr;
+ }
+
public TableScanDesc(final String alias) {
this.alias = alias;
}