hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hashut...@apache.org
Subject svn commit: r1506785 [4/8] - in /hive/branches/vectorization: ./ beeline/src/java/org/apache/hive/beeline/ cli/src/java/org/apache/hadoop/hive/cli/ common/src/java/org/apache/hadoop/hive/conf/ conf/ contrib/src/test/results/clientnegative/ data/files/ ...
Date Thu, 25 Jul 2013 00:15:07 GMT
Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/WriterImpl.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/WriterImpl.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/WriterImpl.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/orc/WriterImpl.java Thu Jul 25 00:14:58 2013
@@ -1484,7 +1484,10 @@ class WriterImpl implements Writer, Memo
     OrcProto.PostScript.Builder builder =
       OrcProto.PostScript.newBuilder()
         .setCompression(writeCompressionKind(compress))
-        .setFooterLength(footerLength);
+        .setFooterLength(footerLength)
+        .setMagic(OrcFile.MAGIC)
+        .addVersion(OrcFile.MAJOR_VERSION)
+        .addVersion(OrcFile.MINOR_VERSION);
     if (compress != CompressionKind.NONE) {
       builder.setCompressionBlockSize(bufferSize);
     }

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/merge/BlockMergeTask.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/merge/BlockMergeTask.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/merge/BlockMergeTask.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/merge/BlockMergeTask.java Thu Jul 25 00:14:58 2013
@@ -35,11 +35,11 @@ import org.apache.hadoop.hive.conf.HiveC
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.DriverContext;
 import org.apache.hadoop.hive.ql.QueryPlan;
-import org.apache.hadoop.hive.ql.exec.HadoopJobExecHelper;
-import org.apache.hadoop.hive.ql.exec.HadoopJobExecHook;
 import org.apache.hadoop.hive.ql.exec.Task;
-import org.apache.hadoop.hive.ql.exec.Throttle;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.HadoopJobExecHelper;
+import org.apache.hadoop.hive.ql.exec.mr.HadoopJobExecHook;
+import org.apache.hadoop.hive.ql.exec.mr.Throttle;
 import org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
 import org.apache.hadoop.hive.ql.io.HiveOutputFormatImpl;
 import org.apache.hadoop.hive.ql.plan.api.StageType;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/stats/PartialScanTask.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/stats/PartialScanTask.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/stats/PartialScanTask.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/stats/PartialScanTask.java Thu Jul 25 00:14:58 2013
@@ -36,11 +36,11 @@ import org.apache.hadoop.hive.ql.Context
 import org.apache.hadoop.hive.ql.DriverContext;
 import org.apache.hadoop.hive.ql.ErrorMsg;
 import org.apache.hadoop.hive.ql.QueryPlan;
-import org.apache.hadoop.hive.ql.exec.HadoopJobExecHelper;
-import org.apache.hadoop.hive.ql.exec.HadoopJobExecHook;
 import org.apache.hadoop.hive.ql.exec.Task;
-import org.apache.hadoop.hive.ql.exec.Throttle;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.HadoopJobExecHelper;
+import org.apache.hadoop.hive.ql.exec.mr.HadoopJobExecHook;
+import org.apache.hadoop.hive.ql.exec.mr.Throttle;
 import org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
 import org.apache.hadoop.hive.ql.io.HiveOutputFormatImpl;
 import org.apache.hadoop.hive.ql.metadata.HiveException;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateTask.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateTask.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateTask.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateTask.java Thu Jul 25 00:14:58 2013
@@ -28,11 +28,11 @@ import org.apache.hadoop.hive.conf.HiveC
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.DriverContext;
 import org.apache.hadoop.hive.ql.QueryPlan;
-import org.apache.hadoop.hive.ql.exec.HadoopJobExecHelper;
-import org.apache.hadoop.hive.ql.exec.HadoopJobExecHook;
 import org.apache.hadoop.hive.ql.exec.Task;
-import org.apache.hadoop.hive.ql.exec.Throttle;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.HadoopJobExecHelper;
+import org.apache.hadoop.hive.ql.exec.mr.HadoopJobExecHook;
+import org.apache.hadoop.hive.ql.exec.mr.Throttle;
 import org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;
 import org.apache.hadoop.hive.ql.io.HiveOutputFormatImpl;
 import org.apache.hadoop.hive.ql.plan.api.StageType;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/lib/DefaultGraphWalker.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/lib/DefaultGraphWalker.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/lib/DefaultGraphWalker.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/lib/DefaultGraphWalker.java Thu Jul 25 00:14:58 2013
@@ -75,6 +75,13 @@ public class DefaultGraphWalker implemen
    * @throws SemanticException
    */
   public void dispatch(Node nd, Stack<Node> ndStack) throws SemanticException {
+    dispatchAndReturn(nd, ndStack);
+  }
+
+  /**
+   * Returns dispatch result
+   */
+  public <T> T dispatchAndReturn(Node nd, Stack<Node> ndStack) throws SemanticException {
     Object[] nodeOutputs = null;
     if (nd.getChildren() != null) {
       nodeOutputs = new Object[nd.getChildren().size()];
@@ -86,6 +93,7 @@ public class DefaultGraphWalker implemen
 
     Object retVal = dispatcher.dispatch(nd, ndStack, nodeOutputs);
     retMap.put(nd, retVal);
+    return (T) retVal;
   }
 
   /**

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockObject.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockObject.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockObject.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockObject.java Thu Jul 25 00:14:58 2013
@@ -36,17 +36,29 @@ public class HiveLockObject {
     private String queryStr;
     private String clientIp;
 
+    /**
+     * Constructor
+     *
+     * Note: The parameters are used to uniquely identify a HiveLockObject. 
+     * The parameters will be stripped off any ':' characters in order not 
+     * to interfere with the way the data is serialized (':' delimited string).
+     */
     public HiveLockObjectData(String queryId,
         String lockTime,
         String lockMode,
         String queryStr) {
-      this.queryId = queryId;
-      this.lockTime = lockTime;
-      this.lockMode = lockMode;
-      this.queryStr = queryStr.trim();
+      this.queryId = removeDelimiter(queryId);
+      this.lockTime = removeDelimiter(lockTime);
+      this.lockMode = removeDelimiter(lockMode);
+      this.queryStr = removeDelimiter(queryStr.trim());
     }
 
-
+    /**
+     * Constructor
+     * 
+     * @param data String of the form "queryId:lockTime:lockMode:queryStr". 
+     * No ':' characters are allowed in any of the components.
+     */
     public HiveLockObjectData(String data) {
       if (data == null) {
         return;
@@ -203,4 +215,11 @@ public class HiveLockObject {
         data == null ? tgt.getData() == null :
         tgt.getData() != null && data.equals(tgt.getData());
   }
+
+  private static String removeDelimiter(String in) {
+    if (in == null) {
+      return null;
+    }
+    return in.replaceAll(":","");
+  }
 }

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/metadata/Partition.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/metadata/Partition.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/metadata/Partition.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/metadata/Partition.java Thu Jul 25 00:14:58 2013
@@ -47,6 +47,7 @@ import org.apache.hadoop.hive.ql.io.Hive
 import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
 import org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat;
 import org.apache.hadoop.hive.serde2.Deserializer;
+import org.apache.hadoop.hive.serde2.SerDeUtils;
 import org.apache.hadoop.mapred.InputFormat;
 import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TBinaryProtocol;
@@ -200,7 +201,7 @@ public class Partition implements Serial
           }
         }
         // set default if columns are not set
-        if (tPartition.getSd().getCols() == null || tPartition.getSd().getCols().size() == 0) {
+        if (tPartition.getSd().getCols() == null) {
           if (table.getCols() != null) {
             tPartition.getSd().setCols(table.getCols());
           }
@@ -519,7 +520,19 @@ public class Partition implements Serial
   }
 
   public List<FieldSchema> getCols() {
-    return tPartition.getSd().getCols();
+    if (!SerDeUtils.shouldGetColsFromSerDe(
+        tPartition.getSd().getSerdeInfo().getSerializationLib())) {
+      return tPartition.getSd().getCols();
+    }
+
+    try {
+      return Hive.getFieldsFromDeserializer(table.getTableName(), getDeserializer());
+    } catch (HiveException e) {
+      LOG.error("Unable to get cols from serde: " +
+          tPartition.getSd().getSerdeInfo().getSerializationLib(), e);
+    }
+
+    return new ArrayList<FieldSchema>();
   }
 
   public String getLocation() {

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java Thu Jul 25 00:14:58 2013
@@ -91,6 +91,8 @@ public class GenMRFileSink1 implements N
     ParseContext parseCtx = ctx.getParseCtx();
     boolean chDir = false;
     Task<? extends Serializable> currTask = ctx.getCurrTask();
+    ctx.addRootIfPossible(currTask);
+
     FileSinkOperator fsOp = (FileSinkOperator) nd;
     boolean isInsertTable = // is INSERT OVERWRITE TABLE
     fsOp.getConf().getTableInfo().getTableName() != null &&
@@ -106,7 +108,7 @@ public class GenMRFileSink1 implements N
     if (fileSinkDescs != null) {
       Task<? extends Serializable> childTask = fileSinkDescs.get(fsOp.getConf());
       processLinkedFileDesc(ctx, childTask);
-      return null;
+      return true;
     }
 
     // Has the user enabled merging of files for map-only jobs or for all jobs
@@ -181,7 +183,7 @@ public class GenMRFileSink1 implements N
       }
     }
 
