class ShuffleStageGenerator {

	public static Map<ExecNode<?, ?>, ShuffleStage> generate(List<ExecNode<?, ?>> sinkNodes, Map<ExecNode<?, ?>, Integer> finalParallelismNodeMap) {
		ShuffleStageGenerator generator = new ShuffleStageGenerator(finalParallelismNodeMap);
		sinkNodes.forEach(generator::buildShuffleStages);
		Map<ExecNode<?, ?>, ShuffleStage> result = generator.getNodeShuffleStageMap();
		result.values().forEach(s -> {
			List<ExecNode<?, ?>> virtualNodeList = s.getExecNodeSet().stream().filter(ShuffleStageGenerator::isVirtualNode).collect(toList());
			virtualNodeList.forEach(s::removeNode);
		});
		return generator.getNodeShuffleStageMap().entrySet().stream()
				.filter(x -> !isVirtualNode(x.getKey()))
				.collect(Collectors.toMap(Map.Entry::getKey,
						Map.Entry::getValue,
						(e1, e2) -> e1,
						LinkedHashMap::new));
	}

	private  ShuffleStageGenerator(Map<ExecNode<?, ?>, Integer> nodeToFinalParallelismMap);

}