private void fixParallelEdges(OperatorGraph og) throws SemanticException {

    // Identify edge operators
    ActualEdgePredicate actualEdgePredicate = new ActualEdgePredicate();

    ListValuedMap<Pair<Cluster, Cluster>, Pair<Operator<?>, Operator<?>>> edgeOperators =
        new ArrayListValuedHashMap<>();
    for (Cluster cluster: og.getClusters()) {
      for (Cluster parentCluster: cluster.parentClusters(actualEdgePredicate)) {
        Set<Operator<?>> parentOperators = parentCluster.getMembers();
        for (Operator<?> operator: cluster.getMembers()) {
          for (Operator<?> parentOperator: operator.getParentOperators()) {
            if (parentOperators.contains(parentOperator)) {
              edgeOperators.put(new Pair<>(parentCluster, cluster), new Pair<>(parentOperator, operator));
            }
          }
        }
      }
    }

    HashSet<Pair<Operator<?>, Operator<?>>> processedEdge = new HashSet<>();
    // process all edges and fix parallel edges if there are any
    for (Pair<Cluster, Cluster> key : edgeOperators.keySet()) {
      List<Pair<Operator<?>, Operator<?>>> values = edgeOperators.get(key);
      if (values.size() <= 1) {
        continue;
      }
      // operator order must in stabile order - or we end up with falky plans causing flaky tests...
      values.sort(new OperatorPairComparator());

      // remove one optionally unsupported edge (it will be kept as is)
      removeOneEdge(values);

      Iterator<Pair<Operator<?>, Operator<?>>> it = values.iterator();
      while (it.hasNext()) {
        Pair<Operator<?>, Operator<?>> pair = it.next();
        if (!processedEdge.contains(pair)) {
          fixParallelEdge(pair.left, pair.right);
          processedEdge.add(pair);
        }
      }
    }
  }
To achieve reproducible plans - the operators must be placed in some stable order.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/ParallelEdgeFixer.java