-    return null;
+    return true;
   }
 
   /*
@@ -189,26 +191,12 @@ public class GenMRFileSink1 implements N
    * Use the task created by the first linked file descriptor
    */
   private void processLinkedFileDesc(GenMRProcContext ctx,
-    Task<? extends Serializable> childTask)
-    throws SemanticException {
-    Operator<? extends OperatorDesc> currTopOp = ctx.getCurrTopOp();
-    String currAliasId = ctx.getCurrAliasId();
-    List<Operator<? extends OperatorDesc>> seenOps = ctx.getSeenOps();
-    List<Task<? extends Serializable>> rootTasks = ctx.getRootTasks();
+      Task<? extends Serializable> childTask) throws SemanticException {
     Task<? extends Serializable> currTask = ctx.getCurrTask();
-
-    if (currTopOp != null) {
-      if (!seenOps.contains(currTopOp)) {
-        seenOps.add(currTopOp);
-        GenMapRedUtils.setTaskPlan(currAliasId, currTopOp,
-          (MapredWork) currTask.getWork(), false, ctx);
-      }
-
-      if (!rootTasks.contains(currTask)
-          && (currTask.getParentTasks() == null
-              || currTask.getParentTasks().isEmpty())) {
-        rootTasks.add(currTask);
-      }
+    Operator<? extends OperatorDesc> currTopOp = ctx.getCurrTopOp();
+    if (currTopOp != null && !ctx.isSeenOp(currTask, currTopOp)) {
+      String currAliasId = ctx.getCurrAliasId();
+      GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, currTask, false, ctx);
     }
 
     if (childTask != null) {
@@ -702,8 +690,6 @@ public class GenMRFileSink1 implements N
     String currAliasId = ctx.getCurrAliasId();
     HashMap<Operator<? extends OperatorDesc>, Task<? extends Serializable>> opTaskMap =
         ctx.getOpTaskMap();
-    List<Operator<? extends OperatorDesc>> seenOps = ctx.getSeenOps();
-    List<Task<? extends Serializable>> rootTasks = ctx.getRootTasks();
 
     // Set the move task to be dependent on the current task
     if (mvTask != null) {
@@ -717,22 +703,13 @@ public class GenMRFileSink1 implements N
     if (currTopOp != null) {
       Task<? extends Serializable> mapTask = opTaskMap.get(null);
       if (mapTask == null) {
-        if (!seenOps.contains(currTopOp)) {
-          seenOps.add(currTopOp);
-          GenMapRedUtils.setTaskPlan(currAliasId, currTopOp,
-              (MapredWork) currTask.getWork(), false, ctx);
+        if (!ctx.isSeenOp(currTask, currTopOp)) {
+          GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, currTask, false, ctx);
         }
         opTaskMap.put(null, currTask);
-        if (!rootTasks.contains(currTask)
-            && (currTask.getParentTasks() == null
-                || currTask.getParentTasks().isEmpty())) {
-          rootTasks.add(currTask);
-        }
       } else {
-        if (!seenOps.contains(currTopOp)) {
-          seenOps.add(currTopOp);
-          GenMapRedUtils.setTaskPlan(currAliasId, currTopOp,
-              (MapredWork) mapTask.getWork(), false, ctx);
+        if (!ctx.isSeenOp(currTask, currTopOp)) {
+          GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, mapTask, false, ctx);
         } else {
           UnionOperator currUnionOp = ctx.getCurrUnionOp();
           if (currUnionOp != null) {

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMROperator.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMROperator.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMROperator.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMROperator.java Thu Jul 25 00:14:58 2013
@@ -53,8 +53,7 @@ public class GenMROperator implements No
         .getMapCurrCtx();
     GenMapRedCtx mapredCtx = mapCurrCtx.get(stack.get(stack.size() - 2));
     mapCurrCtx.put((Operator<? extends OperatorDesc>) nd, new GenMapRedCtx(
-        mapredCtx.getCurrTask(), mapredCtx.getCurrTopOp(), mapredCtx
-        .getCurrAliasId()));
-    return null;
+        mapredCtx.getCurrTask(), mapredCtx.getCurrAliasId()));
+    return true;
   }
 }

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRProcContext.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRProcContext.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRProcContext.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRProcContext.java Thu Jul 25 00:14:58 2013
@@ -55,7 +55,6 @@ public class GenMRProcContext implements
    */
   public static class GenMapRedCtx {
     Task<? extends Serializable> currTask;
-    Operator<? extends OperatorDesc> currTopOp;
     String currAliasId;
 
     public GenMapRedCtx() {
@@ -64,15 +63,10 @@ public class GenMRProcContext implements
     /**
      * @param currTask
      *          the current task
-     * @param currTopOp
-     *          the current top operator being traversed
      * @param currAliasId
-     *          the current alias for the to operator
      */
-    public GenMapRedCtx(Task<? extends Serializable> currTask,
-        Operator<? extends OperatorDesc> currTopOp, String currAliasId) {
+    public GenMapRedCtx(Task<? extends Serializable> currTask, String currAliasId) {
       this.currTask = currTask;
-      this.currTopOp = currTopOp;
       this.currAliasId = currAliasId;
     }
 
@@ -84,13 +78,6 @@ public class GenMRProcContext implements
     }
 
     /**
-     * @return current top operator
-     */
-    public Operator<? extends OperatorDesc> getCurrTopOp() {
-      return currTopOp;
-    }
-
-    /**
      * @return current alias
      */
     public String getCurrAliasId() {
@@ -103,13 +90,13 @@ public class GenMRProcContext implements
    *
    */
   public static class GenMRUnionCtx {
-    Task<? extends Serializable> uTask;
+    final Task<? extends Serializable> uTask;
     List<String> taskTmpDir;
     List<TableDesc> tt_desc;
     List<Operator<? extends OperatorDesc>> listTopOperators;
 
-    public GenMRUnionCtx() {
-      uTask = null;
+    public GenMRUnionCtx(Task<? extends Serializable> uTask) {
+      this.uTask = uTask;
       taskTmpDir = new ArrayList<String>();
       tt_desc = new ArrayList<TableDesc>();
       listTopOperators = new ArrayList<Operator<? extends OperatorDesc>>();
@@ -119,10 +106,6 @@ public class GenMRProcContext implements
       return uTask;
     }
 
-    public void setUTask(Task<? extends Serializable> uTask) {
-      this.uTask = uTask;
-    }
-
     public void addTaskTmpDir(String taskTmpDir) {
       this.taskTmpDir.add(taskTmpDir);
     }
@@ -156,8 +139,10 @@ public class GenMRProcContext implements
   private HiveConf conf;
   private
     HashMap<Operator<? extends OperatorDesc>, Task<? extends Serializable>> opTaskMap;
+  private
+    HashMap<Task<? extends Serializable>, List<Operator<? extends OperatorDesc>>> taskToSeenOps;
+
   private HashMap<UnionOperator, GenMRUnionCtx> unionTaskMap;
-  private List<Operator<? extends OperatorDesc>> seenOps;
   private List<FileSinkOperator> seenFileSinkOps;
 
   private ParseContext parseCtx;
@@ -169,7 +154,6 @@ public class GenMRProcContext implements
   private Operator<? extends OperatorDesc> currTopOp;
   private UnionOperator currUnionOp;
   private String currAliasId;
-  private List<Operator<? extends OperatorDesc>> rootOps;
   private DependencyCollectionTask dependencyTaskForMultiInsert;
 
   // If many fileSinkDescs are linked to each other, it is a good idea to keep track of
@@ -213,14 +197,13 @@ public class GenMRProcContext implements
   public GenMRProcContext(
       HiveConf conf,
       HashMap<Operator<? extends OperatorDesc>, Task<? extends Serializable>> opTaskMap,
-      List<Operator<? extends OperatorDesc>> seenOps, ParseContext parseCtx,
+      ParseContext parseCtx,
       List<Task<MoveWork>> mvTask,
       List<Task<? extends Serializable>> rootTasks,
       LinkedHashMap<Operator<? extends OperatorDesc>, GenMapRedCtx> mapCurrCtx,
       Set<ReadEntity> inputs, Set<WriteEntity> outputs) {
     this.conf = conf;
     this.opTaskMap = opTaskMap;
-    this.seenOps = seenOps;
     this.mvTask = mvTask;
     this.parseCtx = parseCtx;
     this.rootTasks = rootTasks;
@@ -231,9 +214,9 @@ public class GenMRProcContext implements
     currTopOp = null;
     currUnionOp = null;
     currAliasId = null;
-    rootOps = new ArrayList<Operator<? extends OperatorDesc>>();
-    rootOps.addAll(parseCtx.getTopOps().values());
     unionTaskMap = new HashMap<UnionOperator, GenMRUnionCtx>();
+    taskToSeenOps = new HashMap<Task<? extends Serializable>,
+        List<Operator<? extends OperatorDesc>>>();
     dependencyTaskForMultiInsert = null;
     linkedFileDescTasks = null;
   }
@@ -255,11 +238,17 @@ public class GenMRProcContext implements
     this.opTaskMap = opTaskMap;
   }
 
-  /**
-   * @return operators already visited
-   */
-  public List<Operator<? extends OperatorDesc>> getSeenOps() {
-    return seenOps;
+  public boolean isSeenOp(Task task, Operator operator) {
+    List<Operator<?extends OperatorDesc>> seenOps = taskToSeenOps.get(task);
+    return seenOps != null && seenOps.contains(operator);
+  }
+
+  public void addSeenOp(Task task, Operator operator) {
+    List<Operator<?extends OperatorDesc>> seenOps = taskToSeenOps.get(task);
+    if (seenOps == null) {
+      taskToSeenOps.put(task, seenOps = new ArrayList<Operator<? extends OperatorDesc>>());
+    }
+    seenOps.add(operator);
   }
 
   /**
@@ -270,14 +259,6 @@ public class GenMRProcContext implements
   }
 
   /**
-   * @param seenOps
-   *          operators already visited
-   */
-  public void setSeenOps(List<Operator<? extends OperatorDesc>> seenOps) {
-    this.seenOps = seenOps;
-  }
-
-  /**
    * @param seenFileSinkOps
    *          file sink operators already visited
    */
@@ -286,21 +267,6 @@ public class GenMRProcContext implements
   }
 
   /**
-   * @return top operators for tasks
-   */
-  public List<Operator<? extends OperatorDesc>> getRootOps() {
-    return rootOps;
-  }
-
-  /**
-   * @param rootOps
-   *          top operators for tasks
-   */
-  public void setRootOps(List<Operator<? extends OperatorDesc>> rootOps) {
-    this.rootOps = rootOps;
-  }
-
-  /**
    * @return current parse context
    */
   public ParseContext getParseCtx() {
@@ -345,6 +311,15 @@ public class GenMRProcContext implements
     this.rootTasks = rootTasks;
   }
 
+  public boolean addRootIfPossible(Task<? extends Serializable> task) {
+    if (task.getParentTasks() == null || task.getParentTasks().isEmpty()) {
+      if (!rootTasks.contains(task)) {
+        return rootTasks.add(task);
+      }
+    }
+    return false;
+  }
+
   /**
    * @return operator to task mappings
    */

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink1.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink1.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink1.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink1.java Thu Jul 25 00:14:58 2013
@@ -19,7 +19,6 @@
 package org.apache.hadoop.hive.ql.optimizer;
 
 import java.io.Serializable;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Stack;
 
@@ -64,19 +63,20 @@ public class GenMRRedSink1 implements No
     GenMapRedCtx mapredCtx = mapCurrCtx.get(stack.get(stack.size() - 2));
     Task<? extends Serializable> currTask = mapredCtx.getCurrTask();
     MapredWork currPlan = (MapredWork) currTask.getWork();
-    Operator<? extends OperatorDesc> currTopOp = mapredCtx.getCurrTopOp();
     String currAliasId = mapredCtx.getCurrAliasId();
+
+    if (op.getNumChild() != 1) {
+      throw new IllegalStateException("Expecting operator " + op + " to have one child. " +
+          "But found multiple children : " + op.getChildOperators());
+    }
     Operator<? extends OperatorDesc> reducer = op.getChildOperators().get(0);
-    HashMap<Operator<? extends OperatorDesc>, Task<? extends Serializable>> opTaskMap = ctx
-        .getOpTaskMap();
-    Task<? extends Serializable> opMapTask = opTaskMap.get(reducer);
+    Task<? extends Serializable> oldTask = ctx.getOpTaskMap().get(reducer);
 
-    ctx.setCurrTopOp(currTopOp);
     ctx.setCurrAliasId(currAliasId);
     ctx.setCurrTask(currTask);
 
     // If the plan for this reducer does not exist, initialize the plan
-    if (opMapTask == null) {
+    if (oldTask == null) {
       if (currPlan.getReducer() == null) {
         GenMapRedUtils.initPlan(op, ctx);
       } else {
@@ -85,14 +85,18 @@ public class GenMRRedSink1 implements No
     } else {
       // This will happen in case of joins. The current plan can be thrown away
       // after being merged with the original plan
-      GenMapRedUtils.joinPlan(op, null, opMapTask, ctx, -1, false);
-      currTask = opMapTask;
+      GenMapRedUtils.joinPlan(currTask, oldTask, ctx);
+      currTask = oldTask;
       ctx.setCurrTask(currTask);
     }
 
-    mapCurrCtx.put(op, new GenMapRedCtx(ctx.getCurrTask(), ctx.getCurrTopOp(),
-        ctx.getCurrAliasId()));
-    return null;
-  }
+    mapCurrCtx.put(op, new GenMapRedCtx(ctx.getCurrTask(), ctx.getCurrAliasId()));
 
+    if (GenMapRedUtils.hasBranchFinished(nodeOutputs)) {
+      ctx.addRootIfPossible(currTask);
+      return false;
+    }
+
+    return true;
+  }
 }

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink2.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink2.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink2.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink2.java Thu Jul 25 00:14:58 2013
@@ -57,28 +57,32 @@ public class GenMRRedSink2 implements No
         .getMapCurrCtx();
     GenMapRedCtx mapredCtx = mapCurrCtx.get(op.getParentOperators().get(0));
     Task<? extends Serializable> currTask = mapredCtx.getCurrTask();
-    Operator<? extends OperatorDesc> currTopOp = mapredCtx.getCurrTopOp();
     String currAliasId = mapredCtx.getCurrAliasId();
     Operator<? extends OperatorDesc> reducer = op.getChildOperators().get(0);
     Map<Operator<? extends OperatorDesc>, Task<? extends Serializable>> opTaskMap = ctx
         .getOpTaskMap();
-    Task<? extends Serializable> opMapTask = opTaskMap.get(reducer);
+    Task<? extends Serializable> oldTask = opTaskMap.get(reducer);
 
-    ctx.setCurrTopOp(currTopOp);
     ctx.setCurrAliasId(currAliasId);
     ctx.setCurrTask(currTask);
 
-    if (opMapTask == null) {
+    if (oldTask == null) {
       GenMapRedUtils.splitPlan(op, ctx);
     } else {
-      GenMapRedUtils.joinPlan(op, currTask, opMapTask, ctx, -1, true);
-      currTask = opMapTask;
+      GenMapRedUtils.splitPlan(op, currTask, oldTask, ctx);
+      currTask = oldTask;
       ctx.setCurrTask(currTask);
     }
 
-    mapCurrCtx.put(op, new GenMapRedCtx(ctx.getCurrTask(), ctx.getCurrTopOp(),
+    mapCurrCtx.put(op, new GenMapRedCtx(ctx.getCurrTask(),
         ctx.getCurrAliasId()));
-    return null;
+
+    if (GenMapRedUtils.hasBranchFinished(nodeOutputs)) {
+      ctx.addRootIfPossible(currTask);
+      return false;
+    }
+
+    return true;
   }
 
 }

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink3.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink3.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink3.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRRedSink3.java Thu Jul 25 00:14:58 2013
@@ -101,11 +101,11 @@ public class GenMRRedSink3 implements No
       ctx.setCurrTask(reducerTask);
     }
 
-    mapCurrCtx.put(op, new GenMapRedCtx(ctx.getCurrTask(), ctx.getCurrTopOp(),
+    mapCurrCtx.put(op, new GenMapRedCtx(ctx.getCurrTask(),
         ctx.getCurrAliasId()));
 
     // the union operator has been processed
     ctx.setCurrUnionOp(null);
-    return null;
+    return true;
   }
 }

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRTableScan1.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRTableScan1.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRTableScan1.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRTableScan1.java Thu Jul 25 00:14:58 2013
@@ -84,7 +84,7 @@ public class GenMRTableScan1 implements 
       if (currOp == op) {
         String currAliasId = alias;
         ctx.setCurrAliasId(currAliasId);
-        mapCurrCtx.put(op, new GenMapRedCtx(currTask, currTopOp, currAliasId));
+        mapCurrCtx.put(op, new GenMapRedCtx(currTask, currAliasId));
 
         QBParseInfo parseInfo = parseCtx.getQB().getParseInfo();
         if (parseInfo.isAnalyzeCommand()) {
@@ -139,12 +139,12 @@ public class GenMRTableScan1 implements 
             Table source = parseCtx.getQB().getMetaData().getTableForAlias(alias);
             PrunedPartitionList partList = new PrunedPartitionList(source, confirmedPartns,
                 new HashSet<Partition>(), null);
-            GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, currWork, false, ctx, partList);
+            GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, currTask, false, ctx, partList);
           } else { // non-partitioned table
-            GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, currWork, false, ctx);
+            GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, currTask, false, ctx);
           }
         }
-        return null;
+        return true;
       }
     }
     assert false;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java Thu Jul 25 00:14:58 2013
