**/
private static void splitTasks(ReduceSinkOperator op,
Task<? extends Serializable> parentTask,
Task<? extends Serializable> childTask,
GenMRProcContext opProcCtx) throws SemanticException {
mapredWork plan = (mapredWork) childTask.getWork();
Operator<? extends Serializable> currTopOp = opProcCtx.getCurrTopOp();
ParseContext parseCtx = opProcCtx.getParseCtx();
parentTask.addDependentTask(childTask);
// Root Task cannot depend on any other task, therefore childTask cannot be a root Task
List<Task<? extends Serializable>> rootTasks = opProcCtx.getRootTasks();
if (rootTasks.contains(childTask))
rootTasks.remove(childTask);
// generate the temporary file
String scratchDir = opProcCtx.getScratchDir();
int randomid = opProcCtx.getRandomId();
int pathid = opProcCtx.getPathId();
String taskTmpDir = (new Path(scratchDir + File.separator + randomid + '.' + pathid)).toString();
pathid++;
opProcCtx.setPathId(pathid);
Operator<? extends Serializable> parent = op.getParentOperators().get(0);
tableDesc tt_desc =
PlanUtils.getBinaryTableDesc(PlanUtils.getFieldSchemasFromRowSchema(parent.getSchema(), "temporarycol"));
// Create a file sink operator for this file name
Operator<? extends Serializable> fs_op =
putOpInsertMap(OperatorFactory.get
(new fileSinkDesc(taskTmpDir, tt_desc,
parseCtx.getConf().getBoolVar(HiveConf.ConfVars.COMPRESSINTERMEDIATE)),
parent.getSchema()), null, parseCtx);
// replace the reduce child with this operator
List<Operator<? extends Serializable>> childOpList = parent.getChildOperators();
for (int pos = 0; pos < childOpList.size(); pos++) {
if (childOpList.get(pos) == op) {
childOpList.set(pos, fs_op);
break;
}
}
List<Operator<? extends Serializable>> parentOpList = new ArrayList<Operator<? extends Serializable>>();
parentOpList.add(parent);
fs_op.setParentOperators(parentOpList);
Operator<? extends Serializable> reducer = op.getChildOperators().get(0);
String streamDesc;
mapredWork cplan = (mapredWork) childTask.getWork();
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));
}
else
streamDesc = taskTmpDir;
// Add the path to alias mapping
if (cplan.getPathToAliases().get(taskTmpDir) == null) {
cplan.getPathToAliases().put(taskTmpDir, new ArrayList<String>());
}
cplan.getPathToAliases().get(taskTmpDir).add(streamDesc);
cplan.getPathToPartitionInfo().put(taskTmpDir, new partitionDesc(tt_desc, null));
cplan.getAliasToWork().put(streamDesc, op);
// TODO: Allocate work to remove the temporary files and make that
// dependent on the redTask
if (reducer.getClass() == JoinOperator.class)
cplan.setNeedsTagging(true);
currTopOp = null;
String currAliasId = null;
opProcCtx.setCurrTopOp(currTopOp);