diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java index c77e081..590c179 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java @@ -22,8 +22,11 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.Map.Entry; import java.util.TreeMap; @@ -154,12 +157,12 @@ void init(JobConf jconf, ProcessorContext processorContext, MRTaskReporter mrRep mapOp = new MapOperator(); } + processorContext.waitForAllInputsReady(Collections.unmodifiableCollection(inputs.values())); + connectOps.clear(); if (mergeWorkList != null) { MapOperator mergeMapOp = null; for (MapWork mergeMapWork : mergeWorkList) { - processorContext.waitForAnyInputReady(Collections.singletonList((Input) (inputs - .get(mergeMapWork.getName())))); if (mergeMapWork.getVectorMode()) { mergeMapOp = new VectorMapOperator(); } else {