@@ -82,14 +82,13 @@ public class GenMRUnion1 implements Node
     UnionParseContext uPrsCtx = uCtx.getUnionParseContext(union);
     ctx.getMapCurrCtx().put(
         (Operator<? extends OperatorDesc>) union,
-        new GenMapRedCtx(ctx.getCurrTask(), ctx.getCurrTopOp(),
+        new GenMapRedCtx(ctx.getCurrTask(),
             ctx.getCurrAliasId()));
 
     // if the union is the first time seen, set current task to GenMRUnionCtx
     uCtxTask = ctx.getUnionTask(union);
     if (uCtxTask == null) {
-      uCtxTask = new GenMRUnionCtx();
-      uCtxTask.setUTask(ctx.getCurrTask());
+      uCtxTask = new GenMRUnionCtx(ctx.getCurrTask());
       ctx.setUnionTask(union, uCtxTask);
     }
 
@@ -101,7 +100,7 @@ public class GenMRUnion1 implements Node
       }
     }
 
-    return null;
+    return true;
   }
 
   /**
@@ -192,14 +191,11 @@ public class GenMRUnion1 implements Node
     // The current plan can be thrown away after being merged with the union
     // plan
     Task<? extends Serializable> uTask = uCtxTask.getUTask();
-    MapredWork plan = (MapredWork) uTask.getWork();
     ctx.setCurrTask(uTask);
-    List<Operator<? extends OperatorDesc>> seenOps = ctx.getSeenOps();
     Operator<? extends OperatorDesc> topOp = ctx.getCurrTopOp();
-    if (!seenOps.contains(topOp) && topOp != null) {
-      seenOps.add(topOp);
+    if (topOp != null && !ctx.isSeenOp(uTask, topOp)) {
       GenMapRedUtils.setTaskPlan(ctx.getCurrAliasId(), ctx
-          .getCurrTopOp(), plan, false, ctx);
+          .getCurrTopOp(), uTask, false, ctx);
     }
   }
 
@@ -226,6 +222,14 @@ public class GenMRUnion1 implements Node
     // future
     Map<Operator<? extends OperatorDesc>, GenMapRedCtx> mapCurrCtx = ctx.getMapCurrCtx();
 
+    if (union.getConf().isAllInputsInSameReducer()) {
+      // All inputs of this UnionOperator are in the same Reducer.
+      // We do not need to break the operator tree.
+      mapCurrCtx.put((Operator<? extends OperatorDesc>) nd,
+        new GenMapRedCtx(ctx.getCurrTask(),ctx.getCurrAliasId()));
+      return null;
+    }
+
     UnionParseContext uPrsCtx = uCtx.getUnionParseContext(union);
 
     ctx.setCurrUnionOp(union);
@@ -246,10 +250,9 @@ public class GenMRUnion1 implements Node
     // union is encountered for the first time
     GenMRUnionCtx uCtxTask = ctx.getUnionTask(union);
     if (uCtxTask == null) {
-      uCtxTask = new GenMRUnionCtx();
       uPlan = GenMapRedUtils.getMapRedWork(parseCtx);
       uTask = TaskFactory.get(uPlan, parseCtx.getConf());
-      uCtxTask.setUTask(uTask);
+      uCtxTask = new GenMRUnionCtx(uTask);
       ctx.setUnionTask(union, uCtxTask);
     }
     else {
@@ -284,9 +287,9 @@ public class GenMRUnion1 implements Node
     ctx.setCurrTask(uTask);
 
     mapCurrCtx.put((Operator<? extends OperatorDesc>) nd,
-        new GenMapRedCtx(ctx.getCurrTask(), null, null));
+        new GenMapRedCtx(ctx.getCurrTask(), null));
 
-    return null;
+    return true;
   }
 
   private boolean shouldBeRootTask(

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java Thu Jul 25 00:14:58 2013
@@ -33,7 +33,7 @@ import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.exec.ConditionalTask;
-import org.apache.hadoop.hive.ql.exec.ExecDriver;
+import org.apache.hadoop.hive.ql.exec.DemuxOperator;
 import org.apache.hadoop.hive.ql.exec.JoinOperator;
 import org.apache.hadoop.hive.ql.exec.Operator;
 import org.apache.hadoop.hive.ql.exec.OperatorFactory;
@@ -43,6 +43,8 @@ import org.apache.hadoop.hive.ql.exec.Ta
 import org.apache.hadoop.hive.ql.exec.TaskFactory;
 import org.apache.hadoop.hive.ql.exec.UnionOperator;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecDriver;
+import org.apache.hadoop.hive.ql.exec.mr.MapRedTask;
 import org.apache.hadoop.hive.ql.hooks.ReadEntity;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.metadata.Partition;
@@ -59,7 +61,6 @@ import org.apache.hadoop.hive.ql.plan.Ex
 import org.apache.hadoop.hive.ql.plan.FetchWork;
 import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
 import org.apache.hadoop.hive.ql.plan.FilterDesc.sampleDesc;
-import org.apache.hadoop.hive.ql.plan.MapJoinDesc;
 import org.apache.hadoop.hive.ql.plan.MapredLocalWork;
 import org.apache.hadoop.hive.ql.plan.MapredWork;
 import org.apache.hadoop.hive.ql.plan.OperatorDesc;
@@ -80,6 +81,10 @@ public final class GenMapRedUtils {
     LOG = LogFactory.getLog("org.apache.hadoop.hive.ql.optimizer.GenMapRedUtils");
   }
 
+  private static boolean needsTagging(Operator<? extends OperatorDesc> reducer) {
+    return (reducer.getClass() == JoinOperator.class ||
+        reducer.getClass() == DemuxOperator.class);
+  }
   /**
    * Initialize the current plan by adding it to root tasks.
    *
@@ -106,24 +111,15 @@ public final class GenMapRedUtils {
 
     plan.setNumReduceTasks(desc.getNumReducers());
 
-    List<Task<? extends Serializable>> rootTasks = opProcCtx.getRootTasks();
-
-    if (!rootTasks.contains(currTask)
-        && (currTask.getParentTasks() == null
-            || currTask.getParentTasks().isEmpty())) {
-      rootTasks.add(currTask);
-    }
-    if (reducer.getClass() == JoinOperator.class) {
+    if (needsTagging(reducer)) {
       plan.setNeedsTagging(true);
     }
 
     assert currTopOp != null;
-    List<Operator<? extends OperatorDesc>> seenOps = opProcCtx.getSeenOps();
     String currAliasId = opProcCtx.getCurrAliasId();
 
-    if (!seenOps.contains(currTopOp)) {
-      seenOps.add(currTopOp);
-      setTaskPlan(currAliasId, currTopOp, plan, false, opProcCtx);
+    if (!opProcCtx.isSeenOp(currTask, currTopOp)) {
+      setTaskPlan(currAliasId, currTopOp, currTask, false, opProcCtx);
     }
 
     currTopOp = null;
@@ -158,7 +154,7 @@ public final class GenMapRedUtils {
 
     plan.setNumReduceTasks(desc.getNumReducers());
 
-    if (reducer.getClass() == JoinOperator.class) {
+    if (needsTagging(reducer)) {
       plan.setNeedsTagging(true);
     }
 
@@ -166,16 +162,14 @@ public final class GenMapRedUtils {
   }
 
   private static void setUnionPlan(GenMRProcContext opProcCtx,
-      boolean local, MapredWork plan, GenMRUnionCtx uCtx,
+      boolean local, Task<? extends Serializable> currTask, GenMRUnionCtx uCtx,
       boolean mergeTask) throws SemanticException {
     Operator<? extends OperatorDesc> currTopOp = opProcCtx.getCurrTopOp();
 
     if (currTopOp != null) {
-      List<Operator<? extends OperatorDesc>> seenOps = opProcCtx.getSeenOps();
       String currAliasId = opProcCtx.getCurrAliasId();
-      if (!seenOps.contains(currTopOp) || mergeTask) {
-        seenOps.add(currTopOp);
-        setTaskPlan(currAliasId, currTopOp, plan, local, opProcCtx);
+      if (mergeTask || !opProcCtx.isSeenOp(currTask, currTopOp)) {
+        setTaskPlan(currAliasId, currTopOp, currTask, local, opProcCtx);
       }
       currTopOp = null;
       opProcCtx.setCurrTopOp(currTopOp);
@@ -191,6 +185,7 @@ public final class GenMapRedUtils {
         List<Operator<? extends OperatorDesc>> topOperators =
             uCtx.getListTopOperators();
 
+        MapredWork plan = (MapredWork) currTask.getWork();
         for (int pos = 0; pos < size; pos++) {
           String taskTmpDir = taskTmpDirLst.get(pos);
           TableDesc tt_desc = tt_descLst.get(pos);
@@ -214,14 +209,12 @@ public final class GenMapRedUtils {
   public static void initUnionPlan(GenMRProcContext opProcCtx, UnionOperator currUnionOp,
       Task<? extends Serializable> currTask, boolean local)
       throws SemanticException {
-    MapredWork plan = (MapredWork) currTask.getWork();
-
     // In case of lateral views followed by a join, the same tree
     // can be traversed more than one
     if (currUnionOp != null) {
       GenMRUnionCtx uCtx = opProcCtx.getUnionTask(currUnionOp);
       assert uCtx != null;
-      setUnionPlan(opProcCtx, local, plan, uCtx, false);
+      setUnionPlan(opProcCtx, local, currTask, uCtx, false);
     }
   }
 
@@ -233,12 +226,11 @@ public final class GenMapRedUtils {
       Task<? extends Serializable> currentUnionTask,
       Task<? extends Serializable> existingTask, boolean local)
       throws SemanticException {
-    MapredWork plan = (MapredWork) existingTask.getWork();
     assert currUnionOp != null;
     GenMRUnionCtx uCtx = opProcCtx.getUnionTask(currUnionOp);
     assert uCtx != null;
 
-    setUnionPlan(opProcCtx, local, plan, uCtx, true);
+    setUnionPlan(opProcCtx, local, existingTask, uCtx, true);
 
     List<Task<? extends Serializable>> parTasks = null;
     if (opProcCtx.getRootTasks().contains(currentUnionTask)) {
@@ -273,104 +265,105 @@ public final class GenMapRedUtils {
   }
 
   /**
-   * Merge the current task with the task for the current reducer.
+   * Merge the current task into the old task for the reducer
    *
-   * @param op
-   *          operator being processed
+   * @param currTask
+   *          the current task for the current reducer
    * @param oldTask
    *          the old task for the current reducer
-   * @param task
-   *          the current task for the current reducer
    * @param opProcCtx
    *          processing context
-   * @param pos
-   *          position of the parent in the stack
    */
