@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 dop = getDegreeOfParallelism();
		final int subPerInstance = getSubtasksPerInstance();
		final int inDop = getPredecessorNode().getDegreeOfParallelism();
		final int inSubPerInstance = getPredecessorNode().getSubtasksPerInstance();
		final int numInstances = dop / subPerInstance + (dop % subPerInstance == 0 ? 0 : 1);
		final int inNumInstances = inDop / inSubPerInstance + (inDop % inSubPerInstance == 0 ? 0 : 1);
		
		final boolean globalDopChange = numInstances != inNumInstances;
		final boolean localDopChange = numInstances == inNumInstances & subPerInstance != inSubPerInstance;
		
		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, globalDopChange, localDopChange);

					if (lp.isMetBy(c.getLocalPropertiesAfterShippingOnly())) {
						c.setLocalStrategy(LocalStrategy.NONE);
					} else {
						lp.parameterizeChannel(c);
					}
					
					// 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.getPactContract().getName()+")" ,c));
				}
			}
		}
		
		// cost and prune the plans
		for (PlanNode node : outputPlans) {
			estimator.costOperator(node);
		}
		prunePlanAlternatives(outputPlans);

		this.cachedPlans = outputPlans;
		return outputPlans;
	}
@ Override public List < PlanNode > getAlternativePlans ( CostEstimator estimator ) { if ( this . cachedPlans != null ) { return this . cachedPlans ; } List < ? extends PlanNode > subPlans = getPredecessorNode ( ) . getAlternativePlans ( estimator ) ; List < PlanNode > outputPlans = new ArrayList < PlanNode > ( ) ; final int dop = getDegreeOfParallelism ( ) ; final int subPerInstance = getSubtasksPerInstance ( ) ; final int inDop = getPredecessorNode ( ) . getDegreeOfParallelism ( ) ; final int inSubPerInstance = getPredecessorNode ( ) . getSubtasksPerInstance ( ) ; final int numInstances = dop / subPerInstance + ( dop % subPerInstance == 0 ? 0 : 1 ) ; final int inNumInstances = inDop / inSubPerInstance + ( inDop % inSubPerInstance == 0 ? 0 : 1 ) ; final boolean globalDopChange = numInstances != inNumInstances ; final boolean localDopChange = numInstances == inNumInstances & subPerInstance != inSubPerInstance ; 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 , globalDopChange , localDopChange ) ; if ( lp . isMetBy ( c . getLocalPropertiesAfterShippingOnly ( ) ) ) { c . setLocalStrategy ( LocalStrategy . NONE ) ; } else { lp . parameterizeChannel ( c ) ; } outputPlans . add ( new SinkPlanNode ( this , "DataSink(" + this . getPactContract ( ) . getName ( ) + ")" , c ) ) ; } } } for ( PlanNode node : outputPlans ) { estimator . costOperator ( node ) ; } prunePlanAlternatives ( outputPlans ) ; this . cachedPlans = outputPlans ; return outputPlans ; }
--------------------------------------------------------------------------------------------