You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by na...@apache.org on 2010/11/17 18:33:11 UTC
svn commit: r1036128 [1/19] - in /hive/trunk: ./
common/src/java/org/apache/hadoop/hive/conf/ conf/
ql/src/java/org/apache/hadoop/hive/ql/
ql/src/java/org/apache/hadoop/hive/ql/exec/
ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/ ql/src/java/o...
Author: namit
Date: Wed Nov 17 17:33:06 2010
New Revision: 1036128
URL: http://svn.apache.org/viewvc?rev=1036128&view=rev
Log:
HIVE-1642 Convert join queries to map-join based on size of table/row
(Liyin Tang via namit)
Added:
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/PathUtil.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lib/TaskGraphWalker.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinResolver.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverCommonJoin.java
hive/trunk/ql/src/test/queries/clientpositive/auto_join0.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join1.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join10.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join11.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join12.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join13.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join14.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join15.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join16.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join17.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join18.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join18_multi_distinct.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join19.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join2.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join20.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join21.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join22.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join23.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join24.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join25.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join3.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join4.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join5.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join6.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join7.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join8.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join9.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join_filters.q
hive/trunk/ql/src/test/queries/clientpositive/auto_join_nulls.q
hive/trunk/ql/src/test/results/clientpositive/auto_join0.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join1.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join10.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join11.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join12.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join13.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join14.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join15.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join16.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join17.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join18.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join18_multi_distinct.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join19.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join2.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join20.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join21.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join22.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join23.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join24.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join25.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join3.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join4.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join5.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join6.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join7.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join8.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join9.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join_filters.q.out
hive/trunk/ql/src/test/results/clientpositive/auto_join_nulls.q.out
Modified:
hive/trunk/CHANGES.txt
hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
hive/trunk/conf/hive-default.xml
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecDriver.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecMapper.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MapRedTask.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MapredLocalTask.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/OperatorFactory.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/TaskFactory.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HashMapWrapper.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lib/Dispatcher.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkDeDuplication.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/GenMRSkewJoinProcessor.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/ASTNode.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/OpParseContext.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/QBJoinTree.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/RowResolver.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolver.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/HashTableSinkDesc.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MapJoinDesc.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredWork.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ppd/OpWalkerInfo.java
hive/trunk/ql/src/test/results/compiler/plan/case_sensitivity.q.xml
hive/trunk/ql/src/test/results/compiler/plan/cast1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby3.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby5.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input20.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input3.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input5.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input7.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input8.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input9.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input_part1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input_testsequencefile.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input_testxpath.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input_testxpath2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join3.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join5.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join7.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join8.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample3.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample5.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample7.q.xml
hive/trunk/ql/src/test/results/compiler/plan/subq.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf_case.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf_when.q.xml
hive/trunk/ql/src/test/results/compiler/plan/union.q.xml
hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/objectinspector/ObjectInspectorUtils.java
Modified: hive/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hive/trunk/CHANGES.txt?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/CHANGES.txt (original)
+++ hive/trunk/CHANGES.txt Wed Nov 17 17:33:06 2010
@@ -107,6 +107,9 @@ Trunk - Unreleased
HIVE-1779 Add udf str_to_map
(Siying Dong via namit)
+ HIVE-1642 Convert join queries to map-join based on size of table/row
+ (Liyin Tang via namit)
+
IMPROVEMENTS
HIVE-1712. Migrating metadata from derby to mysql thrown NullPointerException (Jake Farrell via pauly)
Modified: hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
URL: http://svn.apache.org/viewvc/hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java (original)
+++ hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java Wed Nov 17 17:33:06 2010
@@ -231,6 +231,8 @@ public class HiveConf extends Configurat
// mapper/reducer memory in local mode
HIVEHADOOPMAXMEM("hive.mapred.local.mem", 0),
+ //small table file size
+ HIVESMALLTABLESFILESIZE("hive.smalltable.filesize",25000000L), //25M
// test mode in hive mode
HIVETESTMODE("hive.test.mode", false),
HIVETESTMODEPREFIX("hive.test.mode.prefix", "test_"),
@@ -243,6 +245,7 @@ public class HiveConf extends Configurat
HIVEMERGEMAPFILESAVGSIZE("hive.merge.smallfiles.avgsize", (long) (16 * 1000 * 1000)),
HIVESKEWJOIN("hive.optimize.skewjoin", false),
+ HIVECONVERTJOIN("hive.auto.convert.join", false),
HIVESKEWJOINKEY("hive.skewjoin.key", 1000000),
HIVESKEWJOINMAPJOINNUMMAPTASK("hive.skewjoin.mapjoin.map.tasks", 10000),
HIVESKEWJOINMAPJOINMINSPLIT("hive.skewjoin.mapjoin.min.split", 33554432), //32M
@@ -251,6 +254,12 @@ public class HiveConf extends Configurat
HIVESENDHEARTBEAT("hive.heartbeat.interval", 1000),
HIVEMAXMAPJOINSIZE("hive.mapjoin.maxsize", 100000),
+ HIVEHASHTABLETHRESHOLD("hive.hashtable.initialCapacity", 100000),
+ HIVEHASHTABLELOADFACTOR("hive.hashtable.loadfactor", (float) 0.75),
+ HIVEHASHTABLEMAXMEMORYUSAGE("hive.hashtable.max.memory.usage", (float) 0.90),
+ HIVEHASHTABLESCALE("hive.hashtable.scale", (long)100000),
+
+ HIVEDEBUGLOCALTASK("hive.debug.localtask",false),
HIVEJOBPROGRESS("hive.task.progress", false),
Modified: hive/trunk/conf/hive-default.xml
URL: http://svn.apache.org/viewvc/hive/trunk/conf/hive-default.xml?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/conf/hive-default.xml (original)
+++ hive/trunk/conf/hive-default.xml Wed Nov 17 17:33:06 2010
@@ -473,6 +473,44 @@
<description>When the average output file size of a job is less than this number, Hive will start an additional map-reduce job to merge the output files into bigger files. This is only done for map-only jobs if hive.merge.mapfiles is true, and for map-reduce jobs if hive.merge.mapredfiles is true.</description>
</property>
+
+<property>
+ <name>hive.mapjoin.hashtable.initialCapacity</name>
+ <value>100000</value>
+ <description>In the process of Mapjoin, the key/value will be hold in the hashtable. This value means the initialCapacity of the in-memory hashtable</description>
+</property>
+
+<property>
+ <name>hive.mapjoin.hashtable.loadfactor</name>
+ <value>0.75</value>
+ <description>In the process of Mapjoin, the key/value will be hold in the hashtable. This value means the load factor for the in-memory hashtable</description>
+</property>
+
+<property>
+ <name>hive.mapjoin.smalltable.filesize</name>
+ <value>25000000</value>
+ <description>The threshold for the input file size of the small tables; if the file size is smaller than this threshold, it will try to convert the common join into map join</description>
+</property>
+
+<property>
+ <name>hive.mapjoin.localtask.max.memory.usage</name>
+ <value>0.90</value>
+ <description>This number means how much memory the local task can take to hold the key/value into in-memory hash table; If the local task's memory usage is more than this number, the local task will be abort by themself. It means the data of small table is too large to be hold in the memory.</description>
+</property>
+
+<property>
+ <name>hive.mapjoin.check.memory.rows</name>
+ <value>100000</value>
+ <description>The number means after how many rows processed it needs to check the memory usage</description>
+</property>
+
+<property>
+ <name>hive.auto.convert.join</name>
+ <value>false</value>
+ <description>Whether Hive enable the optimization about converting common join into mapjoin based on the input file size</description>
+</property>
+
+
<property>
<name>hive.script.auto.progress</name>
<value>false</value>
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java Wed Nov 17 17:33:06 2010
@@ -24,10 +24,10 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
-import java.util.HashSet;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@@ -124,7 +124,7 @@ public class Driver implements CommandPr
errorMessage = "FAILED: Error in semantic analysis: " + e.getMessage();
SQLState = ErrorMsg.findSQLState(e.getMessage());
console.printError(errorMessage, "\n"
- + org.apache.hadoop.util.StringUtils.stringifyException(e));
+ + org.apache.hadoop.util.StringUtils.stringifyException(e));
return (12);
}
}
@@ -140,8 +140,8 @@ public class Driver implements CommandPr
}
try {
- hiveLockMgr = (HiveLockManager)
- ReflectionUtils.newInstance(conf.getClassByName(lockMgr), conf);
+ hiveLockMgr = (HiveLockManager) ReflectionUtils.newInstance(conf.getClassByName(lockMgr),
+ conf);
hiveLockMgr.setContext(new HiveLockManagerCtx(conf));
} catch (Exception e) {
throw new SemanticException(ErrorMsg.LOCKMGR_NOT_INITIALIZED.getMsg() + e.getMessage());
@@ -197,8 +197,7 @@ public class Driver implements CommandPr
// Let's
// try to fetch the desc for the first partition and use it's
// deserializer.
- if (td == null && ft.getWork() != null
- && ft.getWork().getPartDesc() != null) {
+ if (td == null && ft.getWork() != null && ft.getWork().getPartDesc() != null) {
if (ft.getWork().getPartDesc().size() > 0) {
td = ft.getWork().getPartDesc().get(0).getTableDesc();
}
@@ -210,11 +209,10 @@ public class Driver implements CommandPr
String tableName = "result";
List<FieldSchema> lst = null;
try {
- lst = MetaStoreUtils.getFieldsFromDeserializer(
- tableName, td.getDeserializer());
+ lst = MetaStoreUtils.getFieldsFromDeserializer(tableName, td.getDeserializer());
} catch (Exception e) {
- LOG.warn("Error getting schema: " +
- org.apache.hadoop.util.StringUtils.stringifyException(e));
+ LOG.warn("Error getting schema: "
+ + org.apache.hadoop.util.StringUtils.stringifyException(e));
}
if (lst != null) {
schema = new Schema(lst, null);
@@ -296,8 +294,7 @@ public class Driver implements CommandPr
}
/**
- * Compile a new query. Any currently-planned query associated with this
- * Driver is discarded.
+ * Compile a new query. Any currently-planned query associated with this Driver is discarded.
*
* @param command
* The SQL query to compile.
@@ -321,15 +318,14 @@ public class Driver implements CommandPr
String hookName = HiveConf.getVar(conf, ConfVars.SEMANTIC_ANALYZER_HOOK);
// Do semantic analysis and plan generation
- if (hookName != null){
- AbstractSemanticAnalyzerHook hook = HiveUtils.getSemanticAnalyzerHook(conf, hookName);
+ if (hookName != null) {
+ AbstractSemanticAnalyzerHook hook = HiveUtils.getSemanticAnalyzerHook(conf, hookName);
HiveSemanticAnalyzerHookContext hookCtx = new HiveSemanticAnalyzerHookContextImpl();
hookCtx.setConf(conf);
hook.preAnalyze(hookCtx, tree);
sem.analyze(tree, ctx);
hook.postAnalyze(hookCtx, sem.getRootTasks());
- }
- else{
+ } else {
sem.analyze(tree, ctx);
}
@@ -348,19 +344,19 @@ public class Driver implements CommandPr
schema = getSchema(sem, conf);
// test Only - serialize the query plan and deserialize it
- if("true".equalsIgnoreCase(System.getProperty("test.serialize.qplan"))) {
+ if ("true".equalsIgnoreCase(System.getProperty("test.serialize.qplan"))) {
String queryPlanFileName = ctx.getLocalScratchDir(true) + Path.SEPARATOR_CHAR
- + "queryplan.xml";
+ + "queryplan.xml";
LOG.info("query plan = " + queryPlanFileName);
queryPlanFileName = new Path(queryPlanFileName).toUri().getPath();
- // serialize the queryPlan
+ // serialize the queryPlan
FileOutputStream fos = new FileOutputStream(queryPlanFileName);
Utilities.serializeQueryPlan(plan, fos);
fos.close();
- // deserialize the queryPlan
+ // deserialize the queryPlan
FileInputStream fis = new FileInputStream(queryPlanFileName);
QueryPlan newPlan = Utilities.deserializeQueryPlan(fis, conf);
fis.close();
@@ -405,10 +401,10 @@ public class Driver implements CommandPr
public static class LockObject {
HiveLockObject obj;
- HiveLockMode mode;
+ HiveLockMode mode;
public LockObject(HiveLockObject obj, HiveLockMode mode) {
- this.obj = obj;
+ this.obj = obj;
this.mode = mode;
}
@@ -447,13 +443,17 @@ public class Driver implements CommandPr
}
/**
- * @param t The table to be locked
- * @param p The partition to be locked
- * @param mode The mode of the lock (SHARED/EXCLUSIVE)
- * Get the list of objects to be locked. If a partition needs to be locked (in any mode), all its parents
- * should also be locked in SHARED mode.
+ * @param t
+ * The table to be locked
+ * @param p
+ * The partition to be locked
+ * @param mode
+ * The mode of the lock (SHARED/EXCLUSIVE) Get the list of objects to be locked. If a
+ * partition needs to be locked (in any mode), all its parents should also be locked in
+ * SHARED mode.
**/
- private List<LockObject> getLockObjects(Table t, Partition p, HiveLockMode mode) throws SemanticException {
+ private List<LockObject> getLockObjects(Table t, Partition p, HiveLockMode mode)
+ throws SemanticException {
List<LockObject> locks = new LinkedList<LockObject>();
if (t != null) {
@@ -476,14 +476,13 @@ public class Driver implements CommandPr
String partName = name;
String partialName = "";
String[] partns = name.split("/");
- for (int idx = 0; idx < partns.length -1; idx++) {
+ for (int idx = 0; idx < partns.length - 1; idx++) {
String partn = partns[idx];
partialName += partialName + partn;
try {
- locks.add(new LockObject(new HiveLockObject(
- new DummyPartition(p.getTable(),
- p.getTable().getDbName() + "@" + p.getTable().getTableName() + "@" + partialName),
- plan.getQueryId()), mode));
+ locks.add(new LockObject(new HiveLockObject(new DummyPartition(p.getTable(), p.getTable()
+ .getDbName()
+ + "@" + p.getTable().getTableName() + "@" + partialName), plan.getQueryId()), mode));
partialName += "/";
} catch (HiveException e) {
throw new SemanticException(e.getMessage());
@@ -496,10 +495,10 @@ public class Driver implements CommandPr
}
/**
- * Acquire read and write locks needed by the statement. The list of objects to be locked are obtained
- * from he inputs and outputs populated by the compiler. The lock acuisition scheme is pretty simple.
- * If all the locks cannot be obtained, error out. Deadlock is avoided by making sure that the locks
- * are lexicographically sorted.
+ * Acquire read and write locks needed by the statement. The list of objects to be locked are
+ * obtained from he inputs and outputs populated by the compiler. The lock acuisition scheme is
+ * pretty simple. If all the locks cannot be obtained, error out. Deadlock is avoided by making
+ * sure that the locks are lexicographically sorted.
**/
public int acquireReadWriteLocks() {
try {
@@ -515,12 +514,12 @@ public class Driver implements CommandPr
List<LockObject> lockObjects = new ArrayList<LockObject>();
// Sort all the inputs, outputs.
- // If a lock needs to be acquired on any partition, a read lock needs to be acquired on all its parents also
+ // If a lock needs to be acquired on any partition, a read lock needs to be acquired on all
+ // its parents also
for (ReadEntity input : plan.getInputs()) {
if (input.getType() == ReadEntity.Type.TABLE) {
lockObjects.addAll(getLockObjects(input.getTable(), null, HiveLockMode.SHARED));
- }
- else {
+ } else {
lockObjects.addAll(getLockObjects(null, input.getPartition(), HiveLockMode.SHARED));
}
}
@@ -528,9 +527,8 @@ public class Driver implements CommandPr
for (WriteEntity output : plan.getOutputs()) {
if (output.getTyp() == WriteEntity.Type.TABLE) {
lockObjects.addAll(getLockObjects(output.getTable(), null,
- output.isComplete() ? HiveLockMode.EXCLUSIVE : HiveLockMode.SHARED));
- }
- else if (output.getTyp() == WriteEntity.Type.PARTITION) {
+ output.isComplete() ? HiveLockMode.EXCLUSIVE : HiveLockMode.SHARED));
+ } else if (output.getTyp() == WriteEntity.Type.PARTITION) {
lockObjects.addAll(getLockObjects(null, output.getPartition(), HiveLockMode.EXCLUSIVE));
}
// In case of dynamic queries, it is possible to have incomplete dummy partitions
@@ -553,25 +551,26 @@ public class Driver implements CommandPr
Collections.sort(lockObjects, new Comparator<LockObject>() {
- @Override
- public int compare(LockObject o1, LockObject o2) {
- int cmp = o1.getName().compareTo(o2.getName());
- if (cmp == 0) {
- if (o1.getMode() == o2.getMode()) {
- return cmp;
- }
- // EXCLUSIVE locks occur before SHARED locks
- if (o1.getMode() == HiveLockMode.EXCLUSIVE) {
- return -1;
- }
- return +1;
+ @Override
+ public int compare(LockObject o1, LockObject o2) {
+ int cmp = o1.getName().compareTo(o2.getName());
+ if (cmp == 0) {
+ if (o1.getMode() == o2.getMode()) {
+ return cmp;
+ }
+ // EXCLUSIVE locks occur before SHARED locks
+ if (o1.getMode() == HiveLockMode.EXCLUSIVE) {
+ return -1;
}
- return cmp;
+ return +1;
}
+ return cmp;
+ }
- });
+ });
- // walk the list and acquire the locks - if any lock cant be acquired, release all locks, sleep and retry
+ // walk the list and acquire the locks - if any lock cant be acquired, release all locks,
+ // sleep and retry
LockObjectContainer notFound = new LockObjectContainer();
while (true) {
notFound.setLck(null);
@@ -579,8 +578,8 @@ public class Driver implements CommandPr
if (hiveLocks == null) {
if (tryNum == numRetries) {
- console.printError("Lock for " + notFound.getLck().getObj().getName() +
- " cannot be acquired in " + notFound.getLck().getMode());
+ console.printError("Lock for " + notFound.getLck().getObj().getName()
+ + " cannot be acquired in " + notFound.getLck().getMode());
throw new SemanticException(ErrorMsg.LOCK_CANNOT_BE_ACQUIRED.getMsg());
}
tryNum++;
@@ -589,8 +588,7 @@ public class Driver implements CommandPr
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
}
- }
- else {
+ } else {
ctx.setHiveLocks(hiveLocks);
break;
}
@@ -600,22 +598,25 @@ public class Driver implements CommandPr
errorMessage = "FAILED: Error in acquiring locks: " + e.getMessage();
SQLState = ErrorMsg.findSQLState(e.getMessage());
console.printError(errorMessage, "\n"
- + org.apache.hadoop.util.StringUtils.stringifyException(e));
+ + org.apache.hadoop.util.StringUtils.stringifyException(e));
return (10);
}
}
/**
- * @param lockObjects The list of objects to be locked
- * Lock the objects specified in the list. The same object is not locked twice, and the list passed is sorted
- * such that EXCLUSIVE locks occur before SHARED locks.
+ * @param lockObjects
+ * The list of objects to be locked Lock the objects specified in the list. The same
+ * object is not locked twice, and the list passed is sorted such that EXCLUSIVE locks
+ * occur before SHARED locks.
**/
- private List<HiveLock> acquireLocks(List<LockObject> lockObjects, LockObjectContainer notFound) throws SemanticException {
- // walk the list and acquire the locks - if any lock cant be acquired, release all locks, sleep and retry
+ private List<HiveLock> acquireLocks(List<LockObject> lockObjects, LockObjectContainer notFound)
+ throws SemanticException {
+ // walk the list and acquire the locks - if any lock cant be acquired, release all locks, sleep
+ // and retry
LockObject prevLockObj = null;
List<HiveLock> hiveLocks = new ArrayList<HiveLock>();
- for (LockObject lockObject: lockObjects) {
+ for (LockObject lockObject : lockObjects) {
// No need to acquire a lock twice on the same object
// It is ensured that EXCLUSIVE locks occur before SHARED locks on the same object
if ((prevLockObj != null) && (prevLockObj.getName().equals(lockObject.getName()))) {
@@ -644,8 +645,8 @@ public class Driver implements CommandPr
}
/**
- * Release all the locks acquired implicitly by the statement. Note that the locks acquired
- * with 'keepAlive' set to True are not released.
+ * Release all the locks acquired implicitly by the statement. Note that the locks acquired with
+ * 'keepAlive' set to True are not released.
**/
private void releaseLocks() {
if (ctx != null && ctx.getHiveLockMgr() != null) {
@@ -658,12 +659,13 @@ public class Driver implements CommandPr
}
/**
- * @param hiveLocks list of hive locks to be released
- * Release all the locks specified. If some of the locks have already been released, ignore them
+ * @param hiveLocks
+ * list of hive locks to be released Release all the locks specified. If some of the
+ * locks have already been released, ignore them
**/
private void releaseLocks(List<HiveLock> hiveLocks) {
if (hiveLocks != null) {
- for (HiveLock hiveLock: hiveLocks) {
+ for (HiveLock hiveLock : hiveLocks) {
try {
ctx.getHiveLockMgr().unlock(hiveLock);
} catch (LockException e) {
@@ -712,8 +714,8 @@ public class Driver implements CommandPr
for (String peClass : peClasses) {
try {
- pehooks.add((PreExecute) Class.forName(peClass.trim(), true,
- JavaUtils.getClassLoader()).newInstance());
+ pehooks.add((PreExecute) Class.forName(peClass.trim(), true, JavaUtils.getClassLoader())
+ .newInstance());
} catch (ClassNotFoundException e) {
console.printError("Pre Exec Hook Class not found:" + e.getMessage());
throw e;
@@ -735,8 +737,8 @@ public class Driver implements CommandPr
for (String peClass : peClasses) {
try {
- pehooks.add((PostExecute) Class.forName(peClass.trim(), true,
- JavaUtils.getClassLoader()).newInstance());
+ pehooks.add((PostExecute) Class.forName(peClass.trim(), true, JavaUtils.getClassLoader())
+ .newInstance());
} catch (ClassNotFoundException e) {
console.printError("Post Exec Hook Class not found:" + e.getMessage());
throw e;
@@ -747,8 +749,7 @@ public class Driver implements CommandPr
}
public int execute() {
- boolean noName = StringUtils.isEmpty(conf
- .getVar(HiveConf.ConfVars.HADOOPJOBNAME));
+ boolean noName = StringUtils.isEmpty(conf.getVar(HiveConf.ConfVars.HADOOPJOBNAME));
int maxlen = conf.getIntVar(HiveConf.ConfVars.HIVEJOBNAMELENGTH);
int curJobNo = 0;
@@ -774,8 +775,8 @@ public class Driver implements CommandPr
// Get all the pre execution hooks and execute them.
for (PreExecute peh : getPreExecHooks()) {
- peh.run(SessionState.get(), plan.getInputs(), plan.getOutputs(),
- ShimLoader.getHadoopShims().getUGIForConf(conf));
+ peh.run(SessionState.get(), plan.getInputs(), plan.getOutputs(), ShimLoader
+ .getHadoopShims().getUGIForConf(conf));
}
int jobs = Utilities.getMRTasks(plan.getRootTasks()).size();
@@ -783,10 +784,9 @@ public class Driver implements CommandPr
console.printInfo("Total MapReduce jobs = " + jobs);
}
if (SessionState.get() != null) {
- SessionState.get().getHiveHistory().setQueryProperty(queryId,
- Keys.QUERY_NUM_TASKS, String.valueOf(jobs));
- SessionState.get().getHiveHistory().setIdToTableMap(
- plan.getIdToTableNameMap());
+ SessionState.get().getHiveHistory().setQueryProperty(queryId, Keys.QUERY_NUM_TASKS,
+ String.valueOf(jobs));
+ SessionState.get().getHiveHistory().setIdToTableMap(plan.getIdToTableNameMap());
}
String jobname = Utilities.abbreviate(queryStr, maxlen - 6);
@@ -823,20 +823,37 @@ public class Driver implements CommandPr
int exitVal = tskRes.getExitVal();
if (exitVal != 0) {
- // TODO: This error messaging is not very informative. Fix that.
- errorMessage = "FAILED: Execution Error, return code " + exitVal
- + " from " + tsk.getClass().getName();
- SQLState = "08S01";
- console.printError(errorMessage);
- if (running.size() != 0) {
- taskCleanup();
+ Task<? extends Serializable> backupTask = tsk.getAndInitBackupTask();
+ if (backupTask != null) {
+ errorMessage = "FAILED: Execution Error, return code " + exitVal + " from "
+ + tsk.getClass().getName();
+ console.printInfo(errorMessage);
+
+ errorMessage = "ATTEMPT: Execute BackupTask: " + backupTask.getClass().getName();
+ console.printInfo(errorMessage);
+
+ // add backup task to runnable
+ if (DriverContext.isLaunchable(backupTask)) {
+ driverCxt.addToRunnable(backupTask);
+ }
+ continue;
+
+ } else {
+ // TODO: This error messaging is not very informative. Fix that.
+ errorMessage = "FAILED: Execution Error, return code " + exitVal + " from "
+ + tsk.getClass().getName();
+ SQLState = "08S01";
+ console.printError(errorMessage);
+ if (running.size() != 0) {
+ taskCleanup();
+ }
+ return 9;
}
- return 9;
}
if (SessionState.get() != null) {
- SessionState.get().getHiveHistory().setTaskProperty(queryId,
- tsk.getId(), Keys.TASK_RET_CODE, String.valueOf(exitVal));
+ SessionState.get().getHiveHistory().setTaskProperty(queryId, tsk.getId(),
+ Keys.TASK_RET_CODE, String.valueOf(exitVal));
SessionState.get().getHiveHistory().endTask(queryId, tsk);
}
@@ -870,19 +887,19 @@ public class Driver implements CommandPr
// Get all the post execution hooks and execute them.
for (PostExecute peh : getPostExecHooks()) {
peh.run(SessionState.get(), plan.getInputs(), plan.getOutputs(),
- (SessionState.get() != null ? SessionState.get().getLineageState().getLineageInfo() : null),
- ShimLoader.getHadoopShims().getUGIForConf(conf));
+ (SessionState.get() != null ? SessionState.get().getLineageState().getLineageInfo()
+ : null), ShimLoader.getHadoopShims().getUGIForConf(conf));
}
if (SessionState.get() != null) {
- SessionState.get().getHiveHistory().setQueryProperty(queryId,
- Keys.QUERY_RET_CODE, String.valueOf(0));
+ SessionState.get().getHiveHistory().setQueryProperty(queryId, Keys.QUERY_RET_CODE,
+ String.valueOf(0));
SessionState.get().getHiveHistory().printRowCount(queryId);
}
} catch (Exception e) {
if (SessionState.get() != null) {
- SessionState.get().getHiveHistory().setQueryProperty(queryId,
- Keys.QUERY_RET_CODE, String.valueOf(12));
+ SessionState.get().getHiveHistory().setQueryProperty(queryId, Keys.QUERY_RET_CODE,
+ String.valueOf(12));
}
// TODO: do better with handling types of Exception here
errorMessage = "FAILED: Hive Internal Error: " + Utilities.getNameMessage(e);
@@ -931,18 +948,15 @@ public class Driver implements CommandPr
* @return the updated number of last the map-reduce job launched
*/
- public void launchTask(Task<? extends Serializable> tsk, String queryId,
- boolean noName, Map<TaskResult, TaskRunner> running, String jobname,
- int jobs, DriverContext cxt) {
+ public void launchTask(Task<? extends Serializable> tsk, String queryId, boolean noName,
+ Map<TaskResult, TaskRunner> running, String jobname, int jobs, DriverContext cxt) {
if (SessionState.get() != null) {
- SessionState.get().getHiveHistory().startTask(queryId, tsk,
- tsk.getClass().getName());
+ SessionState.get().getHiveHistory().startTask(queryId, tsk, tsk.getClass().getName());
}
if (tsk.isMapRedTask() && !(tsk instanceof ConditionalTask)) {
if (noName) {
- conf.setVar(HiveConf.ConfVars.HADOOPJOBNAME, jobname + "("
- + tsk.getId() + ")");
+ conf.setVar(HiveConf.ConfVars.HADOOPJOBNAME, jobname + "(" + tsk.getId() + ")");
}
cxt.incCurJobNo(1);
console.printInfo("Launching Job " + cxt.getCurJobNo() + " out of " + jobs);
@@ -952,8 +966,7 @@ public class Driver implements CommandPr
TaskRunner tskRun = new TaskRunner(tsk, tskRes);
// Launch Task
- if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.EXECPARALLEL)
- && tsk.isMapRedTask()) {
+ if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.EXECPARALLEL) && tsk.isMapRedTask()) {
// Launch it in the parallel mode, as a separate thread only for MR tasks
tskRun.start();
} else {
@@ -1045,8 +1058,7 @@ public class Driver implements CommandPr
res.add(row);
}
} catch (IOException e) {
- console.printError("FAILED: Unexpected IO exception : "
- + e.getMessage());
+ console.printError("FAILED: Unexpected IO exception : " + e.getMessage());
res = null;
return false;
}
@@ -1076,8 +1088,7 @@ public class Driver implements CommandPr
releaseLocks();
}
- public org.apache.hadoop.hive.ql.plan.api.Query getQueryPlan()
- throws IOException {
+ public org.apache.hadoop.hive.ql.plan.api.Query getQueryPlan() throws IOException {
return plan.getQueryPlan();
}
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java Wed Nov 17 17:33:06 2010
@@ -49,7 +49,7 @@ public class ColumnInfo implements Seria
private boolean isVirtualCol;
private transient TypeInfo type;
-
+
private boolean isHiddenVirtualCol;
public ColumnInfo() {
@@ -66,7 +66,7 @@ public class ColumnInfo implements Seria
.getPrimitiveTypeInfoFromPrimitiveWritable(type), tabAlias,
isVirtualCol, false);
}
-
+
public ColumnInfo(String internalName, TypeInfo type, String tabAlias,
boolean isVirtualCol, boolean isHiddenVirtualCol) {
this.internalName = internalName;
@@ -99,7 +99,7 @@ public class ColumnInfo implements Seria
public boolean getIsVirtualCol() {
return isVirtualCol;
}
-
+
public boolean isHiddenVirtualCol() {
return isHiddenVirtualCol;
}
@@ -120,4 +120,17 @@ public class ColumnInfo implements Seria
return alias;
}
+ public void setTabAlias(String tabAlias) {
+ this.tabAlias = tabAlias;
+ }
+
+ public void setVirtualCol(boolean isVirtualCol) {
+ this.isVirtualCol = isVirtualCol;
+ }
+
+ public void setHiddenVirtualCol(boolean isHiddenVirtualCol) {
+ this.isHiddenVirtualCol = isHiddenVirtualCol;
+ }
+
+
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java Wed Nov 17 17:33:06 2010
@@ -437,9 +437,10 @@ transient boolean newGroupStarted = fals
boolean newObjNull) {
// newObj is null if is already null or
// if the row corresponding to the left alias does not pass through filter
- newObjNull = newObjNull ||
- ((BooleanWritable) (intObj.getObjs()[left].get(
- joinValues.get(order[left]).size()))).get();
+ int filterIndex = joinValues.get(order[left]).size();
+ if(filterIndex < intObj.getObjs()[left].size()) {
+ newObjNull = newObjNull || ((BooleanWritable) (intObj.getObjs()[left].get(filterIndex))).get();
+ }
Iterator<boolean[]> nullsIter = inputNulls.iterator();
while (nullsIter.hasNext()) {
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java Wed Nov 17 17:33:06 2010
@@ -32,13 +32,13 @@ import org.apache.hadoop.hive.ql.plan.ap
/**
* Conditional Task implementation.
*/
-public class ConditionalTask extends Task<ConditionalWork> implements
- Serializable {
+public class ConditionalTask extends Task<ConditionalWork> implements Serializable {
private static final long serialVersionUID = 1L;
private List<Task<? extends Serializable>> listTasks;
private boolean resolved = false;
+
private List<Task<? extends Serializable>> resTasks;
private ConditionalResolver resolver;
@@ -48,6 +48,7 @@ public class ConditionalTask extends Tas
super();
}
+
@Override
public boolean isMapRedTask() {
for (Task<? extends Serializable> task : listTasks) {
@@ -71,8 +72,7 @@ public class ConditionalTask extends Tas
}
@Override
- public void initialize(HiveConf conf, QueryPlan queryPlan,
- DriverContext driverContext) {
+ public void initialize(HiveConf conf, QueryPlan queryPlan, DriverContext driverContext) {
super.initialize(conf, queryPlan, driverContext);
}
@@ -80,30 +80,28 @@ public class ConditionalTask extends Tas
public int execute(DriverContext driverContext) {
resTasks = resolver.getTasks(conf, resolverCtx);
resolved = true;
+
for (Task<? extends Serializable> tsk : getListTasks()) {
if (!resTasks.contains(tsk)) {
driverContext.getRunnable().remove(tsk);
- console.printInfo(ExecDriver.getJobEndMsg(""
- + Utilities.randGen.nextInt())
+ console.printInfo(ExecDriver.getJobEndMsg("" + Utilities.randGen.nextInt())
+ ", job is filtered out (removed at runtime).");
- if(tsk.isMapRedTask()) {
+ if (tsk.isMapRedTask()) {
driverContext.incCurJobNo(1);
}
- if (tsk.getChildTasks() != null) {
- for (Task<? extends Serializable> child : tsk.getChildTasks()) {
- child.parentTasks.remove(tsk);
- if (DriverContext.isLaunchable(child)) {
- driverContext.addToRunnable(child);
- }
- }
+ //recursively remove this task from its children's parent task
+ tsk.removeFromChildrenTasks();
+ } else {
+ // resolved task
+ if (!driverContext.getRunnable().contains(tsk)) {
+ driverContext.addToRunnable(tsk);
}
- } else if (!driverContext.getRunnable().contains(tsk)) {
- driverContext.addToRunnable(tsk);
}
}
return 0;
}
+
/**
* @return the resolver
*/
@@ -184,9 +182,8 @@ public class ConditionalTask extends Tas
}
/**
- * Add a dependent task on the current conditional task. The task will not be
- * a direct child of conditional task. Actually it will be added as child task
- * of associated tasks.
+ * Add a dependent task on the current conditional task. The task will not be a direct child of
+ * conditional task. Actually it will be added as child task of associated tasks.
*
* @return true if the task got added false if it already existed
*/
@@ -204,7 +201,7 @@ public class ConditionalTask extends Tas
@Override
protected void localizeMRTmpFilesImpl(Context ctx) {
if (getListTasks() != null) {
- for(Task<? extends Serializable> t: getListTasks()) {
+ for (Task<? extends Serializable> t : getListTasks()) {
t.localizeMRTmpFiles(ctx);
}
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecDriver.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecDriver.java?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecDriver.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecDriver.java Wed Nov 17 17:33:06 2010
@@ -104,6 +104,7 @@ public class ExecDriver extends Task<Map
protected transient int mapProgress = 0;
protected transient int reduceProgress = 0;
public static MemoryMXBean memoryMXBean;
+
/**
* Constructor when invoked from QL.
*/
@@ -111,8 +112,7 @@ public class ExecDriver extends Task<Map
super();
}
- protected static String getResourceFiles(Configuration conf,
- SessionState.ResourceType t) {
+ protected static String getResourceFiles(Configuration conf, SessionState.ResourceType t) {
// fill in local files to be added to the task environment
SessionState ss = SessionState.get();
Set<String> files = (ss == null) ? null : ss.list_resource(t, null);
@@ -122,8 +122,8 @@ public class ExecDriver extends Task<Map
try {
realFiles.add(Utilities.realFile(one, conf));
} catch (IOException e) {
- throw new RuntimeException("Cannot validate file " + one
- + "due to exception: " + e.getMessage(), e);
+ throw new RuntimeException("Cannot validate file " + one + "due to exception: "
+ + e.getMessage(), e);
}
}
return StringUtils.join(realFiles, ",");
@@ -143,8 +143,7 @@ public class ExecDriver extends Task<Map
* Initialization when invoked from QL.
*/
@Override
- public void initialize(HiveConf conf, QueryPlan queryPlan,
- DriverContext driverContext) {
+ public void initialize(HiveConf conf, QueryPlan queryPlan, DriverContext driverContext) {
super.initialize(conf, queryPlan, driverContext);
job = new JobConf(conf, ExecDriver.class);
@@ -164,8 +163,7 @@ public class ExecDriver extends Task<Map
if (StringUtils.isNotBlank(addedJars)) {
HiveConf.setVar(job, ConfVars.HIVEADDEDJARS, addedJars);
}
- String addedArchives = getResourceFiles(job,
- SessionState.ResourceType.ARCHIVE);
+ String addedArchives = getResourceFiles(job, SessionState.ResourceType.ARCHIVE);
if (StringUtils.isNotBlank(addedArchives)) {
HiveConf.setVar(job, ConfVars.HIVEADDEDARCHIVES, addedArchives);
}
@@ -182,24 +180,24 @@ public class ExecDriver extends Task<Map
}
/**
- * A list of the currently running jobs spawned in this Hive instance that is
- * used to kill all running jobs in the event of an unexpected shutdown -
- * i.e., the JVM shuts down while there are still jobs running.
+ * A list of the currently running jobs spawned in this Hive instance that is used to kill all
+ * running jobs in the event of an unexpected shutdown - i.e., the JVM shuts down while there are
+ * still jobs running.
*/
- private static Map<String, String> runningJobKillURIs =
- Collections.synchronizedMap(new HashMap<String, String>());
+ private static Map<String, String> runningJobKillURIs = Collections
+ .synchronizedMap(new HashMap<String, String>());
/**
- * In Hive, when the user control-c's the command line, any running jobs
- * spawned from that command line are best-effort killed.
+ * In Hive, when the user control-c's the command line, any running jobs spawned from that command
+ * line are best-effort killed.
*
- * This static constructor registers a shutdown thread to iterate over all the
- * running job kill URLs and do a get on them.
+ * This static constructor registers a shutdown thread to iterate over all the running job kill
+ * URLs and do a get on them.
*
*/
static {
- if (new org.apache.hadoop.conf.Configuration().getBoolean(
- "webinterface.private.actions", false)) {
+ if (new org.apache.hadoop.conf.Configuration()
+ .getBoolean("webinterface.private.actions", false)) {
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
@@ -207,13 +205,13 @@ public class ExecDriver extends Task<Map
for (String uri : runningJobKillURIs.values()) {
try {
System.err.println("killing job with: " + uri);
- java.net.HttpURLConnection conn = (java.net.HttpURLConnection) new java.net.URL(
- uri).openConnection();
+ java.net.HttpURLConnection conn = (java.net.HttpURLConnection) new java.net.URL(uri)
+ .openConnection();
conn.setRequestMethod("POST");
int retCode = conn.getResponseCode();
if (retCode != 200) {
- System.err.println("Got an error trying to kill job with URI: "
- + uri + " = " + retCode);
+ System.err.println("Got an error trying to kill job with URI: " + uri + " = "
+ + retCode);
}
} catch (Exception e) {
System.err.println("trying to kill job, caught: " + e);
@@ -235,22 +233,19 @@ public class ExecDriver extends Task<Map
} else {
String hp = job.get("mapred.job.tracker");
if (SessionState.get() != null) {
- SessionState.get().getHiveHistory().setTaskProperty(
- SessionState.get().getQueryId(), getId(), Keys.TASK_HADOOP_ID,
- rj.getJobID());
- }
- console.printInfo(ExecDriver.getJobStartMsg(rj.getJobID())
- + ", Tracking URL = " + rj.getTrackingURL());
- console.printInfo("Kill Command = "
- + HiveConf.getVar(job, HiveConf.ConfVars.HADOOPBIN)
+ SessionState.get().getHiveHistory().setTaskProperty(SessionState.get().getQueryId(),
+ getId(), Keys.TASK_HADOOP_ID, rj.getJobID());
+ }
+ console.printInfo(ExecDriver.getJobStartMsg(rj.getJobID()) + ", Tracking URL = "
+ + rj.getTrackingURL());
+ console.printInfo("Kill Command = " + HiveConf.getVar(job, HiveConf.ConfVars.HADOOPBIN)
+ " job -Dmapred.job.tracker=" + hp + " -kill " + rj.getJobID());
}
}
/**
- * This class contains the state of the running task Going forward, we will
- * return this handle from execute and Driver can split execute into start,
- * monitorProgess and postProcess.
+ * This class contains the state of the running task Going forward, we will return this handle
+ * from execute and Driver can split execute into start, monitorProgess and postProcess.
*/
private static class ExecDriverTaskHandle extends TaskHandle {
JobClient jc;
@@ -280,16 +275,14 @@ public class ExecDriver extends Task<Map
}
/**
- * Fatal errors are those errors that cannot be recovered by retries. These
- * are application dependent. Examples of fatal errors include: - the small
- * table in the map-side joins is too large to be feasible to be handled by
- * one mapper. The job should fail and the user should be warned to use
- * regular joins rather than map-side joins. Fatal errors are indicated by
- * counters that are set at execution time. If the counter is non-zero, a
- * fatal error occurred. The value of the counter indicates the error type.
+ * Fatal errors are those errors that cannot be recovered by retries. These are application
+ * dependent. Examples of fatal errors include: - the small table in the map-side joins is too
+ * large to be feasible to be handled by one mapper. The job should fail and the user should be
+ * warned to use regular joins rather than map-side joins. Fatal errors are indicated by counters
+ * that are set at execution time. If the counter is non-zero, a fatal error occurred. The value
+ * of the counter indicates the error type.
*
- * @return true if fatal errors happened during job execution, false
- * otherwise.
+ * @return true if fatal errors happened during job execution, false otherwise.
*/
private boolean checkFatalErrors(Counters ctrs, StringBuilder errMsg) {
if (ctrs == null) {
@@ -299,7 +292,7 @@ public class ExecDriver extends Task<Map
}
// check for number of created files
long numFiles = ctrs.getCounter(ProgressCounter.CREATED_FILES);
- long upperLimit = HiveConf.getLongVar(job, HiveConf.ConfVars.MAXCREATEDFILES);
+ long upperLimit = HiveConf.getLongVar(job, HiveConf.ConfVars.MAXCREATEDFILES);
if (numFiles > upperLimit) {
errMsg.append("total number of created files exceeds ").append(upperLimit);
return true;
@@ -335,8 +328,7 @@ public class ExecDriver extends Task<Map
} catch (InterruptedException e) {
}
- if (initializing &&
- ShimLoader.getHadoopShims().isJobPreparing(rj)) {
+ if (initializing && ShimLoader.getHadoopShims().isJobPreparing(rj)) {
// No reason to poll untill the job is initialized
continue;
} else {
@@ -366,8 +358,7 @@ public class ExecDriver extends Task<Map
Counters ctrs = th.getCounters();
if (fatal = checkFatalErrors(ctrs, errMsg)) {
- console.printError("[Fatal Error] " + errMsg.toString()
- + ". Killing the job.");
+ console.printError("[Fatal Error] " + errMsg.toString() + ". Killing the job.");
rj.killJob();
continue;
}
@@ -375,8 +366,8 @@ public class ExecDriver extends Task<Map
updateCounters(ctrs, rj);
- String report = " " + getId() + " map = " + mapProgress + "%, reduce = "
- + reduceProgress + "%";
+ String report = " " + getId() + " map = " + mapProgress + "%, reduce = " + reduceProgress
+ + "%";
if (!report.equals(lastReport)
|| System.currentTimeMillis() >= reportTime + maxReportInterval) {
@@ -386,12 +377,10 @@ public class ExecDriver extends Task<Map
String output = dateFormat.format(Calendar.getInstance().getTime()) + report;
SessionState ss = SessionState.get();
if (ss != null) {
- ss.getHiveHistory().setTaskCounters(SessionState.get().getQueryId(),
- getId(), ctrs);
- ss.getHiveHistory().setTaskProperty(SessionState.get().getQueryId(),
- getId(), Keys.TASK_HADOOP_PROGRESS, output);
- ss.getHiveHistory().progressTask(SessionState.get().getQueryId(),
- this);
+ ss.getHiveHistory().setTaskCounters(SessionState.get().getQueryId(), getId(), ctrs);
+ ss.getHiveHistory().setTaskProperty(SessionState.get().getQueryId(), getId(),
+ Keys.TASK_HADOOP_PROGRESS, output);
+ ss.getHiveHistory().progressTask(SessionState.get().getQueryId(), this);
ss.getHiveHistory().logPlanProgress(queryPlan);
}
console.printInfo(output);
@@ -411,7 +400,7 @@ public class ExecDriver extends Task<Map
if (checkFatalErrors(ctrs, errMsg)) {
console.printError("[Fatal Error] " + errMsg.toString());
success = false;
- } else {
+ } else {
success = rj.isSuccessful();
}
}
@@ -434,10 +423,8 @@ public class ExecDriver extends Task<Map
private void updateCounters(Counters ctrs, RunningJob rj) throws IOException {
mapProgress = Math.round(rj.mapProgress() * 100);
reduceProgress = Math.round(rj.reduceProgress() * 100);
- taskCounters.put("CNTR_NAME_" + getId() + "_MAP_PROGRESS", Long
- .valueOf(mapProgress));
- taskCounters.put("CNTR_NAME_" + getId() + "_REDUCE_PROGRESS", Long
- .valueOf(reduceProgress));
+ taskCounters.put("CNTR_NAME_" + getId() + "_MAP_PROGRESS", Long.valueOf(mapProgress));
+ taskCounters.put("CNTR_NAME_" + getId() + "_REDUCE_PROGRESS", Long.valueOf(reduceProgress));
if (ctrs == null) {
// hadoop might return null if it cannot locate the job.
// we may still be able to retrieve the job status - so ignore
@@ -512,8 +499,8 @@ public class ExecDriver extends Task<Map
job.setMapOutputValueClass(BytesWritable.class);
try {
- job.setPartitionerClass((Class<? extends Partitioner>)
- (Class.forName(HiveConf.getVar(job, HiveConf.ConfVars.HIVEPARTITIONER))));
+ job.setPartitionerClass((Class<? extends Partitioner>) (Class.forName(HiveConf.getVar(job,
+ HiveConf.ConfVars.HIVEPARTITIONER))));
} catch (ClassNotFoundException e) {
throw new RuntimeException(e.getMessage());
}
@@ -522,8 +509,8 @@ public class ExecDriver extends Task<Map
job.setNumMapTasks(work.getNumMapTasks().intValue());
}
if (work.getMinSplitSize() != null) {
- HiveConf.setIntVar(job, HiveConf.ConfVars.MAPREDMINSPLITSIZE,
- work.getMinSplitSize().intValue());
+ HiveConf.setIntVar(job, HiveConf.ConfVars.MAPREDMINSPLITSIZE, work.getMinSplitSize()
+ .intValue());
}
job.setNumReduceTasks(work.getNumReduceTasks().intValue());
job.setReducerClass(ExecReducer.class);
@@ -533,11 +520,9 @@ public class ExecDriver extends Task<Map
}
// Turn on speculative execution for reducers
- boolean useSpeculativeExecReducers =
- HiveConf.getBoolVar(job, HiveConf.ConfVars.HIVESPECULATIVEEXECREDUCERS);
- HiveConf.setBoolVar(
- job,
- HiveConf.ConfVars.HADOOPSPECULATIVEEXECREDUCERS,
+ boolean useSpeculativeExecReducers = HiveConf.getBoolVar(job,
+ HiveConf.ConfVars.HIVESPECULATIVEEXECREDUCERS);
+ HiveConf.setBoolVar(job, HiveConf.ConfVars.HADOOPSPECULATIVEEXECREDUCERS,
useSpeculativeExecReducers);
String inpFormat = HiveConf.getVar(job, HiveConf.ConfVars.HIVEINPUTFORMAT);
@@ -548,8 +533,7 @@ public class ExecDriver extends Task<Map
LOG.info("Using " + inpFormat);
try {
- job.setInputFormat((Class<? extends InputFormat>) (Class
- .forName(inpFormat)));
+ job.setInputFormat((Class<? extends InputFormat>) (Class.forName(inpFormat)));
} catch (ClassNotFoundException e) {
throw new RuntimeException(e.getMessage());
}
@@ -563,8 +547,9 @@ public class ExecDriver extends Task<Map
String auxJars = HiveConf.getVar(job, HiveConf.ConfVars.HIVEAUXJARS);
String addedJars = HiveConf.getVar(job, HiveConf.ConfVars.HIVEADDEDJARS);
if (StringUtils.isNotBlank(auxJars) || StringUtils.isNotBlank(addedJars)) {
- String allJars = StringUtils.isNotBlank(auxJars) ? (StringUtils
- .isNotBlank(addedJars) ? addedJars + "," + auxJars : auxJars)
+ String allJars = StringUtils.isNotBlank(auxJars) ? (StringUtils.isNotBlank(addedJars) ? addedJars
+ + "," + auxJars
+ : auxJars)
: addedJars;
LOG.info("adding libjars: " + allJars);
initializeFiles("tmpjars", allJars);
@@ -576,8 +561,7 @@ public class ExecDriver extends Task<Map
initializeFiles("tmpfiles", addedFiles);
}
// Transfer HIVEADDEDARCHIVES to "tmparchives" so hadoop understands it
- String addedArchives = HiveConf.getVar(job,
- HiveConf.ConfVars.HIVEADDEDARCHIVES);
+ String addedArchives = HiveConf.getVar(job, HiveConf.ConfVars.HIVEADDEDARCHIVES);
if (StringUtils.isNotBlank(addedArchives)) {
initializeFiles("tmparchives", addedArchives);
}
@@ -585,21 +569,18 @@ public class ExecDriver extends Task<Map
int returnVal = 0;
RunningJob rj = null;
- boolean noName = StringUtils.isEmpty(HiveConf.getVar(job,
- HiveConf.ConfVars.HADOOPJOBNAME));
+ boolean noName = StringUtils.isEmpty(HiveConf.getVar(job, HiveConf.ConfVars.HADOOPJOBNAME));
if (noName) {
// This is for a special case to ensure unit tests pass
- HiveConf.setVar(job, HiveConf.ConfVars.HADOOPJOBNAME, "JOB"
- + Utilities.randGen.nextInt());
+ HiveConf.setVar(job, HiveConf.ConfVars.HADOOPJOBNAME, "JOB" + Utilities.randGen.nextInt());
}
try {
- //propagate the file to distributed cache
- MapredLocalWork localwork =work.getMapLocalWork();
- if(localwork != null){
+ // propagate the file to distributed cache
+ MapredLocalWork localwork = work.getMapLocalWork();
+ if (localwork != null) {
boolean localMode = HiveConf.getVar(job, HiveConf.ConfVars.HADOOPJT).equals("local");
- if(!localMode){
- //LOG.info("=========Adding files to HDFS ================");
+ if (!localMode) {
Path localPath = new Path(localwork.getTmpFileURI());
Path hdfsPath = new Path(work.getTmpHDFSFileURI());
@@ -609,28 +590,24 @@ public class ExecDriver extends Task<Map
for (int i = 0; i < hashtableFiles.length; i++) {
FileStatus file = hashtableFiles[i];
Path path = file.getPath();
- String fileName=path.getName();
- String hdfsFile = hdfsPath+Path.SEPARATOR+fileName;
+ String fileName = path.getName();
+ String hdfsFile = hdfsPath + Path.SEPARATOR + fileName;
LOG.info("Upload 1 HashTable from" + path + " to: " + hdfsFile);
Path hdfsFilePath = new Path(hdfsFile);
- //hdfs.setVerifyChecksum(false);
- hdfs.copyFromLocalFile(path,hdfsFilePath );
- short replication = (short)job.getInt("mapred.submit.replication", 10);
+ hdfs.copyFromLocalFile(path, hdfsFilePath);
+ short replication = (short) job.getInt("mapred.submit.replication", 10);
hdfs.setReplication(hdfsFilePath, replication);
}
- //LOG.info("=========Adding files to distributed cache================");
-
FileStatus[] hashtableRemoteFiles = hdfs.listStatus(hdfsPath);
for (int i = 0; i < hashtableRemoteFiles.length; i++) {
FileStatus file = hashtableRemoteFiles[i];
Path path = file.getPath();
DistributedCache.addCacheFile(path.toUri(), job);
- LOG.info("add 1 hashtable file to distributed cache: "
- + path.toUri());
+
+ LOG.info("add 1 hashtable file to distributed cache: " + path.toUri());
}
- //LOG.info("===========Finishing adding files to distributed cache==========");
}
}
@@ -670,8 +647,7 @@ public class ExecDriver extends Task<Map
// add to list of running jobs to kill in case of abnormal shutdown
- runningJobKillURIs.put(rj.getJobID(), rj.getTrackingURL()
- + "&action=kill");
+ runningJobKillURIs.put(rj.getJobID(), rj.getTrackingURL() + "&action=kill");
ExecDriverTaskHandle th = new ExecDriverTaskHandle(jc, rj);
jobInfo(rj);
@@ -701,15 +677,14 @@ public class ExecDriver extends Task<Map
// Has to use full name to make sure it does not conflict with
// org.apache.commons.lang.StringUtils
- console.printError(mesg, "\n"
- + org.apache.hadoop.util.StringUtils.stringifyException(e));
+ console.printError(mesg, "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
success = false;
returnVal = 1;
} finally {
Utilities.clearMapRedWork(job);
try {
- if(ctxCreated) {
+ if (ctxCreated) {
ctx.clear();
}
@@ -728,8 +703,7 @@ public class ExecDriver extends Task<Map
if (rj != null) {
JobCloseFeedBack feedBack = new JobCloseFeedBack();
if (work.getAliasToWork() != null) {
- for (Operator<? extends Serializable> op : work.getAliasToWork()
- .values()) {
+ for (Operator<? extends Serializable> op : work.getAliasToWork().values()) {
op.jobClose(job, success, feedBack);
}
}
@@ -742,10 +716,8 @@ public class ExecDriver extends Task<Map
if (success) {
success = false;
returnVal = 3;
- String mesg = "Job Commit failed with exception '"
- + Utilities.getNameMessage(e) + "'";
- console.printError(mesg, "\n"
- + org.apache.hadoop.util.StringUtils.stringifyException(e));
+ String mesg = "Job Commit failed with exception '" + Utilities.getNameMessage(e) + "'";
+ console.printError(mesg, "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
}
}
@@ -772,8 +744,7 @@ public class ExecDriver extends Task<Map
return "Ended Job = " + jobId;
}
- private String getTaskAttemptLogUrl(String taskTrackerHttpAddress,
- String taskAttemptId) {
+ private String getTaskAttemptLogUrl(String taskTrackerHttpAddress, String taskAttemptId) {
return taskTrackerHttpAddress + "/tasklog?taskid=" + taskAttemptId + "&all=true";
}
@@ -786,12 +757,15 @@ public class ExecDriver extends Task<Map
this.jobId = jobId;
logUrls = new HashSet<String>();
}
+
public void addLogUrl(String logUrl) {
logUrls.add(logUrl);
}
+
public HashSet<String> getLogUrls() {
return logUrls;
}
+
public String getJobId() {
return jobId;
}
@@ -811,8 +785,7 @@ public class ExecDriver extends Task<Map
// Loop to get all task completion events because getTaskCompletionEvents
// only returns a subset per call
while (true) {
- TaskCompletionEvent[] taskCompletions =
- rj.getTaskCompletionEvents(startIndex);
+ TaskCompletionEvent[] taskCompletions = rj.getTaskCompletionEvents(startIndex);
if (taskCompletions == null || taskCompletions.length == 0) {
break;
@@ -836,14 +809,13 @@ public class ExecDriver extends Task<Map
String jobId = taskJobIds[1];
TaskInfo ti = taskIdToInfo.get(taskId);
- if(ti == null) {
+ if (ti == null) {
ti = new TaskInfo(jobId);
taskIdToInfo.put(taskId, ti);
}
// These tasks should have come from the same job.
- assert(ti.getJobId() == jobId);
- ti.getLogUrls().add(
- getTaskAttemptLogUrl(t.getTaskTrackerHttp(), t.getTaskId()));
+ assert (ti.getJobId() == jobId);
+ ti.getLogUrls().add(getTaskAttemptLogUrl(t.getTaskTrackerHttp(), t.getTaskId()));
// If a task failed, then keep track of the total number of failures
// for that task (typically, a task gets re-run up to 4 times if it
@@ -889,11 +861,10 @@ public class ExecDriver extends Task<Map
if (failures.get(task).intValue() == maxFailures) {
TaskInfo ti = taskIdToInfo.get(task);
String jobId = ti.getJobId();
- String taskUrl = jtUrl + "/taskdetails.jsp?jobid=" + jobId + "&tipid=" +
- task.toString();
+ String taskUrl = jtUrl + "/taskdetails.jsp?jobid=" + jobId + "&tipid=" + task.toString();
TaskLogProcessor tlp = new TaskLogProcessor(conf);
- for(String logUrl : ti.getLogUrls()) {
+ for (String logUrl : ti.getLogUrls()) {
tlp.addTaskAttemptLogUrl(logUrl);
}
@@ -910,7 +881,7 @@ public class ExecDriver extends Task<Map
sb.append("Task ID:\n " + task + "\n\n");
sb.append("URL:\n " + taskUrl + "\n");
- for(ErrorAndSolution e : errors) {
+ for (ErrorAndSolution e : errors) {
sb.append("\n");
sb.append("Possible error:\n " + e.getError() + "\n\n");
sb.append("Solution:\n " + e.getSolution() + "\n");
@@ -934,36 +905,29 @@ public class ExecDriver extends Task<Map
}
/**
- * we are running the hadoop job via a sub-command. this typically
- * happens when we are running jobs in local mode. the log4j in this
- * mode is controlled as follows:
- * 1. if the admin provides a log4j properties file especially for
- * execution mode - then we pick that up
- * 2. otherwise - we default to the regular hive log4j properties if
- * one is supplied
- * 3. if none of the above two apply - we don't do anything - the log4j
- * properties would likely be determined by hadoop.
+ * we are running the hadoop job via a sub-command. this typically happens when we are running
+ * jobs in local mode. the log4j in this mode is controlled as follows: 1. if the admin provides a
+ * log4j properties file especially for execution mode - then we pick that up 2. otherwise - we
+ * default to the regular hive log4j properties if one is supplied 3. if none of the above two
+ * apply - we don't do anything - the log4j properties would likely be determined by hadoop.
*
- * The intention behind providing a separate option #1 is to be able to
- * collect hive run time logs generated in local mode in a separate
- * (centralized) location if desired. This mimics the behavior of hive
- * run time logs when running against a hadoop cluster where they are available
+ * The intention behind providing a separate option #1 is to be able to collect hive run time logs
+ * generated in local mode in a separate (centralized) location if desired. This mimics the
+ * behavior of hive run time logs when running against a hadoop cluster where they are available
* on the tasktracker nodes.
*/
private static void setupChildLog4j(Configuration conf) {
URL hive_l4j = ExecDriver.class.getClassLoader().getResource(SessionState.HIVE_EXEC_L4J);
- if(hive_l4j == null) {
- hive_l4j = ExecDriver.class.getClassLoader().getResource
- (SessionState.HIVE_L4J);
+ if (hive_l4j == null) {
+ hive_l4j = ExecDriver.class.getClassLoader().getResource(SessionState.HIVE_L4J);
}
if (hive_l4j != null) {
// setting queryid so that log4j configuration can use it to generate
// per query log file
- System.setProperty
- (HiveConf.ConfVars.HIVEQUERYID.toString(),
- HiveConf.getVar(conf, HiveConf.ConfVars.HIVEQUERYID));
+ System.setProperty(HiveConf.ConfVars.HIVEQUERYID.toString(), HiveConf.getVar(conf,
+ HiveConf.ConfVars.HIVEQUERYID));
LogManager.resetConfiguration();
PropertyConfigurator.configure(hive_l4j);
}
@@ -975,8 +939,7 @@ public class ExecDriver extends Task<Map
ArrayList<String> jobConfArgs = new ArrayList<String>();
boolean noLog = false;
String files = null;
- boolean localtask=false;
-
+ boolean localtask = false;
try {
for (int i = 0; i < args.length; i++) {
if (args[i].equals("-plan")) {
@@ -987,7 +950,7 @@ public class ExecDriver extends Task<Map
noLog = true;
} else if (args[i].equals("-files")) {
files = args[++i];
- }else if(args[i].equals("-localtask")) {
+ } else if (args[i].equals("-localtask")) {
localtask = true;
}
}
@@ -996,11 +959,11 @@ public class ExecDriver extends Task<Map
printUsage();
}
- JobConf conf ;
- if(localtask) {
- conf= new JobConf(MapredLocalTask.class);
+ JobConf conf;
+ if (localtask) {
+ conf = new JobConf(MapredLocalTask.class);
} else {
- conf= new JobConf(ExecDriver.class);
+ conf = new JobConf(ExecDriver.class);
}
StringBuilder sb = new StringBuilder("JobConf:\n");
@@ -1013,8 +976,8 @@ public class ExecDriver extends Task<Map
conf.set(key, value);
sb.append(key).append("=").append(value).append("\n");
} catch (UnsupportedEncodingException e) {
- System.err.println("Unexpected error " + e.getMessage()
- + " while encoding " + one.substring(eqIndex + 1));
+ System.err.println("Unexpected error " + e.getMessage() + " while encoding "
+ + one.substring(eqIndex + 1));
System.exit(3);
}
}
@@ -1045,10 +1008,10 @@ public class ExecDriver extends Task<Map
// print out the location of the log file for the user so
// that it's easy to find reason for local mode execution failures
- for (Appender appender: Collections.list
- ((Enumeration<Appender>)LogManager.getRootLogger().getAllAppenders())) {
+ for (Appender appender : Collections.list((Enumeration<Appender>) LogManager.getRootLogger()
+ .getAllAppenders())) {
if (appender instanceof FileAppender) {
- console.printInfo("Execution log at: " + ((FileAppender)appender).getFile());
+ console.printInfo("Execution log at: " + ((FileAppender) appender).getFile());
}
}
@@ -1085,11 +1048,11 @@ public class ExecDriver extends Task<Map
throw new HiveException(e.getMessage(), e);
}
int ret;
- if(localtask) {
+ if (localtask) {
memoryMXBean = ManagementFactory.getMemoryMXBean();
MapredLocalWork plan = Utilities.deserializeMapRedLocalWork(pathData, conf);
MapredLocalTask ed = new MapredLocalTask(plan, conf, isSilent);
- ret= ed.executeFromChildJVM(new DriverContext());
+ ret = ed.executeFromChildJVM(new DriverContext());
} else {
MapredWork plan = Utilities.deserializeMapRedWork(pathData, conf);
@@ -1103,8 +1066,8 @@ public class ExecDriver extends Task<Map
}
/**
- * Given a Hive Configuration object - generate a command line fragment for
- * passing such configuration information to ExecDriver.
+ * Given a Hive Configuration object - generate a command line fragment for passing such
+ * configuration information to ExecDriver.
*/
public static String generateCmdLine(HiveConf hconf) {
try {
@@ -1117,8 +1080,7 @@ public class ExecDriver extends Task<Map
for (Object one : deltaP.keySet()) {
String oneProp = (String) one;
- if (hadoopLocalMode
- && (oneProp.equals(hadoopSysDir) || oneProp.equals(hadoopWorkDir))) {
+ if (hadoopLocalMode && (oneProp.equals(hadoopSysDir) || oneProp.equals(hadoopWorkDir))) {
continue;
}
@@ -1139,15 +1101,15 @@ public class ExecDriver extends Task<Map
sb.append("-jobconf ");
sb.append(hadoopSysDir);
sb.append("=");
- sb.append(URLEncoder.encode(hconf.get(hadoopSysDir) + "/"
- + Utilities.randGen.nextInt(), "UTF-8"));
+ sb.append(URLEncoder.encode(hconf.get(hadoopSysDir) + "/" + Utilities.randGen.nextInt(),
+ "UTF-8"));
sb.append(" ");
sb.append("-jobconf ");
sb.append(hadoopWorkDir);
sb.append("=");
- sb.append(URLEncoder.encode(hconf.get(hadoopWorkDir) + "/"
- + Utilities.randGen.nextInt(), "UTF-8"));
+ sb.append(URLEncoder.encode(hconf.get(hadoopWorkDir) + "/" + Utilities.randGen.nextInt(),
+ "UTF-8"));
}
return sb.toString();
@@ -1170,9 +1132,8 @@ public class ExecDriver extends Task<Map
/**
* Handle a empty/null path for a given alias.
*/
- private int addInputPath(String path, JobConf job, MapredWork work,
- String hiveScratchDir, int numEmptyPaths, boolean isEmptyPath,
- String alias) throws Exception {
+ private int addInputPath(String path, JobConf job, MapredWork work, String hiveScratchDir,
+ int numEmptyPaths, boolean isEmptyPath, String alias) throws Exception {
// either the directory does not exist or it is empty
assert path == null || isEmptyPath;
@@ -1184,7 +1145,7 @@ public class ExecDriver extends Task<Map
outFileFormat = partDesc.getOutputFileFormatClass();
nonNative = partDesc.getTableDesc().isNonNative();
} else {
- TableDesc tableDesc = work.getAliasToPartnInfo().get(alias).getTableDesc();
+ TableDesc tableDesc = work.getAliasToPartnInfo().get(alias).getTableDesc();
outFileFormat = tableDesc.getOutputFileFormatClass();
nonNative = tableDesc.isNonNative();
}
@@ -1237,15 +1198,14 @@ public class ExecDriver extends Task<Map
work.setPathToPartitionInfo(pathToPartitionInfo);
String onefile = newPath.toString();
- RecordWriter recWriter = outFileFormat.newInstance().getHiveRecordWriter(
- job, newFilePath, Text.class, false, new Properties(), null);
+ RecordWriter recWriter = outFileFormat.newInstance().getHiveRecordWriter(job, newFilePath,
+ Text.class, false, new Properties(), null);
recWriter.close(false);
FileInputFormat.addInputPaths(job, onefile);
return numEmptyPaths;
}
- private void addInputPaths(JobConf job, MapredWork work, String hiveScratchDir)
- throws Exception {
+ private void addInputPaths(JobConf job, MapredWork work, String hiveScratchDir) throws Exception {
int numEmptyPaths = 0;
List<String> pathsProcessed = new ArrayList<String>();
@@ -1282,8 +1242,8 @@ public class ExecDriver extends Task<Map
// Create a empty file if the directory is empty
for (String emptyPath : emptyPaths) {
- numEmptyPaths = addInputPath(emptyPath, job, work, hiveScratchDir,
- numEmptyPaths, true, oneAlias);
+ numEmptyPaths = addInputPath(emptyPath, job, work, hiveScratchDir, numEmptyPaths, true,
+ oneAlias);
}
// If the query references non-existent partitions
@@ -1295,8 +1255,8 @@ public class ExecDriver extends Task<Map
// If T is empty and T2 contains 100 rows, the user expects: 0, 100 (2
// rows)
if (path == null) {
- numEmptyPaths = addInputPath(null, job, work, hiveScratchDir,
- numEmptyPaths, false, oneAlias);
+ numEmptyPaths = addInputPath(null, job, work, hiveScratchDir, numEmptyPaths, false,
+ oneAlias);
}
}
}
@@ -1315,15 +1275,15 @@ public class ExecDriver extends Task<Map
protected void localizeMRTmpFilesImpl(Context ctx) {
// localize any map-reduce input paths
- ctx.localizeKeys((Map<String, Object>)((Object)work.getPathToAliases()));
- ctx.localizeKeys((Map<String, Object>)((Object)work.getPathToPartitionInfo()));
+ ctx.localizeKeys((Map<String, Object>) ((Object) work.getPathToAliases()));
+ ctx.localizeKeys((Map<String, Object>) ((Object) work.getPathToPartitionInfo()));
// localize any input paths for maplocal work
MapredLocalWork l = work.getMapLocalWork();
if (l != null) {
Map<String, FetchWork> m = l.getAliasToFetchWork();
if (m != null) {
- for (FetchWork fw: m.values()) {
+ for (FetchWork fw : m.values()) {
String s = fw.getTblDir();
if ((s != null) && ctx.isMRTmpFileURI(s)) {
fw.setTblDir(ctx.localizeMRTmpFileURI(s));
@@ -1335,22 +1295,21 @@ public class ExecDriver extends Task<Map
// fix up outputs
Map<String, ArrayList<String>> pa = work.getPathToAliases();
if (pa != null) {
- for (List<String> ls: pa.values()) {
- for (String a: ls) {
- ArrayList<Operator<? extends Serializable>> opList = new
- ArrayList<Operator<? extends Serializable>> ();
+ for (List<String> ls : pa.values()) {
+ for (String a : ls) {
+ ArrayList<Operator<? extends Serializable>> opList = new ArrayList<Operator<? extends Serializable>>();
opList.add(work.getAliasToWork().get(a));
while (!opList.isEmpty()) {
Operator<? extends Serializable> op = opList.remove(0);
if (op instanceof FileSinkOperator) {
- FileSinkDesc fdesc = ((FileSinkOperator)op).getConf();
+ FileSinkDesc fdesc = ((FileSinkOperator) op).getConf();
String s = fdesc.getDirName();
if ((s != null) && ctx.isMRTmpFileURI(s)) {
fdesc.setDirName(ctx.localizeMRTmpFileURI(s));
}
- ((FileSinkOperator)op).setConf(fdesc);
+ ((FileSinkOperator) op).setConf(fdesc);
}
if (op.getChildOperators() != null) {
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecMapper.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecMapper.java?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecMapper.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExecMapper.java Wed Nov 17 17:33:06 2010
@@ -89,7 +89,7 @@ public class ExecMapper extends MapReduc
// initialize map operator
mo.setChildren(job);
l4j.info(mo.dump(0));
- // initialize map local work
+ // initialize map local work
localWork = mrwork.getMapLocalWork();
execContext.setLocalWork(localWork);
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java?rev=1036128&r1=1036127&r2=1036128&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExplainTask.java Wed Nov 17 17:33:06 2010
@@ -28,17 +28,17 @@ import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
+import java.util.Map.Entry;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.Context;
+import org.apache.hadoop.hive.ql.DriverContext;
import org.apache.hadoop.hive.ql.plan.Explain;
import org.apache.hadoop.hive.ql.plan.ExplainWork;
import org.apache.hadoop.hive.ql.plan.api.StageType;
import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.hive.ql.DriverContext;
-import org.apache.hadoop.hive.ql.Context;
/**
@@ -304,7 +304,7 @@ public class ExplainTask extends Task<Ex
return;
}
dependeciesTaskSet.add(task);
-
+ boolean first = true;
out.print(indentString(indent));
out.printf("%s", task.getId());
if ((task.getParentTasks() == null || task.getParentTasks().isEmpty())) {
@@ -313,7 +313,7 @@ public class ExplainTask extends Task<Ex
}
} else {
out.print(" depends on stages: ");
- boolean first = true;
+ first = true;
for (Task<? extends Serializable> parent : task.getParentTasks()) {
if (!first) {
out.print(", ");
@@ -321,22 +321,34 @@ public class ExplainTask extends Task<Ex
first = false;
out.print(parent.getId());
}
+ }
- if (task instanceof ConditionalTask
- && ((ConditionalTask) task).getListTasks() != null) {
- out.print(" , consists of ");
- first = true;
- for (Task<? extends Serializable> con : ((ConditionalTask) task)
- .getListTasks()) {
- if (!first) {
- out.print(", ");
- }
- first = false;
- out.print(con.getId());
- }
+ Task<? extends Serializable> cuurBackupTask = task.getBackupTask();
+ if(cuurBackupTask != null) {
+ out.print(" has a backup stage: ");
+ if (!first) {
+ out.print(", ");
}
+ first = false;
+ out.print(cuurBackupTask.getId());
+ }
+
+ if (task instanceof ConditionalTask
+ && ((ConditionalTask) task).getListTasks() != null) {
+ out.print(" , consists of ");
+ first = true;
+ for (Task<? extends Serializable> con : ((ConditionalTask) task)
+ .getListTasks()) {
+ if (!first) {
+ out.print(", ");
+ }
+ first = false;
+ out.print(con.getId());
+ }
}
+
+
out.println();
if (task instanceof ConditionalTask
@@ -406,7 +418,7 @@ public class ExplainTask extends Task<Ex
@Override
protected void localizeMRTmpFilesImpl(Context ctx) {
- // explain task has nothing to localize
+ // explain task has nothing to localize
// we don't expect to enter this code path at all
throw new RuntimeException ("Unexpected call");
}