private void fixParallelEdge(Operator<? extends OperatorDesc> p, Operator<?> o) throws SemanticException {
    LOG.info("Fixing parallel by adding a concentrator RS between {} -> {}", p, o);

    ReduceSinkDesc conf = (ReduceSinkDesc) p.getConf();
    ReduceSinkDesc newConf = (ReduceSinkDesc) conf.clone();

    Operator<SelectDesc> newSEL = buildSEL(p, conf);

    Operator<ReduceSinkDesc> newRS =
        OperatorFactory.getAndMakeChild(p.getCompilationOpContext(), newConf, new ArrayList<>());

    conf.setOutputName("forward_to_" + newRS);
    conf.setTag(0);

    newConf.setKeyCols(new ArrayList<>(conf.getKeyCols()));
    newRS.setSchema(new RowSchema(p.getSchema()));

    p.replaceChild(o, newSEL);

    newSEL.setParentOperators(Lists.<Operator<?>> newArrayList(p));
    newSEL.setChildOperators(Lists.<Operator<?>> newArrayList(newRS));
    newRS.setParentOperators(Lists.<Operator<?>> newArrayList(newSEL));
    newRS.setChildOperators(Lists.<Operator<?>> newArrayList(o));

    o.replaceParent(p, newRS);

  }
Fixes a parallel edge going into a mapjoin by introducing a concentrator RS.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/ParallelEdgeFixer.java