-  public static void joinPlan(Operator<? extends OperatorDesc> op,
-      Task<? extends Serializable> oldTask, Task<? extends Serializable> task,
-      GenMRProcContext opProcCtx, int pos, boolean split)
+  public static void joinPlan(Task<? extends Serializable> currTask,
+      Task<? extends Serializable> oldTask, GenMRProcContext opProcCtx)
       throws SemanticException {
-    Task<? extends Serializable> currTask = task;
-    MapredWork plan = (MapredWork) currTask.getWork();
+    assert currTask != null && oldTask != null;
+
     Operator<? extends OperatorDesc> currTopOp = opProcCtx.getCurrTopOp();
     List<Task<? extends Serializable>> parTasks = null;
-
     // terminate the old task and make current task dependent on it
-    if (split) {
-      assert oldTask != null;
-      splitTasks(op, oldTask, currTask, opProcCtx, true, false, 0);
-    } else {
-      if ((oldTask != null) && (oldTask.getParentTasks() != null)
-          && !oldTask.getParentTasks().isEmpty()) {
-        parTasks = new ArrayList<Task<? extends Serializable>>();
-        parTasks.addAll(oldTask.getParentTasks());
-
-        Object[] parTaskArr = parTasks.toArray();
-        for (Object element : parTaskArr) {
-          ((Task<? extends Serializable>) element).removeDependentTask(oldTask);
-        }
+    if (currTask.getParentTasks() != null
+        && !currTask.getParentTasks().isEmpty()) {
+      parTasks = new ArrayList<Task<? extends Serializable>>();
+      parTasks.addAll(currTask.getParentTasks());
+
+      Object[] parTaskArr = parTasks.toArray();
+      for (Object element : parTaskArr) {
+        ((Task<? extends Serializable>) element).removeDependentTask(currTask);
       }
     }
 
     if (currTopOp != null) {
-      List<Operator<? extends OperatorDesc>> seenOps = opProcCtx.getSeenOps();
-      String currAliasId = opProcCtx.getCurrAliasId();
-
-      if (!seenOps.contains(currTopOp)) {
-        seenOps.add(currTopOp);
-        boolean local = false;
-        if (pos != -1) {
-          local = (pos == ((MapJoinDesc) op.getConf()).getPosBigTable()) ? false
-              : true;
-        }
-        setTaskPlan(currAliasId, currTopOp, plan, local, opProcCtx);
-      }
-      currTopOp = null;
-      opProcCtx.setCurrTopOp(currTopOp);
+      mergeInput(currTopOp, opProcCtx, oldTask, false);
     }
 
-    if ((oldTask != null) && (parTasks != null)) {
+    if (parTasks != null) {
       for (Task<? extends Serializable> parTask : parTasks) {
-        parTask.addDependentTask(currTask);
-        if (opProcCtx.getRootTasks().contains(currTask)) {
-          opProcCtx.getRootTasks().remove(currTask);
-        }
+        parTask.addDependentTask(oldTask);
       }
     }
 
-    opProcCtx.setCurrTask(currTask);
+    if (oldTask instanceof MapRedTask && currTask instanceof MapRedTask) {
+      ((MapRedTask)currTask).getWork().mergingInto(((MapRedTask) oldTask).getWork());
+    }
+
+    opProcCtx.setCurrTopOp(null);
+    opProcCtx.setCurrTask(oldTask);
   }
 
   /**
-   * Split the current plan by creating a temporary destination.
+   * If currTopOp is not set for input of the task, add input for to the task
+   */
+  static boolean mergeInput(Operator<? extends OperatorDesc> currTopOp,
+      GenMRProcContext opProcCtx, Task<? extends Serializable> task, boolean local)
+      throws SemanticException {
+    if (!opProcCtx.isSeenOp(task, currTopOp)) {
+      String currAliasId = opProcCtx.getCurrAliasId();
+      setTaskPlan(currAliasId, currTopOp, task, local, opProcCtx);
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * Met cRS in pRS(parentTask)-cRS-OP(childTask) case
+   * Split and link two tasks by temporary file : pRS-FS / TS-cRS-OP
+   */
+  static void splitPlan(ReduceSinkOperator cRS,
+      Task<? extends Serializable> parentTask, Task<? extends Serializable> childTask,
+      GenMRProcContext opProcCtx) throws SemanticException {
+    assert parentTask != null && childTask != null;
+    splitTasks(cRS, parentTask, childTask, opProcCtx);
+  }
+
+  /**
+   * Met cRS in pOP(parentTask with RS)-cRS-cOP(noTask) case
+   * Create new child task for cRS-cOP and link two tasks by temporary file : pOP-FS / TS-cRS-cOP
    *
-   * @param op
+   * @param cRS
    *          the reduce sink operator encountered
    * @param opProcCtx
    *          processing context
    */
-  public static void splitPlan(ReduceSinkOperator op, GenMRProcContext opProcCtx)
+  static void splitPlan(ReduceSinkOperator cRS, GenMRProcContext opProcCtx)
       throws SemanticException {
     // Generate a new task
     ParseContext parseCtx = opProcCtx.getParseCtx();
-    MapredWork cplan = getMapRedWork(parseCtx);
-    Task<? extends Serializable> redTask = TaskFactory.get(cplan, parseCtx
+    Task<? extends Serializable> parentTask = opProcCtx.getCurrTask();
+
+    MapredWork childPlan = getMapRedWork(parseCtx);
+    Task<? extends Serializable> childTask = TaskFactory.get(childPlan, parseCtx
         .getConf());
-    Operator<? extends OperatorDesc> reducer = op.getChildOperators().get(0);
+    Operator<? extends OperatorDesc> reducer = cRS.getChildOperators().get(0);
 
     // Add the reducer
-    cplan.setReducer(reducer);
-    ReduceSinkDesc desc = op.getConf();
-
-    cplan.setNumReduceTasks(new Integer(desc.getNumReducers()));
+    childPlan.setReducer(reducer);
+    ReduceSinkDesc desc = cRS.getConf();
+    childPlan.setNumReduceTasks(new Integer(desc.getNumReducers()));
 
-    HashMap<Operator<? extends OperatorDesc>, Task<? extends Serializable>> opTaskMap =
-        opProcCtx.getOpTaskMap();
-    opTaskMap.put(reducer, redTask);
-    Task<? extends Serializable> currTask = opProcCtx.getCurrTask();
+    opProcCtx.getOpTaskMap().put(reducer, childTask);
 
-    splitTasks(op, currTask, redTask, opProcCtx, true, false, 0);
-    opProcCtx.getRootOps().add(op);
+    splitTasks(cRS, parentTask, childTask, opProcCtx);
   }
 
   /**
@@ -388,9 +381,9 @@ public final class GenMapRedUtils {
    *          processing context
    */
   public static void setTaskPlan(String alias_id,
-      Operator<? extends OperatorDesc> topOp, MapredWork plan, boolean local,
+      Operator<? extends OperatorDesc> topOp, Task<?> task, boolean local,
       GenMRProcContext opProcCtx) throws SemanticException {
-    setTaskPlan(alias_id, topOp, plan, local, opProcCtx, null);
+    setTaskPlan(alias_id, topOp, task, local, opProcCtx, null);
   }
 
   private static ReadEntity getParentViewInfo(String alias_id,
@@ -432,8 +425,9 @@ public final class GenMapRedUtils {
    *          pruned partition list. If it is null it will be computed on-the-fly.
    */
   public static void setTaskPlan(String alias_id,
-      Operator<? extends OperatorDesc> topOp, MapredWork plan, boolean local,
+      Operator<? extends OperatorDesc> topOp, Task<?> task, boolean local,
       GenMRProcContext opProcCtx, PrunedPartitionList pList) throws SemanticException {
+    MapredWork plan = (MapredWork) task.getWork();
     ParseContext parseCtx = opProcCtx.getParseCtx();
     Set<ReadEntity> inputs = opProcCtx.getInputs();
 
@@ -681,6 +675,7 @@ public final class GenMapRedUtils {
       }
       plan.setMapLocalWork(localPlan);
     }
+    opProcCtx.addSeenOp(task, topOp);
   }
 
   /**
@@ -851,20 +846,20 @@ public final class GenMapRedUtils {
 
   @SuppressWarnings("nls")
   /**
-   * Merge the tasks - by creating a temporary file between them.
+   * Split two tasks by creating a temporary file between them.
+   *
    * @param op reduce sink operator being processed
-   * @param oldTask the parent task
-   * @param task the child task
+   * @param parentTask the parent task
+   * @param childTask the child task
    * @param opProcCtx context
-   * @param setReducer does the reducer needs to be set
-   * @param pos position of the parent
    **/
-  public static void splitTasks(Operator<? extends OperatorDesc> op,
-      Task<? extends Serializable> parentTask,
-      Task<? extends Serializable> childTask, GenMRProcContext opProcCtx,
-      boolean setReducer, boolean local, int posn) throws SemanticException {
-    childTask.getWork();
-    Operator<? extends OperatorDesc> currTopOp = opProcCtx.getCurrTopOp();
+  private static void splitTasks(ReduceSinkOperator op,
+      Task<? extends Serializable> parentTask, Task<? extends Serializable> childTask,
+      GenMRProcContext opProcCtx) throws SemanticException {
+    if (op.getNumParent() != 1) {
+      throw new IllegalStateException("Expecting operator " + op + " to have one parent. " +
+          "But found multiple parents : " + op.getParentOperators());
+    }
 
     ParseContext parseCtx = opProcCtx.getParseCtx();
     parentTask.addDependentTask(childTask);
@@ -880,7 +875,7 @@ public final class GenMapRedUtils {
     Context baseCtx = parseCtx.getContext();
     String taskTmpDir = baseCtx.getMRTmpFileURI();
 
-    Operator<? extends OperatorDesc> parent = op.getParentOperators().get(posn);
+    Operator<? extends OperatorDesc> parent = op.getParentOperators().get(0);
     TableDesc tt_desc = PlanUtils.getIntermediateFileTableDesc(PlanUtils
         .getFieldSchemasFromRowSchema(parent.getSchema(), "temporarycol"));
 
@@ -922,40 +917,46 @@ public final class GenMapRedUtils {
     childOpList = new ArrayList<Operator<? extends OperatorDesc>>();
     childOpList.add(op);
     ts_op.setChildOperators(childOpList);
-    op.getParentOperators().set(posn, ts_op);
+    op.getParentOperators().set(0, ts_op);
 
     Map<Operator<? extends OperatorDesc>, GenMapRedCtx> mapCurrCtx =
         opProcCtx.getMapCurrCtx();
-    mapCurrCtx.put(ts_op, new GenMapRedCtx(childTask, null, null));
+    mapCurrCtx.put(ts_op, new GenMapRedCtx(childTask, null));
 
     String streamDesc = taskTmpDir;
     MapredWork cplan = (MapredWork) childTask.getWork();
 
-    if (setReducer) {
-      Operator<? extends OperatorDesc> reducer = op.getChildOperators().get(0);
+    Operator<? extends OperatorDesc> reducer = op.getChildOperators().get(0);
 
-      if (reducer.getClass() == JoinOperator.class) {
-        String origStreamDesc;
-        streamDesc = "$INTNAME";
-        origStreamDesc = streamDesc;
-        int pos = 0;
-        while (cplan.getAliasToWork().get(streamDesc) != null) {
-          streamDesc = origStreamDesc.concat(String.valueOf(++pos));
-        }
+    if (needsTagging(reducer)) {
+      String origStreamDesc;
+      streamDesc = "$INTNAME";
+      origStreamDesc = streamDesc;
+      int pos = 0;
+      while (cplan.getAliasToWork().get(streamDesc) != null) {
+        streamDesc = origStreamDesc.concat(String.valueOf(++pos));
       }
 
       // TODO: Allocate work to remove the temporary files and make that
       // dependent on the redTask
-      if (reducer.getClass() == JoinOperator.class) {
-        cplan.setNeedsTagging(true);
-      }
+      cplan.setNeedsTagging(true);
     }
 
     // Add the path to alias mapping
-    setTaskPlan(taskTmpDir, streamDesc, ts_op, cplan, local, tt_desc);
+    setTaskPlan(taskTmpDir, streamDesc, ts_op, cplan, false, tt_desc);
     opProcCtx.setCurrTopOp(null);
     opProcCtx.setCurrAliasId(null);
     opProcCtx.setCurrTask(childTask);
+    opProcCtx.addRootIfPossible(parentTask);
+  }
+
+  static boolean hasBranchFinished(Object... children) {
+    for (Object child : children) {
+      if (child == null) {
+        return false;
+      }
+    }
+    return true;
   }
 
   private GenMapRedUtils() {

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinFactory.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinFactory.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinFactory.java Thu Jul 25 00:14:58 2013
@@ -144,38 +144,16 @@ public final class MapJoinFactory {
      *          position of the parent
      */
     private static void initMapJoinPlan(AbstractMapJoinOperator<? extends MapJoinDesc> op,
-        GenMRProcContext opProcCtx, int pos)
+        Task<? extends Serializable> currTask,
+        GenMRProcContext opProcCtx, boolean local)
         throws SemanticException {
-      Map<Operator<? extends OperatorDesc>, GenMapRedCtx> mapCurrCtx =
-          opProcCtx.getMapCurrCtx();
-      int parentPos = (pos == -1) ? 0 : pos;
-      GenMapRedCtx mapredCtx = mapCurrCtx.get(op.getParentOperators().get(
-          parentPos));
-      Task<? extends Serializable> currTask = mapredCtx.getCurrTask();
-      MapredWork plan = (MapredWork) currTask.getWork();
-      HashMap<Operator<? extends OperatorDesc>, Task<? extends Serializable>> opTaskMap =
-          opProcCtx.getOpTaskMap();
-      Operator<? extends OperatorDesc> currTopOp = opProcCtx.getCurrTopOp();
-
-      MapJoinDesc desc = (MapJoinDesc) op.getConf();
 
       // The map is overloaded to keep track of mapjoins also
-      opTaskMap.put(op, currTask);
-
-      List<Task<? extends Serializable>> rootTasks = opProcCtx.getRootTasks();
-      if(!rootTasks.contains(currTask)
-         && (currTask.getParentTasks() == null
-             || currTask.getParentTasks().isEmpty())) {
-        rootTasks.add(currTask);
-      }
-
-      assert currTopOp != null;
-      opProcCtx.getSeenOps().add(currTopOp);
+      opProcCtx.getOpTaskMap().put(op, currTask);
 
+      Operator<? extends OperatorDesc> currTopOp = opProcCtx.getCurrTopOp();
       String currAliasId = opProcCtx.getCurrAliasId();
-      boolean local = (pos == desc.getPosBigTable()) ? false : true;
-      GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, plan, local, opProcCtx);
-      setupBucketMapJoinInfo(plan, op);
+      GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, currTask, local, opProcCtx);
     }
 
     /**
@@ -191,29 +169,12 @@ public final class MapJoinFactory {
      * @param pos
      *          position of the parent in the stack
      */
-    public static void joinMapJoinPlan(AbstractMapJoinOperator<? extends OperatorDesc> op,
+    private static void joinMapJoinPlan(AbstractMapJoinOperator<? extends MapJoinDesc> op,
         Task<? extends Serializable> oldTask,
-        GenMRProcContext opProcCtx, int pos)
+        GenMRProcContext opProcCtx, boolean local)
         throws SemanticException {
-      MapredWork plan = (MapredWork) oldTask.getWork();
       Operator<? extends OperatorDesc> currTopOp = opProcCtx.getCurrTopOp();
-
-      List<Operator<? extends OperatorDesc>> seenOps = opProcCtx.getSeenOps();
-      String currAliasId = opProcCtx.getCurrAliasId();
-
-      if (!seenOps.contains(currTopOp)) {
-        seenOps.add(currTopOp);
-        boolean local = false;
-        if (pos != -1) {
-          local = (pos == ((MapJoinDesc) op.getConf()).getPosBigTable()) ? false
-              : true;
-        }
-        GenMapRedUtils.setTaskPlan(currAliasId, currTopOp, plan, local, opProcCtx);
-        setupBucketMapJoinInfo(plan, op);
-      }
-      currTopOp = null;
-      opProcCtx.setCurrTopOp(currTopOp);
-      opProcCtx.setCurrTask(oldTask);
+      GenMapRedUtils.mergeInput(currTopOp, opProcCtx, oldTask, local);
     }
 
     /*
@@ -236,17 +197,14 @@ public final class MapJoinFactory {
 
       Map<Operator<? extends OperatorDesc>, GenMapRedCtx> mapCurrCtx = ctx
           .getMapCurrCtx();
-      GenMapRedCtx mapredCtx = mapCurrCtx.get(mapJoin.getParentOperators().get(
-          pos));
+      GenMapRedCtx mapredCtx = mapCurrCtx.get(mapJoin.getParentOperators().get(pos));
       Task<? extends Serializable> currTask = mapredCtx.getCurrTask();
       MapredWork currPlan = (MapredWork) currTask.getWork();
-      Operator<? extends OperatorDesc> currTopOp = mapredCtx.getCurrTopOp();
       String currAliasId = mapredCtx.getCurrAliasId();
       HashMap<Operator<? extends OperatorDesc>, Task<? extends Serializable>> opTaskMap =
           ctx.getOpTaskMap();
-      Task<? extends Serializable> opMapTask = opTaskMap.get(mapJoin);
+      Task<? extends Serializable> oldTask = opTaskMap.get(mapJoin);
 
-      ctx.setCurrTopOp(currTopOp);
       ctx.setCurrAliasId(currAliasId);
       ctx.setCurrTask(currTask);
 
@@ -254,20 +212,23 @@ public final class MapJoinFactory {
       // If we are seeing this mapjoin for the second or later time then atleast one of the
       // branches for this mapjoin have been encounered. Join the plan with the plan created
       // the first time.
-      if (opMapTask == null) {
+      boolean local = pos != mapJoin.getConf().getPosBigTable();
+      if (oldTask == null) {
         assert currPlan.getReducer() == null;
-        initMapJoinPlan(mapJoin, ctx, pos);
+        initMapJoinPlan(mapJoin, currTask, ctx, local);
       } else {
         // The current plan can be thrown away after being merged with the
         // original plan
-        joinMapJoinPlan(mapJoin, opMapTask, ctx, pos);
-        currTask = opMapTask;
-        ctx.setCurrTask(currTask);
+        joinMapJoinPlan(mapJoin, oldTask, ctx, local);
+        ctx.setCurrTask(currTask = oldTask);
       }
+      MapredWork plan = (MapredWork) currTask.getWork();
+      setupBucketMapJoinInfo(plan, mapJoin);
+
+      mapCurrCtx.put(mapJoin, new GenMapRedCtx(ctx.getCurrTask(), ctx.getCurrAliasId()));
 
-      mapCurrCtx.put(mapJoin, new GenMapRedCtx(ctx.getCurrTask(), ctx
-          .getCurrTopOp(), ctx.getCurrAliasId()));
-      return null;
+      // local aliases need not to hand over context further
+      return !local;
     }
   }
 

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java Thu Jul 25 00:14:58 2013
@@ -263,7 +263,8 @@ public class MapJoinProcessor implements
 
     } catch (Exception e) {
       e.printStackTrace();
-      throw new SemanticException("Generate New MapJoin Opertor Exeception " + e.getMessage());
+      throw new SemanticException("Failed to generate new mapJoin operator " +
+          "by exception : " + e.getMessage());
     }
   }
 

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/Optimizer.java Thu Jul 25 00:14:58 2013
@@ -22,6 +22,8 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.optimizer.correlation.CorrelationOptimizer;
+import org.apache.hadoop.hive.ql.optimizer.correlation.ReduceSinkDeDuplication;
 import org.apache.hadoop.hive.ql.optimizer.index.RewriteGBUsingIndex;
 import org.apache.hadoop.hive.ql.optimizer.lineage.Generator;
 import org.apache.hadoop.hive.ql.optimizer.listbucketingpruner.ListBucketingPruner;
@@ -103,6 +105,11 @@ public class Optimizer {
     if (HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVELIMITOPTENABLE)) {
       transformations.add(new GlobalLimitOptimizer());
     }
+    if(HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEOPTCORRELATION) &&
+        !HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVEGROUPBYSKEW) &&
+        !HiveConf.getBoolVar(hiveConf, HiveConf.ConfVars.HIVE_OPTIMIZE_SKEWJOIN_COMPILETIME)) {
+      transformations.add(new CorrelationOptimizer());
+    }
     transformations.add(new SimpleFetchOptimizer());  // must be called last
   }
 

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/AbstractJoinTaskDispatcher.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/AbstractJoinTaskDispatcher.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/AbstractJoinTaskDispatcher.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/AbstractJoinTaskDispatcher.java Thu Jul 25 00:14:58 2013
@@ -27,9 +27,9 @@ import java.util.Stack;
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.exec.ConditionalTask;
-import org.apache.hadoop.hive.ql.exec.MapRedTask;
 import org.apache.hadoop.hive.ql.exec.Task;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.MapRedTask;
 import org.apache.hadoop.hive.ql.lib.Dispatcher;
 import org.apache.hadoop.hive.ql.lib.Node;
 import org.apache.hadoop.hive.ql.lib.TaskGraphWalker.TaskGraphWalkerContext;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java Thu Jul 25 00:14:58 2013
@@ -23,7 +23,6 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.hadoop.hive.ql.exec.ExecDriver;
 import org.apache.hadoop.hive.ql.exec.ExtractOperator;
 import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
 import org.apache.hadoop.hive.ql.exec.FilterOperator;
@@ -36,6 +35,7 @@ import org.apache.hadoop.hive.ql.exec.Li
 import org.apache.hadoop.hive.ql.exec.Operator;
 import org.apache.hadoop.hive.ql.exec.SelectOperator;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecDriver;
 import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
 import org.apache.hadoop.hive.ql.lib.Dispatcher;
 import org.apache.hadoop.hive.ql.lib.GraphWalker;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java Thu Jul 25 00:14:58 2013
@@ -34,12 +34,12 @@ import org.apache.hadoop.hive.ql.Context
 import org.apache.hadoop.hive.ql.exec.ConditionalTask;
 import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
 import org.apache.hadoop.hive.ql.exec.JoinOperator;
-import org.apache.hadoop.hive.ql.exec.MapRedTask;
 import org.apache.hadoop.hive.ql.exec.Operator;
 import org.apache.hadoop.hive.ql.exec.TableScanOperator;
 import org.apache.hadoop.hive.ql.exec.Task;
 import org.apache.hadoop.hive.ql.exec.TaskFactory;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.MapRedTask;
 import org.apache.hadoop.hive.ql.lib.Dispatcher;
 import org.apache.hadoop.hive.ql.optimizer.GenMapRedUtils;
 import org.apache.hadoop.hive.ql.optimizer.MapJoinProcessor;
@@ -416,6 +416,22 @@ public class CommonJoinTaskDispatcher ex
     copyReducerConf(mapJoinTask, childTask);
   }
 
+  public static boolean cannotConvert(String bigTableAlias,
+      Map<String, Long> aliasToSize, long aliasTotalKnownInputSize,
+      long ThresholdOfSmallTblSizeSum) {
+    boolean ret = false;
+    Long aliasKnownSize = aliasToSize.get(bigTableAlias);
+    if (aliasKnownSize != null && aliasKnownSize.longValue() > 0) {
+      long smallTblTotalKnownSize = aliasTotalKnownInputSize
+          - aliasKnownSize.longValue();
+      if (smallTblTotalKnownSize > ThresholdOfSmallTblSizeSum) {
+        //this table is not good to be a big table.
+        ret = true;
+      }
+    }
+    return ret;
+  }
+
   @Override
   public Task<? extends Serializable> processCurrentTask(MapRedTask currTask,
       ConditionalTask conditionalTask, Context context)
@@ -564,14 +580,9 @@ public class CommonJoinTaskDispatcher ex
         MapRedTask newTask = newTaskAlias.getFirst();
         bigTableAlias = newTaskAlias.getSecond();
 
-        Long aliasKnownSize = aliasToSize.get(bigTableAlias);
-        if (aliasKnownSize != null && aliasKnownSize.longValue() > 0) {
-          long smallTblTotalKnownSize = aliasTotalKnownInputSize
-              - aliasKnownSize.longValue();
-          if (smallTblTotalKnownSize > ThresholdOfSmallTblSizeSum) {
-            // this table is not good to be a big table.
-            continue;
-          }
+        if (cannotConvert(bigTableAlias, aliasToSize,
+            aliasTotalKnownInputSize, ThresholdOfSmallTblSizeSum)) {
+          continue;
         }
 
         // add into conditional task

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java Thu Jul 25 00:14:58 2013
@@ -28,12 +28,12 @@ import java.util.Stack;
 
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.exec.ConditionalTask;
-import org.apache.hadoop.hive.ql.exec.MapredLocalTask;
 import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
 import org.apache.hadoop.hive.ql.exec.Operator;
 import org.apache.hadoop.hive.ql.exec.Task;
 import org.apache.hadoop.hive.ql.exec.TaskFactory;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.MapredLocalTask;
 import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
 import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
 import org.apache.hadoop.hive.ql.lib.Dispatcher;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/SamplingOptimizer.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/SamplingOptimizer.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/SamplingOptimizer.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/SamplingOptimizer.java Thu Jul 25 00:14:58 2013
@@ -20,12 +20,12 @@
 
 package org.apache.hadoop.hive.ql.optimizer.physical;
 
-import org.apache.hadoop.hive.ql.exec.MapRedTask;
 import org.apache.hadoop.hive.ql.exec.Operator;
 import org.apache.hadoop.hive.ql.exec.OperatorUtils;
 import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
 import org.apache.hadoop.hive.ql.exec.TableScanOperator;
 import org.apache.hadoop.hive.ql.exec.Task;
+import org.apache.hadoop.hive.ql.exec.mr.MapRedTask;
 import org.apache.hadoop.hive.ql.parse.SemanticException;
 import org.apache.hadoop.hive.ql.plan.MapredWork;
 

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/SortMergeJoinTaskDispatcher.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/SortMergeJoinTaskDispatcher.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/SortMergeJoinTaskDispatcher.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/SortMergeJoinTaskDispatcher.java Thu Jul 25 00:14:58 2013
@@ -35,13 +35,13 @@ import org.apache.hadoop.hive.ql.exec.Co
 import org.apache.hadoop.hive.ql.exec.DummyStoreOperator;
 import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
 import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
-import org.apache.hadoop.hive.ql.exec.MapRedTask;
 import org.apache.hadoop.hive.ql.exec.Operator;
 import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
 import org.apache.hadoop.hive.ql.exec.SMBMapJoinOperator;
 import org.apache.hadoop.hive.ql.exec.Task;
 import org.apache.hadoop.hive.ql.exec.TaskFactory;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.MapRedTask;
 import org.apache.hadoop.hive.ql.lib.Dispatcher;
 import org.apache.hadoop.hive.ql.optimizer.MapJoinProcessor;
 import org.apache.hadoop.hive.ql.parse.OpParseContext;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/index/IndexWhereProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/index/IndexWhereProcessor.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/index/IndexWhereProcessor.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/index/IndexWhereProcessor.java Thu Jul 25 00:14:58 2013
@@ -31,10 +31,10 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.hive.metastore.api.Index;
-import org.apache.hadoop.hive.ql.exec.MapRedTask;
 import org.apache.hadoop.hive.ql.exec.TableScanOperator;
 import org.apache.hadoop.hive.ql.exec.Task;
 import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.MapRedTask;
 import org.apache.hadoop.hive.ql.hooks.ReadEntity;
 import org.apache.hadoop.hive.ql.index.HiveIndexHandler;
 import org.apache.hadoop.hive.ql.index.HiveIndexQueryContext;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/FromClauseParser.g Thu Jul 25 00:14:58 2013
@@ -112,13 +112,13 @@ joinToken
 @init { gParent.msgs.push("join type specifier"); }
 @after { gParent.msgs.pop(); }
     :
-      KW_JOIN                     -> TOK_JOIN
-    | KW_INNER  KW_JOIN            -> TOK_JOIN
-    | KW_CROSS KW_JOIN            -> TOK_CROSSJOIN
-    | KW_LEFT  KW_OUTER KW_JOIN   -> TOK_LEFTOUTERJOIN
-    | KW_RIGHT KW_OUTER KW_JOIN   -> TOK_RIGHTOUTERJOIN
-    | KW_FULL  KW_OUTER KW_JOIN   -> TOK_FULLOUTERJOIN
-    | KW_LEFT  KW_SEMI  KW_JOIN   -> TOK_LEFTSEMIJOIN
+      KW_JOIN                      -> TOK_JOIN
+    | KW_INNER KW_JOIN             -> TOK_JOIN
+    | KW_CROSS KW_JOIN             -> TOK_CROSSJOIN
+    | KW_LEFT  (KW_OUTER)? KW_JOIN -> TOK_LEFTOUTERJOIN
+    | KW_RIGHT (KW_OUTER)? KW_JOIN -> TOK_RIGHTOUTERJOIN
+    | KW_FULL  (KW_OUTER)? KW_JOIN -> TOK_FULLOUTERJOIN
+    | KW_LEFT KW_SEMI KW_JOIN      -> TOK_LEFTSEMIJOIN
     ;
 
 lateralView

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/GenMapRedWalker.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/GenMapRedWalker.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/GenMapRedWalker.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/GenMapRedWalker.java Thu Jul 25 00:14:58 2013
@@ -20,7 +20,6 @@ package org.apache.hadoop.hive.ql.parse;
 
 import java.util.List;
 
-import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
 import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
 import org.apache.hadoop.hive.ql.lib.Dispatcher;
 import org.apache.hadoop.hive.ql.lib.Node;
@@ -52,12 +51,11 @@ public class GenMapRedWalker extends Def
 
     // maintain the stack of operators encountered
     opStack.push(nd);
-    dispatch(nd, opStack);
+    Boolean result = dispatchAndReturn(nd, opStack);
 
-    // kids of reduce sink operator need not be traversed again
-    if ((children == null)
-        || ((nd instanceof ReduceSinkOperator) && (getDispatchedList()
-        .containsAll(children)))) {
+    // kids of reduce sink operator or mapjoin operators merged into root task
+    // need not be traversed again
+    if (children == null || result == Boolean.FALSE) {
       opStack.pop();
       return;
     }

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/IndexUpdater.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/IndexUpdater.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/IndexUpdater.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/IndexUpdater.java Thu Jul 25 00:14:58 2013
@@ -40,7 +40,6 @@ import org.apache.commons.logging.LogFac
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.hive.metastore.api.Index;
 import org.apache.hadoop.hive.ql.exec.FilterOperator;
-import org.apache.hadoop.hive.ql.exec.MapRedTask;
 import org.apache.hadoop.hive.ql.exec.TableScanOperator;
 import org.apache.hadoop.hive.ql.exec.Task;
 import org.apache.hadoop.hive.ql.exec.Utilities;

Modified: hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/MapReduceCompiler.java
URL: http://svn.apache.org/viewvc/hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/MapReduceCompiler.java?rev=1506785&r1=1506784&r2=1506785&view=diff
==============================================================================
--- hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/MapReduceCompiler.java (original)
+++ hive/branches/vectorization/ql/src/java/org/apache/hadoop/hive/ql/parse/MapReduceCompiler.java Thu Jul 25 00:14:58 2013
@@ -50,8 +50,8 @@ import org.apache.hadoop.hive.ql.exec.Ta
 import org.apache.hadoop.hive.ql.exec.TaskFactory;
 import org.apache.hadoop.hive.ql.exec.UnionOperator;
 import org.apache.hadoop.hive.ql.exec.Utilities;
-import org.apache.hadoop.hive.ql.exec.ExecDriver;
-import org.apache.hadoop.hive.ql.exec.MapRedTask;
+import org.apache.hadoop.hive.ql.exec.mr.ExecDriver;
+import org.apache.hadoop.hive.ql.exec.mr.MapRedTask;
 import org.apache.hadoop.hive.ql.hooks.ReadEntity;
 import org.apache.hadoop.hive.ql.hooks.WriteEntity;
 import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
@@ -217,8 +217,7 @@ public class MapReduceCompiler {
     GenMRProcContext procCtx = new GenMRProcContext(
         conf,
         new HashMap<Operator<? extends OperatorDesc>, Task<? extends Serializable>>(),
-        new ArrayList<Operator<? extends OperatorDesc>>(), tempParseContext,
-        mvTask, rootTasks,
+        tempParseContext, mvTask, rootTasks,
         new LinkedHashMap<Operator<? extends OperatorDesc>, GenMapRedCtx>(),
         inputs, outputs);
 
@@ -599,8 +598,7 @@ public class MapReduceCompiler {
     boolean hasNonLocalJob = false;
     for (ExecDriver mrtask : mrtasks) {
       try {
-        ContentSummary inputSummary = Utilities.getInputSummary
-            (ctx, (MapredWork) mrtask.getWork(), p);
+        ContentSummary inputSummary = Utilities.getInputSummary(ctx, mrtask.getWork(), p);
         int numReducers = getNumberOfReducers(mrtask.getWork(), conf);
 
         long estimatedInput;



Mime
View raw message