code: @Override
  	public List<PlanNode> getAlternativePlans(CostEstimator estimator) {
  		// check if we have a cached version
  		if (this.cachedPlans != null) {
  			return this.cachedPlans;
  		}
  		
  		// calculate alternative sub-plans for predecessor
  		List<? extends PlanNode> subPlans = getPredecessorNode().getAlternativePlans(estimator);
  		List<PlanNode> outputPlans = new ArrayList<PlanNode>();
  		
  		final int parallelism = getParallelism();
  		final int inDop = getPredecessorNode().getParallelism();
  
  		final ExecutionMode executionMode = this.input.getDataExchangeMode();
  		final boolean dopChange = parallelism != inDop;
  		final boolean breakPipeline = this.input.isBreakingPipeline();
  
  		InterestingProperties ips = this.input.getInterestingProperties();
  		for (PlanNode p : subPlans) {
  			for (RequestedGlobalProperties gp : ips.getGlobalProperties()) {
  				for (RequestedLocalProperties lp : ips.getLocalProperties()) {
  					Channel c = new Channel(p);
  					gp.parameterizeChannel(c, dopChange, executionMode, breakPipeline);
  					lp.parameterizeChannel(c);
  					c.setRequiredLocalProps(lp);
  					c.setRequiredGlobalProps(gp);
  					
  					// no need to check whether the created properties meet what we need in case
  					// of ordering or global ordering, because the only interesting properties we have
  					// are what we require
  					outputPlans.add(new SinkPlanNode(this, "DataSink ("+this.getOperator().getName()+")" ,c));
  				}
  			}
  		}
  		
  		// cost and prune the plans
  		for (PlanNode node : outputPlans) {
  			estimator.costOperator(node);
  		}
  		prunePlanAlternatives(outputPlans);
  
  		this.cachedPlans = outputPlans;
  		return outputPlans;
  	}
docs: --------------------------------------------------------------------------------------------