[13/53] [abbrv] flink git commit: [optimizer] Rename optimizer project to "flink-optimizer" (previously flink-compiler)

[se...@apache.org]

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/BulkPartialSolutionPlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/BulkPartialSolutionPlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/BulkPartialSolutionPlanNode.java
new file mode 100644
index 0000000..df05b64
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/BulkPartialSolutionPlanNode.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.plan;
+
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE_AND_DAM;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.NOT_FOUND;
+
+import java.util.Collections;
+import java.util.HashMap;
+
+import org.apache.flink.optimizer.costs.Costs;
+import org.apache.flink.optimizer.dag.BulkPartialSolutionNode;
+import org.apache.flink.optimizer.dag.OptimizerNode;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.runtime.operators.DamBehavior;
+import org.apache.flink.runtime.operators.DriverStrategy;
+import org.apache.flink.util.Visitor;
+
+/**
+ * Plan candidate node for partial solution of a bulk iteration.
+ */
+public class BulkPartialSolutionPlanNode extends PlanNode {
+	
+	private static final Costs NO_COSTS = new Costs();
+	
+	private BulkIterationPlanNode containingIterationNode;
+	
+	private Channel initialInput;
+	
+	public Object postPassHelper;
+	
+	
+	public BulkPartialSolutionPlanNode(BulkPartialSolutionNode template, String nodeName,
+			GlobalProperties gProps, LocalProperties lProps,
+			Channel initialInput)
+	{
+		super(template, nodeName, DriverStrategy.NONE);
+		
+		this.globalProps = gProps;
+		this.localProps = lProps;
+		this.initialInput = initialInput;
+		
+		// the partial solution does not cost anything
+		this.nodeCosts = NO_COSTS;
+		this.cumulativeCosts = NO_COSTS;
+		
+		if (initialInput.getSource().branchPlan != null && initialInput.getSource().branchPlan.size() > 0) {
+			if (this.branchPlan == null) {
+				this.branchPlan = new HashMap<OptimizerNode, PlanNode>();
+			}
+			
+			this.branchPlan.putAll(initialInput.getSource().branchPlan);
+		}
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	public BulkPartialSolutionNode getPartialSolutionNode() {
+		return (BulkPartialSolutionNode) this.template;
+	}
+	
+	public BulkIterationPlanNode getContainingIterationNode() {
+		return this.containingIterationNode;
+	}
+	
+	public void setContainingIterationNode(BulkIterationPlanNode containingIterationNode) {
+		this.containingIterationNode = containingIterationNode;
+	}
+
+	// --------------------------------------------------------------------------------------------
+	
+	@Override
+	public void accept(Visitor<PlanNode> visitor) {
+		if (visitor.preVisit(this)) {
+			visitor.postVisit(this);
+		}
+	}
+
+	@Override
+	public Iterable<PlanNode> getPredecessors() {
+		return Collections.<PlanNode>emptyList();
+	}
+
+	@Override
+	public Iterable<Channel> getInputs() {
+		return Collections.<Channel>emptyList();
+	}
+
+	@Override
+	public SourceAndDamReport hasDamOnPathDownTo(PlanNode source) {
+		if (source == this) {
+			return FOUND_SOURCE;
+		}
+		SourceAndDamReport res = this.initialInput.getSource().hasDamOnPathDownTo(source);
+		if (res == FOUND_SOURCE_AND_DAM) {
+			return FOUND_SOURCE_AND_DAM;
+		}
+		else if (res == FOUND_SOURCE) {
+			return (this.initialInput.getLocalStrategy().dams() || 
+					this.initialInput.getTempMode().breaksPipeline() ||
+					getDriverStrategy().firstDam() == DamBehavior.FULL_DAM) ?
+				FOUND_SOURCE_AND_DAM : FOUND_SOURCE;
+		}
+		else {
+			return NOT_FOUND;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/Channel.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/Channel.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/Channel.java
new file mode 100644
index 0000000..875d1c3
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/Channel.java
@@ -0,0 +1,538 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.plan;
+
+import org.apache.flink.api.common.distributions.DataDistribution;
+import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.api.common.operators.util.FieldList;
+import org.apache.flink.api.common.typeutils.TypeComparatorFactory;
+import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
+import org.apache.flink.optimizer.CompilerException;
+import org.apache.flink.optimizer.dag.EstimateProvider;
+import org.apache.flink.optimizer.dag.TempMode;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.optimizer.dataproperties.RequestedGlobalProperties;
+import org.apache.flink.optimizer.dataproperties.RequestedLocalProperties;
+import org.apache.flink.optimizer.plandump.DumpableConnection;
+import org.apache.flink.optimizer.util.Utils;
+import org.apache.flink.runtime.io.network.DataExchangeMode;
+import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
+import org.apache.flink.runtime.operators.util.LocalStrategy;
+
+/**
+ * A Channel represents the result produced by an operator and the data exchange
+ * before the consumption by the target operator.
+ *
+ * The channel defines and tracks various properties and characteristics of the
+ * data set and data exchange.
+ *
+ * Data set characteristics:
+ * <ul>
+ *     <li>The "global properties" of the data, i.e., how the data is distributed across
+ *         partitions</li>
+ *     <li>The "required global properties" of the data, i.e., the global properties that, if absent,
+ *         would cause the program to return a wrong result.</li>
+ *     <li>The "local properties" of the data, i.e., how the data is organized within a partition</li>
+ *     <li>The "required local properties" of the data, i.e., the local properties that, if absent,
+ *         would cause the program to return a wrong result.</li>
+ * </ul>
+ *
+ * Data exchange parameters:
+ * <ul>
+ *     <li>The "ship strategy", i.e., whether to forward the data, shuffle it, broadcast it, ...</li>
+ *     <li>The "ship keys", which are the positions of the key fields in the exchanged records.</li>
+ *     <li>The "data exchange mode", which defines whether to pipeline or batch the exchange</li>
+ *     <li>Several more...</li>
+ * </ul>
+ */
+public class Channel implements EstimateProvider, Cloneable, DumpableConnection<PlanNode> {
+	
+	private PlanNode source;
+	
+	private PlanNode target;
+
+	private ShipStrategyType shipStrategy = ShipStrategyType.NONE;
+
+	private DataExchangeMode dataExchangeMode;
+	
+	private LocalStrategy localStrategy = LocalStrategy.NONE;
+	
+	private FieldList shipKeys;
+	
+	private FieldList localKeys;
+	
+	private boolean[] shipSortOrder;
+	
+	private boolean[] localSortOrder;
+	
+	private RequestedGlobalProperties requiredGlobalProps;
+	
+	private RequestedLocalProperties requiredLocalProps;
+	
+	private GlobalProperties globalProps;
+	
+	private LocalProperties localProps;
+	
+	private TypeSerializerFactory<?> serializer;
+	
+	private TypeComparatorFactory<?> shipStrategyComparator;
+	
+	private TypeComparatorFactory<?> localStrategyComparator;
+	
+	private DataDistribution dataDistribution;
+	
+	private Partitioner<?> partitioner;
+	
+	private TempMode tempMode;
+	
+	private double relativeTempMemory;
+	
+	private double relativeMemoryLocalStrategy;
+	
+	private int replicationFactor = 1;
+	
+	// --------------------------------------------------------------------------------------------
+	
+	public Channel(PlanNode sourceNode) {
+		this(sourceNode, null);
+	}
+	
+	public Channel(PlanNode sourceNode, TempMode tempMode) {
+		this.source = sourceNode;
+		this.tempMode = (tempMode == null ? TempMode.NONE : tempMode);
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	//                                         Accessors
+	// --------------------------------------------------------------------------------------------
+	
+	/**
+	 * Gets the source of this Channel.
+	 *
+	 * @return The source.
+	 */
+	@Override
+	public PlanNode getSource() {
+		return this.source;
+	}
+	
+	/**
+	 * Sets the target of this Channel.
+	 *
+	 * @param target The target.
+	 */
+	public void setTarget(PlanNode target) {
+		this.target = target;
+	}
+	
+	/**
+	 * Gets the target of this Channel.
+	 *
+	 * @return The target.
+	 */
+	public PlanNode getTarget() {
+		return this.target;
+	}
+
+	public void setShipStrategy(ShipStrategyType strategy, DataExchangeMode dataExchangeMode) {
+		setShipStrategy(strategy, null, null, null, dataExchangeMode);
+	}
+	
+	public void setShipStrategy(ShipStrategyType strategy, FieldList keys, DataExchangeMode dataExchangeMode) {
+		setShipStrategy(strategy, keys, null, null, dataExchangeMode);
+	}
+	
+	public void setShipStrategy(ShipStrategyType strategy, FieldList keys,
+								boolean[] sortDirection, DataExchangeMode dataExchangeMode) {
+		setShipStrategy(strategy, keys, sortDirection, null, dataExchangeMode);
+	}
+	
+	public void setShipStrategy(ShipStrategyType strategy, FieldList keys,
+								Partitioner<?> partitioner, DataExchangeMode dataExchangeMode) {
+		setShipStrategy(strategy, keys, null, partitioner, dataExchangeMode);
+	}
+	
+	public void setShipStrategy(ShipStrategyType strategy, FieldList keys,
+								boolean[] sortDirection, Partitioner<?> partitioner,
+								DataExchangeMode dataExchangeMode) {
+		this.shipStrategy = strategy;
+		this.shipKeys = keys;
+		this.shipSortOrder = sortDirection;
+		this.partitioner = partitioner;
+		this.dataExchangeMode = dataExchangeMode;
+		this.globalProps = null;		// reset the global properties
+	}
+
+	/**
+	 * Gets the data exchange mode (batch / streaming) to use for the data
+	 * exchange of this channel.
+	 *
+	 * @return The data exchange mode of this channel.
+	 */
+	public DataExchangeMode getDataExchangeMode() {
+		return dataExchangeMode;
+	}
+
+	public ShipStrategyType getShipStrategy() {
+		return this.shipStrategy;
+	}
+	
+	public FieldList getShipStrategyKeys() {
+		return this.shipKeys;
+	}
+	
+	public boolean[] getShipStrategySortOrder() {
+		return this.shipSortOrder;
+	}
+	
+	public void setLocalStrategy(LocalStrategy strategy) {
+		setLocalStrategy(strategy, null, null);
+	}
+	
+	public void setLocalStrategy(LocalStrategy strategy, FieldList keys, boolean[] sortDirection) {
+		this.localStrategy = strategy;
+		this.localKeys = keys;
+		this.localSortOrder = sortDirection;
+		this.localProps = null;		// reset the local properties
+	}
+	
+	public LocalStrategy getLocalStrategy() {
+		return this.localStrategy;
+	}
+	
+	public FieldList getLocalStrategyKeys() {
+		return this.localKeys;
+	}
+	
+	public boolean[] getLocalStrategySortOrder() {
+		return this.localSortOrder;
+	}
+	
+	public void setDataDistribution(DataDistribution dataDistribution) {
+		this.dataDistribution = dataDistribution;
+	}
+	
+	public DataDistribution getDataDistribution() {
+		return this.dataDistribution;
+	}
+	
+	public Partitioner<?> getPartitioner() {
+		return partitioner;
+	}
+	
+	public TempMode getTempMode() {
+		return this.tempMode;
+	}
+
+	/**
+	 * Sets the temp mode of the connection.
+	 * 
+	 * @param tempMode
+	 *        The temp mode of the connection.
+	 */
+	public void setTempMode(TempMode tempMode) {
+		this.tempMode = tempMode;
+	}
+	
+	/**
+	 * Gets the memory for materializing the channel's result from this Channel.
+	 *
+	 * @return The temp memory.
+	 */
+	public double getRelativeTempMemory() {
+		return this.relativeTempMemory;
+	}
+	
+	/**
+	 * Sets the memory for materializing the channel's result from this Channel.
+	 *
+	 * @param relativeTempMemory The memory for materialization.
+	 */
+	public void setRelativeTempMemory(double relativeTempMemory) {
+		this.relativeTempMemory = relativeTempMemory;
+	}
+	
+	/**
+	 * Sets the replication factor of the connection.
+	 * 
+	 * @param factor The replication factor of the connection.
+	 */
+	public void setReplicationFactor(int factor) {
+		this.replicationFactor = factor;
+	}
+	
+	/**
+	 * Returns the replication factor of the connection.
+	 * 
+	 * @return The replication factor of the connection.
+	 */
+	public int getReplicationFactor() {
+		return this.replicationFactor;
+	}
+	
+	/**
+	 * Gets the serializer from this Channel.
+	 *
+	 * @return The serializer.
+	 */
+	public TypeSerializerFactory<?> getSerializer() {
+		return serializer;
+	}
+	
+	/**
+	 * Sets the serializer for this Channel.
+	 *
+	 * @param serializer The serializer to set.
+	 */
+	public void setSerializer(TypeSerializerFactory<?> serializer) {
+		this.serializer = serializer;
+	}
+	
+	/**
+	 * Gets the ship strategy comparator from this Channel.
+	 *
+	 * @return The ship strategy comparator.
+	 */
+	public TypeComparatorFactory<?> getShipStrategyComparator() {
+		return shipStrategyComparator;
+	}
+	
+	/**
+	 * Sets the ship strategy comparator for this Channel.
+	 *
+	 * @param shipStrategyComparator The ship strategy comparator to set.
+	 */
+	public void setShipStrategyComparator(TypeComparatorFactory<?> shipStrategyComparator) {
+		this.shipStrategyComparator = shipStrategyComparator;
+	}
+	
+	/**
+	 * Gets the local strategy comparator from this Channel.
+	 *
+	 * @return The local strategy comparator.
+	 */
+	public TypeComparatorFactory<?> getLocalStrategyComparator() {
+		return localStrategyComparator;
+	}
+	
+	/**
+	 * Sets the local strategy comparator for this Channel.
+	 *
+	 * @param localStrategyComparator The local strategy comparator to set.
+	 */
+	public void setLocalStrategyComparator(TypeComparatorFactory<?> localStrategyComparator) {
+		this.localStrategyComparator = localStrategyComparator;
+	}
+	
+	public double getRelativeMemoryLocalStrategy() {
+		return relativeMemoryLocalStrategy;
+	}
+	
+	public void setRelativeMemoryLocalStrategy(double relativeMemoryLocalStrategy) {
+		this.relativeMemoryLocalStrategy = relativeMemoryLocalStrategy;
+	}
+	
+	public boolean isOnDynamicPath() {
+		return this.source.isOnDynamicPath();
+	}
+	
+	public int getCostWeight() {
+		return this.source.getCostWeight();
+	}
+
+	// --------------------------------------------------------------------------------------------
+	//                                Statistic Estimates
+	// --------------------------------------------------------------------------------------------
+	
+
+	@Override
+	public long getEstimatedOutputSize() {
+		long estimate = this.source.template.getEstimatedOutputSize();
+		return estimate < 0 ? estimate : estimate * this.replicationFactor;
+	}
+
+	@Override
+	public long getEstimatedNumRecords() {
+		long estimate =  this.source.template.getEstimatedNumRecords();
+		return estimate < 0 ? estimate : estimate * this.replicationFactor;
+	}
+	
+	@Override
+	public float getEstimatedAvgWidthPerOutputRecord() {
+		return this.source.template.getEstimatedAvgWidthPerOutputRecord();
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	//                                Data Property Handling
+	// --------------------------------------------------------------------------------------------
+	
+
+	public RequestedGlobalProperties getRequiredGlobalProps() {
+		return requiredGlobalProps;
+	}
+
+	public void setRequiredGlobalProps(RequestedGlobalProperties requiredGlobalProps) {
+		this.requiredGlobalProps = requiredGlobalProps;
+	}
+
+	public RequestedLocalProperties getRequiredLocalProps() {
+		return requiredLocalProps;
+	}
+
+	public void setRequiredLocalProps(RequestedLocalProperties requiredLocalProps) {
+		this.requiredLocalProps = requiredLocalProps;
+	}
+
+	public GlobalProperties getGlobalProperties() {
+		if (this.globalProps == null) {
+			this.globalProps = this.source.getGlobalProperties().clone();
+			switch (this.shipStrategy) {
+				case BROADCAST:
+					this.globalProps.clearUniqueFieldCombinations();
+					this.globalProps.setFullyReplicated();
+					break;
+				case PARTITION_HASH:
+					this.globalProps.setHashPartitioned(this.shipKeys);
+					break;
+				case PARTITION_RANGE:
+					this.globalProps.setRangePartitioned(Utils.createOrdering(this.shipKeys, this.shipSortOrder));
+					break;
+				case FORWARD:
+					break;
+				case PARTITION_RANDOM:
+					this.globalProps.reset();
+					break;
+				case PARTITION_FORCED_REBALANCE:
+					this.globalProps.setForcedRebalanced();
+					break;
+				case PARTITION_CUSTOM:
+					this.globalProps.setCustomPartitioned(this.shipKeys, this.partitioner);
+					break;
+				case NONE:
+					throw new CompilerException("Cannot produce GlobalProperties before ship strategy is set.");
+			}
+		}
+		
+		return this.globalProps;
+	}
+	
+	public LocalProperties getLocalProperties() {
+		if (this.localProps == null) {
+			computeLocalPropertiesAfterShippingOnly();
+			switch (this.localStrategy) {
+				case NONE:
+					break;
+				case SORT:
+				case COMBININGSORT:
+					this.localProps = LocalProperties.forOrdering(Utils.createOrdering(this.localKeys, this.localSortOrder));
+					break;
+				default:
+					throw new CompilerException("Unsupported local strategy for channel.");
+			}
+		}
+		
+		return this.localProps;
+	}
+	
+	private void computeLocalPropertiesAfterShippingOnly() {
+		switch (this.shipStrategy) {
+			case BROADCAST:
+			case PARTITION_HASH:
+			case PARTITION_CUSTOM:
+			case PARTITION_RANGE:
+			case PARTITION_RANDOM:
+			case PARTITION_FORCED_REBALANCE:
+				this.localProps = new LocalProperties();
+				break;
+			case FORWARD:
+				this.localProps = this.source.getLocalProperties();
+				break;
+			case NONE:
+				throw new CompilerException("ShipStrategy has not yet been set.");
+			default:
+				throw new CompilerException("Unknown ShipStrategy.");
+		}
+	}
+	
+	public void adjustGlobalPropertiesForFullParallelismChange() {
+		if (this.shipStrategy == null || this.shipStrategy == ShipStrategyType.NONE) {
+			throw new IllegalStateException("Cannot adjust channel for degree of parallelism " +
+					"change before the ship strategy is set.");
+		}
+		
+		// make sure the properties are acquired
+		if (this.globalProps == null) {
+			getGlobalProperties();
+		}
+		
+		// some strategies globally reestablish properties
+		switch (this.shipStrategy) {
+		case FORWARD:
+			throw new CompilerException("Cannot use FORWARD strategy between operations " +
+					"with different number of parallel instances.");
+		case NONE: // excluded by sanity check. left here for verification check completion
+		case BROADCAST:
+		case PARTITION_HASH:
+		case PARTITION_RANGE:
+		case PARTITION_RANDOM:
+		case PARTITION_FORCED_REBALANCE:
+		case PARTITION_CUSTOM:
+			return;
+		}
+		throw new CompilerException("Unrecognized Ship Strategy Type: " + this.shipStrategy);
+	}
+
+	// --------------------------------------------------------------------------------------------
+	
+	/**
+	 * Utility method used while swapping binary union nodes for n-ary union nodes.
+	 */
+	public void swapUnionNodes(PlanNode newUnionNode) {
+		if (!(this.source instanceof BinaryUnionPlanNode)) {
+			throw new IllegalStateException();
+		} else {
+			this.source = newUnionNode;
+		}
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	public int getMaxDepth() {
+		return this.source.getOptimizerNode().getMaxDepth() + 1;
+	}
+
+	// --------------------------------------------------------------------------------------------
+
+	@Override
+	public String toString() {
+		return "Channel (" + this.source + (this.target == null ? ')' : ") -> (" + this.target + ')') +
+				'[' + this.shipStrategy + "] [" + this.localStrategy + "] " +
+				(this.tempMode == null || this.tempMode == TempMode.NONE ? "{NO-TEMP}" : this.tempMode);
+	}
+
+	@Override
+	public Channel clone() {
+		try {
+			return (Channel) super.clone();
+		} catch (CloneNotSupportedException cnsex) {
+			throw new RuntimeException(cnsex);
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/DualInputPlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/DualInputPlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/DualInputPlanNode.java
new file mode 100644
index 0000000..01c56dd
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/DualInputPlanNode.java
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.plan;
+
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE_AND_DAM;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.NOT_FOUND;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.flink.api.common.operators.util.FieldList;
+import org.apache.flink.api.common.typeutils.TypeComparatorFactory;
+import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
+import org.apache.flink.optimizer.dag.OptimizerNode;
+import org.apache.flink.optimizer.dag.TwoInputNode;
+import org.apache.flink.runtime.operators.DamBehavior;
+import org.apache.flink.runtime.operators.DriverStrategy;
+import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
+import org.apache.flink.util.Visitor;
+
+/**
+ *
+ */
+public class DualInputPlanNode extends PlanNode {
+	
+	protected final Channel input1;
+	protected final Channel input2;
+	
+	protected final FieldList keys1;
+	protected final FieldList keys2;
+	
+	protected final boolean[] sortOrders;
+	
+	private TypeComparatorFactory<?> comparator1;
+	private TypeComparatorFactory<?> comparator2;
+	private TypePairComparatorFactory<?, ?> pairComparator;
+	
+	public Object postPassHelper1;
+	public Object postPassHelper2;
+	
+	// --------------------------------------------------------------------------------------------
+
+	public DualInputPlanNode(OptimizerNode template, String nodeName, Channel input1, Channel input2, DriverStrategy diverStrategy) {
+		this(template, nodeName, input1, input2, diverStrategy, null, null, null);
+	}
+	
+	public DualInputPlanNode(OptimizerNode template, String nodeName, Channel input1, Channel input2,
+			DriverStrategy diverStrategy, FieldList driverKeyFields1, FieldList driverKeyFields2)
+	{
+		this(template, nodeName, input1, input2, diverStrategy, driverKeyFields1, driverKeyFields2,
+									SingleInputPlanNode.getTrueArray(driverKeyFields1.size()));
+	}
+	
+	public DualInputPlanNode(OptimizerNode template, String nodeName, Channel input1, Channel input2, DriverStrategy diverStrategy,
+			FieldList driverKeyFields1, FieldList driverKeyFields2, boolean[] driverSortOrders)
+	{
+		super(template, nodeName, diverStrategy);
+		this.input1 = input1;
+		this.input2 = input2;
+		this.keys1 = driverKeyFields1;
+		this.keys2 = driverKeyFields2;
+		this.sortOrders = driverSortOrders;
+		
+		if (this.input1.getShipStrategy() == ShipStrategyType.BROADCAST) {
+			this.input1.setReplicationFactor(getParallelism());
+		}
+		if (this.input2.getShipStrategy() == ShipStrategyType.BROADCAST) {
+			this.input2.setReplicationFactor(getParallelism());
+		}
+		
+		mergeBranchPlanMaps(input1.getSource(), input2.getSource());
+	}
+
+	// --------------------------------------------------------------------------------------------
+	
+	public TwoInputNode getTwoInputNode() {
+		if (this.template instanceof TwoInputNode) {
+			return (TwoInputNode) this.template;
+		} else {
+			throw new RuntimeException();
+		}
+	}
+	
+	public FieldList getKeysForInput1() {
+		return this.keys1;
+	}
+	
+	public FieldList getKeysForInput2() {
+		return this.keys2;
+	}
+	
+	public boolean[] getSortOrders() {
+		return this.sortOrders;
+	}
+	
+	public TypeComparatorFactory<?> getComparator1() {
+		return this.comparator1;
+	}
+	
+	public TypeComparatorFactory<?> getComparator2() {
+		return this.comparator2;
+	}
+	
+	public void setComparator1(TypeComparatorFactory<?> comparator) {
+		this.comparator1 = comparator;
+	}
+	
+	public void setComparator2(TypeComparatorFactory<?> comparator) {
+		this.comparator2 = comparator;
+	}
+	
+	public TypePairComparatorFactory<?, ?> getPairComparator() {
+		return this.pairComparator;
+	}
+	
+	public void setPairComparator(TypePairComparatorFactory<?, ?> comparator) {
+		this.pairComparator = comparator;
+	}
+	
+	/**
+	 * Gets the first input channel to this node.
+	 * 
+	 * @return The first input channel to this node.
+	 */
+	public Channel getInput1() {
+		return this.input1;
+	}
+	
+	/**
+	 * Gets the second input channel to this node.
+	 * 
+	 * @return The second input channel to this node.
+	 */
+	public Channel getInput2() {
+		return this.input2;
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+
+	@Override
+	public void accept(Visitor<PlanNode> visitor) {
+		if (visitor.preVisit(this)) {
+			this.input1.getSource().accept(visitor);
+			this.input2.getSource().accept(visitor);
+			
+			for (Channel broadcastInput : getBroadcastInputs()) {
+				broadcastInput.getSource().accept(visitor);
+			}
+			
+			visitor.postVisit(this);
+		}
+	}
+	
+
+	@Override
+	public Iterable<PlanNode> getPredecessors() {
+		if (getBroadcastInputs() == null || getBroadcastInputs().isEmpty()) {
+			return Arrays.asList(this.input1.getSource(), this.input2.getSource());
+		} else {
+			List<PlanNode> preds = new ArrayList<PlanNode>();
+			
+			preds.add(input1.getSource());
+			preds.add(input2.getSource());
+
+			for (Channel c : getBroadcastInputs()) {
+				preds.add(c.getSource());
+			}
+			
+			return preds;
+		}
+	}
+
+	@Override
+	public Iterable<Channel> getInputs() {
+		return Arrays.asList(this.input1, this.input2);
+	}
+
+
+	@Override
+	public SourceAndDamReport hasDamOnPathDownTo(PlanNode source) {
+		if (source == this) {
+			return FOUND_SOURCE;
+		}
+		
+		// check first input
+		SourceAndDamReport res1 = this.input1.getSource().hasDamOnPathDownTo(source);
+		if (res1 == FOUND_SOURCE_AND_DAM) {
+			return FOUND_SOURCE_AND_DAM;
+		}
+		else if (res1 == FOUND_SOURCE) {
+			if (this.input1.getLocalStrategy().dams() || this.input1.getTempMode().breaksPipeline() ||
+					getDriverStrategy().firstDam() == DamBehavior.FULL_DAM) {
+				return FOUND_SOURCE_AND_DAM;
+			} else {
+				return FOUND_SOURCE;
+			}
+		}
+		else {
+			SourceAndDamReport res2 = this.input2.getSource().hasDamOnPathDownTo(source);
+			if (res2 == FOUND_SOURCE_AND_DAM) {
+				return FOUND_SOURCE_AND_DAM;
+			}
+			else if (res2 == FOUND_SOURCE) {
+				if (this.input2.getLocalStrategy().dams() || this.input2.getTempMode().breaksPipeline() ||
+						getDriverStrategy().secondDam() == DamBehavior.FULL_DAM) {
+					return FOUND_SOURCE_AND_DAM;
+				} else {
+					return FOUND_SOURCE;
+				}
+			}
+			else {
+				// NOT_FOUND
+				// check the broadcast inputs
+				
+				for (NamedChannel nc : getBroadcastInputs()) {
+					SourceAndDamReport bcRes = nc.getSource().hasDamOnPathDownTo(source);
+					if (bcRes != NOT_FOUND) {
+						// broadcast inputs are always dams
+						return FOUND_SOURCE_AND_DAM;
+					}
+				}
+				return NOT_FOUND;
+			}
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/FlinkPlan.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/FlinkPlan.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/FlinkPlan.java
new file mode 100644
index 0000000..d146c83
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/FlinkPlan.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.plan;
+
+/**
+ * A common interface for compiled Flink plans for both batch and streaming
+ * processing programs.
+ * 
+ */
+public interface FlinkPlan {
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/IterationPlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/IterationPlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/IterationPlanNode.java
new file mode 100644
index 0000000..38f76b2
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/IterationPlanNode.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.plan;
+
+import org.apache.flink.optimizer.dag.IterationNode;
+import org.apache.flink.util.Visitor;
+
+/**
+ *
+ */
+public interface IterationPlanNode {
+	
+	void acceptForStepFunction(Visitor<PlanNode> visitor);
+	
+	IterationNode getIterationNode();
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/NAryUnionPlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/NAryUnionPlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/NAryUnionPlanNode.java
new file mode 100644
index 0000000..3650eea
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/NAryUnionPlanNode.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.plan;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.flink.optimizer.costs.Costs;
+import org.apache.flink.optimizer.dag.BinaryUnionNode;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.runtime.operators.DriverStrategy;
+import org.apache.flink.util.IterableIterator;
+import org.apache.flink.util.Visitor;
+
+/**
+ * A union operation over multiple inputs (2 or more).
+ */
+public class NAryUnionPlanNode extends PlanNode {
+	
+	private final List<Channel> inputs;
+	
+	/**
+	 * @param template
+	 */
+	public NAryUnionPlanNode(BinaryUnionNode template, List<Channel> inputs, GlobalProperties gProps,
+			Costs cumulativeCosts)
+	{
+		super(template, "Union", DriverStrategy.NONE);
+		
+		this.inputs = inputs;
+		this.globalProps = gProps;
+		this.localProps = new LocalProperties();
+		this.nodeCosts = new Costs();
+		this.cumulativeCosts = cumulativeCosts;
+	}
+
+	@Override
+	public void accept(Visitor<PlanNode> visitor) {
+		visitor.preVisit(this);
+		for (Channel c : this.inputs) {
+			c.getSource().accept(visitor);
+		}
+		visitor.postVisit(this);
+	}
+	
+	public List<Channel> getListOfInputs() {
+		return this.inputs;
+	}
+
+	@Override
+	public Iterable<Channel> getInputs() {
+		return Collections.unmodifiableList(this.inputs);
+	}
+
+	@Override
+	public Iterable<PlanNode> getPredecessors() {
+		final Iterator<Channel> channels = this.inputs.iterator();
+		return new IterableIterator<PlanNode>() {
+
+			@Override
+			public boolean hasNext() {
+				return channels.hasNext();
+			}
+
+			@Override
+			public PlanNode next() {
+				return channels.next().getSource();
+			}
+
+			@Override
+			public void remove() {
+				throw new UnsupportedOperationException();
+			}
+			
+			@Override
+			public Iterator<PlanNode> iterator() {
+				return this;
+			}
+		};
+	}
+
+	@Override
+	public SourceAndDamReport hasDamOnPathDownTo(PlanNode source) {
+		// this node is used after the plan enumeration. consequently, this will never be invoked here
+		throw new UnsupportedOperationException();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/NamedChannel.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/NamedChannel.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/NamedChannel.java
new file mode 100644
index 0000000..da97e61
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/NamedChannel.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.plan;
+
+import org.apache.flink.optimizer.dag.TempMode;
+
+public class NamedChannel extends Channel {
+
+	private final String name;
+
+	/**
+	 * Initializes NamedChannel.
+	 * 
+	 * @param sourceNode
+	 */
+	public NamedChannel(String name, PlanNode sourceNode) {
+		super(sourceNode);
+		this.name = name;
+	}
+
+	public NamedChannel(String name, PlanNode sourceNode, TempMode tempMode) {
+		super(sourceNode, tempMode);
+		this.name = name;
+	}
+
+	public String getName() {
+		return this.name;
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/OptimizedPlan.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/OptimizedPlan.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/OptimizedPlan.java
new file mode 100644
index 0000000..d56be87
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/OptimizedPlan.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.plan;
+
+import java.util.Collection;
+
+import org.apache.flink.api.common.Plan;
+import org.apache.flink.util.Visitable;
+import org.apache.flink.util.Visitor;
+
+/**
+ * The execution plan generated by the Optimizer. It contains {@link PlanNode}s
+ * and {@link Channel}s that describe exactly how the program should be executed.
+ * It defines all ship strategies (local pipe, shuffle, broadcast, rebalance), all
+ * operator strategies (sorting-merge join, hash join, sorted grouping, ...),
+ * and the data exchange modes (batched, pipelined).
+ */
+public class OptimizedPlan implements FlinkPlan, Visitable<PlanNode>  {
+	
+	/** The data sources in the plan. */
+	private final Collection<SourcePlanNode> dataSources;
+
+	/** The data sinks in the plan. */
+	private final Collection<SinkPlanNode> dataSinks;
+
+	/** All nodes in the optimizer plan. */
+	private final Collection<PlanNode> allNodes;
+	
+	/** The original program. */
+	private final Plan originalProgram;
+
+	/** Name of the job */
+	private final String jobName;
+
+	/**
+	 * Creates a new instance of this optimizer plan container. The plan is given and fully
+	 * described by the data sources, sinks and the collection of all nodes.
+	 * 
+	 * @param sources The data sources.
+	 * @param sinks The data sinks.
+	 * @param allNodes A collection containing all nodes in the plan.
+	 * @param jobName The name of the program
+	 */
+	public OptimizedPlan(Collection<SourcePlanNode> sources, Collection<SinkPlanNode> sinks,
+			Collection<PlanNode> allNodes, String jobName, Plan programPlan)
+	{
+		this.dataSources = sources;
+		this.dataSinks = sinks;
+		this.allNodes = allNodes;
+		this.jobName = jobName;
+		this.originalProgram = programPlan;
+	}
+
+	/**
+	 * Gets the data sources from this OptimizedPlan.
+	 * 
+	 * @return The data sources.
+	 */
+	public Collection<SourcePlanNode> getDataSources() {
+		return dataSources;
+	}
+
+	/**
+	 * Gets the data sinks from this OptimizedPlan.
+	 * 
+	 * @return The data sinks.
+	 */
+	public Collection<SinkPlanNode> getDataSinks() {
+		return dataSinks;
+	}
+
+	/**
+	 * Gets all the nodes from this OptimizedPlan.
+	 * 
+	 * @return All nodes.
+	 */
+	public Collection<PlanNode> getAllNodes() {
+		return allNodes;
+	}
+
+	/**
+	 * Returns the name of the program.
+	 * 
+	 * @return The name of the program.
+	 */
+	public String getJobName() {
+		return this.jobName;
+	}
+	
+	/**
+	 * Gets the original program plan from which this optimized plan was created.
+	 * 
+	 * @return The original program plan.
+	 */
+	public Plan getOriginalPactPlan() {
+		return this.originalProgram;
+	}
+
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Applies the given visitor top down to all nodes, starting at the sinks.
+	 * 
+	 * @param visitor
+	 *        The visitor to apply to the nodes in this plan.
+	 * @see org.apache.flink.util.Visitable#accept(org.apache.flink.util.Visitor)
+	 */
+	@Override
+	public void accept(Visitor<PlanNode> visitor) {
+		for (SinkPlanNode node : this.dataSinks) {
+			node.accept(visitor);
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/PlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/PlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/PlanNode.java
new file mode 100644
index 0000000..6f634fb
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/PlanNode.java
@@ -0,0 +1,573 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.plan;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.flink.api.common.operators.Operator;
+import org.apache.flink.api.common.operators.util.FieldSet;
+import org.apache.flink.optimizer.CompilerException;
+import org.apache.flink.optimizer.costs.Costs;
+import org.apache.flink.optimizer.dag.OptimizerNode;
+import org.apache.flink.optimizer.dag.OptimizerNode.UnclosedBranchDescriptor;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.optimizer.plandump.DumpableConnection;
+import org.apache.flink.optimizer.plandump.DumpableNode;
+import org.apache.flink.runtime.operators.DriverStrategy;
+import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
+import org.apache.flink.runtime.operators.util.LocalStrategy;
+import org.apache.flink.util.Visitable;
+
+/**
+ * The representation of a data exchange between to operators. The data exchange can realize a shipping strategy, 
+ * which established global properties, and a local strategy, which establishes local properties.
+ * <p>
+ * Because we currently deal only with plans where the operator order is fixed, many properties are equal
+ * among candidates and are determined prior to the enumeration (such as for example constant/dynamic path membership).
+ * Hence, many methods will delegate to the {@code OptimizerNode} that represents the node this candidate was
+ * created for.
+ */
+public abstract class PlanNode implements Visitable<PlanNode>, DumpableNode<PlanNode> {
+	
+	protected final OptimizerNode template;
+	
+	protected final List<Channel> outChannels;
+	
+	private List<NamedChannel> broadcastInputs;
+	
+	private final String nodeName; 
+	
+	private DriverStrategy driverStrategy;	// The local strategy (sorting / hashing, ...)
+	
+	protected LocalProperties localProps; 			// local properties of the data produced by this node
+
+	protected GlobalProperties globalProps;			// global properties of the data produced by this node
+	
+	protected Map<OptimizerNode, PlanNode> branchPlan; // the actual plan alternative chosen at a branch point
+	
+	protected Costs nodeCosts;						// the costs incurred by this node
+
+	protected Costs cumulativeCosts;					// the cumulative costs of all operators in the sub-tree
+	
+	private double relativeMemoryPerSubTask;					// the amount of memory dedicated to each task, in bytes
+	
+	private int parallelism;
+	
+	private boolean pFlag;							// flag for the internal pruning algorithm
+	
+	// --------------------------------------------------------------------------------------------
+	
+	public PlanNode(OptimizerNode template, String nodeName, DriverStrategy strategy) {
+		this.outChannels = new ArrayList<Channel>(2);
+		this.broadcastInputs = new ArrayList<NamedChannel>();
+		this.template = template;
+		this.nodeName = nodeName;
+		this.driverStrategy = strategy;
+		
+		this.parallelism = template.getParallelism();
+
+		// check, if there is branch at this node. if yes, this candidate must be associated with
+		// the branching template node.
+		if (template.isBranching()) {
+			this.branchPlan = new HashMap<OptimizerNode, PlanNode>(6);
+			this.branchPlan.put(template, this);
+		}
+	}
+	
+	protected void mergeBranchPlanMaps(PlanNode pred1, PlanNode pred2) {
+		mergeBranchPlanMaps(pred1.branchPlan, pred2.branchPlan);
+	}
+	
+	protected void mergeBranchPlanMaps(Map<OptimizerNode, PlanNode> branchPlan1, Map<OptimizerNode, PlanNode> branchPlan2) {
+		// merge the branchPlan maps according the template's uncloseBranchesStack
+		if (this.template.hasUnclosedBranches()) {
+			if (this.branchPlan == null) {
+				this.branchPlan = new HashMap<OptimizerNode, PlanNode>(8);
+			}
+	
+			for (UnclosedBranchDescriptor uc : this.template.getOpenBranches()) {
+				OptimizerNode brancher = uc.getBranchingNode();
+				PlanNode selectedCandidate = null;
+	
+				if (branchPlan1 != null) {
+					// predecessor 1 has branching children, see if it got the branch we are looking for
+					selectedCandidate = branchPlan1.get(brancher);
+				}
+				
+				if (selectedCandidate == null && branchPlan2 != null) {
+					// predecessor 2 has branching children, see if it got the branch we are looking for
+					selectedCandidate = branchPlan2.get(brancher);
+				}
+				
+				// it may be that the branch candidate is only found once the broadcast variables are set
+				if (selectedCandidate != null) {
+					this.branchPlan.put(brancher, selectedCandidate);
+				}
+			}
+		}
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	//                                           Accessors
+	// --------------------------------------------------------------------------------------------
+	
+	/**
+	 * Gets the node from the optimizer DAG for which this plan candidate node was created.
+	 * 
+	 * @return The optimizer's DAG node.
+	 */
+	public OptimizerNode getOriginalOptimizerNode() {
+		return this.template;
+	}
+	
+	/**
+	 * Gets the program operator that this node represents in the plan.
+	 * 
+	 * @return The program operator this node represents in the plan.
+	 */
+	public Operator<?> getProgramOperator() {
+		return this.template.getOperator();
+	}
+	
+	/**
+	 * Gets the name of the plan node.
+	 * 
+	 * @return The name of the plan node.
+	 */
+	public String getNodeName() {
+		return this.nodeName;
+	}
+	
+	public int getMemoryConsumerWeight() {
+		return this.driverStrategy.isMaterializing() ? 1 : 0;
+	}
+	
+	/**
+	 * Gets the memory dedicated to each sub-task for this node.
+	 * 
+	 * @return The memory per task, in bytes.
+	 */
+	public double getRelativeMemoryPerSubTask() {
+		return this.relativeMemoryPerSubTask;
+	}
+
+	/**
+	 * Sets the memory dedicated to each task for this node.
+	 * 
+	 * @param relativeMemoryPerSubtask The relative memory per sub-task
+	 */
+	public void setRelativeMemoryPerSubtask(double relativeMemoryPerSubtask) {
+		this.relativeMemoryPerSubTask = relativeMemoryPerSubtask;
+	}
+	
+	/**
+	 * Gets the driver strategy from this node. This determines for example for a <i>match</i> Pact whether
+	 * to use a merge or a hybrid hash strategy.
+	 * 
+	 * @return The driver strategy.
+	 */
+	public DriverStrategy getDriverStrategy() {
+		return this.driverStrategy;
+	}
+	
+	/**
+	 * Sets the driver strategy for this node. Usually should not be changed.
+	 * 
+	 * @param newDriverStrategy The driver strategy.
+	 */
+	public void setDriverStrategy(DriverStrategy newDriverStrategy) {
+		this.driverStrategy = newDriverStrategy;
+	}
+	
+	public void initProperties(GlobalProperties globals, LocalProperties locals) {
+		if (this.globalProps != null || this.localProps != null) {
+			throw new IllegalStateException();
+		}
+		this.globalProps = globals;
+		this.localProps = locals;
+	}
+	
+	/**
+	 * Gets the local properties from this PlanNode.
+	 *
+	 * @return The local properties.
+	 */
+	public LocalProperties getLocalProperties() {
+		return this.localProps;
+	}
+	
+	/**
+	 * Gets the global properties from this PlanNode.
+	 *
+	 * @return The global properties.
+	 */
+	public GlobalProperties getGlobalProperties() {
+		return this.globalProps;
+	}
+	
+	/**
+	 * Gets the costs incurred by this node. The costs reflect also the costs incurred by the shipping strategies
+	 * of the incoming connections.
+	 * 
+	 * @return The node-costs, or null, if not yet set.
+	 */
+	public Costs getNodeCosts() {
+		return this.nodeCosts;
+	}
+
+	/**
+	 * Gets the cumulative costs of this nose. The cumulative costs are the sum of the costs
+	 * of this node and of all nodes in the subtree below this node.
+	 * 
+	 * @return The cumulative costs, or null, if not yet set.
+	 */
+	public Costs getCumulativeCosts() {
+		return this.cumulativeCosts;
+	}
+
+	public Costs getCumulativeCostsShare() {
+		if (this.cumulativeCosts == null) {
+			return null;
+		} else {
+			Costs result = cumulativeCosts.clone();
+			if (this.template.getOutgoingConnections() != null) {
+				int outDegree = this.template.getOutgoingConnections().size();
+				if (outDegree > 0) {
+					result.divideBy(outDegree);
+				}
+			}
+
+			return result;
+		}
+	}
+
+	
+	/**
+	 * Sets the basic cost for this node to the given value, and sets the cumulative costs
+	 * to those costs plus the cost shares of all inputs (regular and broadcast).
+	 * 
+	 * @param nodeCosts	 The already knows costs for this node
+	 * 						(this cost a produces by a concrete {@code OptimizerNode} subclass.
+	 */
+	public void setCosts(Costs nodeCosts) {
+		// set the node costs
+		this.nodeCosts = nodeCosts;
+		
+		// the cumulative costs are the node costs plus the costs of all inputs
+		this.cumulativeCosts = nodeCosts.clone();
+		
+		// add all the normal inputs
+		for (PlanNode pred : getPredecessors()) {
+			
+			Costs parentCosts = pred.getCumulativeCostsShare();
+			if (parentCosts != null) {
+				this.cumulativeCosts.addCosts(parentCosts);
+			} else {
+				throw new CompilerException("Trying to set the costs of an operator before the predecessor costs are computed.");
+			}
+		}
+		
+		// add all broadcast variable inputs
+		if (this.broadcastInputs != null) {
+			for (NamedChannel nc : this.broadcastInputs) {
+				Costs bcInputCost = nc.getSource().getCumulativeCostsShare();
+				if (bcInputCost != null) {
+					this.cumulativeCosts.addCosts(bcInputCost);
+				} else {
+					throw new CompilerException("Trying to set the costs of an operator before the broadcast input costs are computed.");
+				}
+			}
+		}
+	}
+	
+	public void setParallelism(int parallelism) {
+		this.parallelism = parallelism;
+	}
+	
+	public int getParallelism() {
+		return this.parallelism;
+	}
+	
+	public long getGuaranteedAvailableMemory() {
+		return this.template.getMinimalMemoryAcrossAllSubTasks();
+	}
+
+	public Map<OptimizerNode, PlanNode> getBranchPlan() {
+		return branchPlan;
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	//                               Input, Predecessors, Successors
+	// --------------------------------------------------------------------------------------------
+	
+	public abstract Iterable<Channel> getInputs();
+	
+	@Override
+	public abstract Iterable<PlanNode> getPredecessors();
+	
+	/**
+	 * Sets a list of all broadcast inputs attached to this node.
+	 */
+	public void setBroadcastInputs(List<NamedChannel> broadcastInputs) {
+		if (broadcastInputs != null) {
+			this.broadcastInputs = broadcastInputs;
+			
+			// update the branch map
+			for (NamedChannel nc : broadcastInputs) {
+				PlanNode source = nc.getSource();
+				
+				mergeBranchPlanMaps(branchPlan, source.branchPlan);
+			}
+		}
+		
+		// do a sanity check that if we are branching, we have now candidates for each branch point
+		if (this.template.hasUnclosedBranches()) {
+			if (this.branchPlan == null) {
+				throw new CompilerException("Branching and rejoining logic did not find a candidate for the branching point.");
+			}
+	
+			for (UnclosedBranchDescriptor uc : this.template.getOpenBranches()) {
+				OptimizerNode brancher = uc.getBranchingNode();
+				if (this.branchPlan.get(brancher) == null) {
+					throw new CompilerException("Branching and rejoining logic did not find a candidate for the branching point.");
+				}
+			}
+		}
+	}
+	
+	/**
+	 * Gets a list of all broadcast inputs attached to this node.
+	 */
+	public List<NamedChannel> getBroadcastInputs() {
+		return this.broadcastInputs;
+	}
+	
+	/**
+	 * Adds a channel to a successor node to this node.
+	 * 
+	 * @param channel The channel to the successor.
+	 */
+	public void addOutgoingChannel(Channel channel) {
+		this.outChannels.add(channel);
+	}
+	
+	/**
+	 * Gets a list of all outgoing channels leading to successors.
+	 * 
+	 * @return A list of all channels leading to successors.
+	 */
+	public List<Channel> getOutgoingChannels() {
+		return this.outChannels;
+	}
+
+	// --------------------------------------------------------------------------------------------
+	//                                Miscellaneous
+	// --------------------------------------------------------------------------------------------
+	
+	public void updatePropertiesWithUniqueSets(Set<FieldSet> uniqueFieldCombinations) {
+		if (uniqueFieldCombinations == null || uniqueFieldCombinations.isEmpty()) {
+			return;
+		}
+		for (FieldSet fields : uniqueFieldCombinations) {
+			this.globalProps.addUniqueFieldCombination(fields);
+			this.localProps = this.localProps.addUniqueFields(fields);
+		}
+	}
+
+	public PlanNode getCandidateAtBranchPoint(OptimizerNode branchPoint) {
+		if (branchPlan == null) {
+			return null;
+		} else {
+			return this.branchPlan.get(branchPoint);
+		}
+	}
+	
+	/**
+	 * Sets the pruning marker to true.
+	 */
+	public void setPruningMarker() {
+		this.pFlag = true;
+	}
+	
+	/**
+	 * Checks whether the pruning marker was set.
+	 * 
+	 * @return True, if the pruning marker was set, false otherwise.
+	 */
+	public boolean isPruneMarkerSet() {
+		return this.pFlag;
+	}
+	
+	public boolean isOnDynamicPath() {
+		return this.template.isOnDynamicPath();
+	}
+	
+	public int getCostWeight() {
+		return this.template.getCostWeight();
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	/**
+	 * Checks whether this node has a dam on the way down to the given source node. This method
+	 * returns either that (a) the source node is not found as a (transitive) child of this node,
+	 * (b) the node is found, but no dam is on the path, or (c) the node is found and a dam is on
+	 * the path.
+	 * 
+	 * @param source The node on the path to which the dam is sought.
+	 * @return The result whether the node is found and whether a dam is on the path.
+	 */
+	public abstract SourceAndDamReport hasDamOnPathDownTo(PlanNode source);
+	
+	public FeedbackPropertiesMeetRequirementsReport checkPartialSolutionPropertiesMet(PlanNode partialSolution, GlobalProperties feedbackGlobal, LocalProperties feedbackLocal) {
+		if (this == partialSolution) {
+			return FeedbackPropertiesMeetRequirementsReport.PENDING;
+		}
+		
+		boolean found = false;
+		boolean allMet = true;
+		boolean allLocallyMet = true;
+		
+		for (Channel input : getInputs()) {
+			FeedbackPropertiesMeetRequirementsReport inputState = input.getSource().checkPartialSolutionPropertiesMet(partialSolution, feedbackGlobal, feedbackLocal);
+			
+			if (inputState == FeedbackPropertiesMeetRequirementsReport.NO_PARTIAL_SOLUTION) {
+				continue;
+			}
+			else if (inputState == FeedbackPropertiesMeetRequirementsReport.MET) {
+				found = true;
+				continue;
+			}
+			else if (inputState == FeedbackPropertiesMeetRequirementsReport.NOT_MET) {
+				return FeedbackPropertiesMeetRequirementsReport.NOT_MET;
+			}
+			else {
+				found = true;
+				
+				// the partial solution was on the path here. check whether the channel requires
+				// certain properties that are met, or whether the channel introduces new properties
+				
+				// if the plan introduces new global properties, then we can stop looking whether
+				// the feedback properties are sufficient to meet the requirements
+				if (input.getShipStrategy() != ShipStrategyType.FORWARD && input.getShipStrategy() != ShipStrategyType.NONE) {
+					continue;
+				}
+				
+				// first check whether this channel requires something that is not met
+				if (input.getRequiredGlobalProps() != null && !input.getRequiredGlobalProps().isMetBy(feedbackGlobal)) {
+					return FeedbackPropertiesMeetRequirementsReport.NOT_MET;
+				}
+				
+				// in general, not everything is met here already
+				allMet = false;
+				
+				// if the plan introduces new local properties, we can stop checking for matching local properties
+				if (inputState != FeedbackPropertiesMeetRequirementsReport.PENDING_LOCAL_MET) {
+					
+					if (input.getLocalStrategy() == LocalStrategy.NONE) {
+						
+						if (input.getRequiredLocalProps() != null && !input.getRequiredLocalProps().isMetBy(feedbackLocal)) {
+							return FeedbackPropertiesMeetRequirementsReport.NOT_MET;
+						}
+						
+						allLocallyMet = false;
+					}
+				}
+			}
+		}
+		
+		if (!found) {
+			return FeedbackPropertiesMeetRequirementsReport.NO_PARTIAL_SOLUTION;
+		} else if (allMet) {
+			return FeedbackPropertiesMeetRequirementsReport.MET;
+		} else if (allLocallyMet) {
+			return FeedbackPropertiesMeetRequirementsReport.PENDING_LOCAL_MET;
+		} else {
+			return FeedbackPropertiesMeetRequirementsReport.PENDING;
+		}
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+
+	@Override
+	public String toString() {
+		return this.template.getName() + " \"" + getProgramOperator().getName() + "\" : " + this.driverStrategy +
+				" [[ " + this.globalProps + " ]] [[ " + this.localProps + " ]]";
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	@Override
+	public OptimizerNode getOptimizerNode() {
+		return this.template;
+	}
+
+	@Override
+	public PlanNode getPlanNode() {
+		return this;
+	}
+
+	@Override
+	public Iterable<DumpableConnection<PlanNode>> getDumpableInputs() {
+		List<DumpableConnection<PlanNode>> allInputs = new ArrayList<DumpableConnection<PlanNode>>();
+		
+		for (Channel c : getInputs()) {
+			allInputs.add(c);
+		}
+		
+		for (NamedChannel c : getBroadcastInputs()) {
+			allInputs.add(c);
+		}
+		
+		return allInputs;
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	public static enum SourceAndDamReport {
+		NOT_FOUND, FOUND_SOURCE, FOUND_SOURCE_AND_DAM;
+	}
+	
+	
+	
+	public static enum FeedbackPropertiesMeetRequirementsReport {
+		/** Indicates that the path is irrelevant */
+		NO_PARTIAL_SOLUTION,
+		
+		/** Indicates that the question whether the properties are met has been determined pending
+		 * dependent on global and local properties */
+		PENDING,
+		
+		/** Indicates that the question whether the properties are met has been determined pending
+		 * dependent on global properties only */
+		PENDING_LOCAL_MET,
+		
+		/** Indicates that the question whether the properties are met has been determined true */
+		MET,
+		
+		/** Indicates that the question whether the properties are met has been determined false */
+		NOT_MET;
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SingleInputPlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SingleInputPlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SingleInputPlanNode.java
new file mode 100644
index 0000000..b928be7
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SingleInputPlanNode.java
@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.plan;
+
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE_AND_DAM;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.NOT_FOUND;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+
+import org.apache.flink.api.common.operators.util.FieldList;
+import org.apache.flink.api.common.typeutils.TypeComparatorFactory;
+import org.apache.flink.optimizer.CompilerException;
+import org.apache.flink.optimizer.dag.OptimizerNode;
+import org.apache.flink.optimizer.dag.SingleInputNode;
+import org.apache.flink.runtime.operators.DamBehavior;
+import org.apache.flink.runtime.operators.DriverStrategy;
+import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
+import org.apache.flink.util.Visitor;
+
+/**
+ * 
+ */
+public class SingleInputPlanNode extends PlanNode {
+	
+	protected final Channel input;
+	
+	protected final FieldList[] driverKeys;
+	
+	protected final boolean[][] driverSortOrders;
+	
+	private TypeComparatorFactory<?>[] comparators;
+	
+	public Object postPassHelper;
+	
+	// --------------------------------------------------------------------------------------------
+
+	public SingleInputPlanNode(OptimizerNode template, String nodeName, Channel input, DriverStrategy driverStrategy) {
+		this(template, nodeName, input, driverStrategy, null, null);
+	}
+	
+	public SingleInputPlanNode(OptimizerNode template, String nodeName, Channel input, 
+			DriverStrategy driverStrategy, FieldList driverKeyFields)
+	{
+		this(template, nodeName, input, driverStrategy, driverKeyFields, getTrueArray(driverKeyFields.size()));
+	}
+	
+	public SingleInputPlanNode(OptimizerNode template, String nodeName, Channel input, 
+			DriverStrategy driverStrategy, FieldList driverKeyFields, boolean[] driverSortOrders)
+	{
+		super(template, nodeName, driverStrategy);
+		this.input = input;
+		
+		this.comparators = new TypeComparatorFactory<?>[driverStrategy.getNumRequiredComparators()];
+		this.driverKeys = new FieldList[driverStrategy.getNumRequiredComparators()];
+		this.driverSortOrders = new boolean[driverStrategy.getNumRequiredComparators()][];
+		
+		if(driverStrategy.getNumRequiredComparators() > 0) {
+			this.driverKeys[0] = driverKeyFields;
+			this.driverSortOrders[0] = driverSortOrders;
+		}
+		
+		if (this.input.getShipStrategy() == ShipStrategyType.BROADCAST) {
+			this.input.setReplicationFactor(getParallelism());
+		}
+		
+		final PlanNode predNode = input.getSource();
+		
+		if (predNode.branchPlan != null && !predNode.branchPlan.isEmpty()) {
+			
+			if (this.branchPlan == null) {
+				this.branchPlan = new HashMap<OptimizerNode, PlanNode>();
+			}
+			this.branchPlan.putAll(predNode.branchPlan);
+		}
+	}
+
+	// --------------------------------------------------------------------------------------------
+	
+	public SingleInputNode getSingleInputNode() {
+		if (this.template instanceof SingleInputNode) {
+			return (SingleInputNode) this.template;
+		} else {
+			throw new RuntimeException();
+		}
+	}
+	
+	/**
+	 * Gets the input channel to this node.
+	 * 
+	 * @return The input channel to this node.
+	 */
+	public Channel getInput() {
+		return this.input;
+	}
+	
+	/**
+	 * Gets the predecessor of this node, i.e. the source of the input channel.
+	 * 
+	 * @return The predecessor of this node.
+	 */
+	public PlanNode getPredecessor() {
+		return this.input.getSource();
+	}
+	
+	/**
+	 * Sets the key field indexes for the specified driver comparator.
+	 * 
+	 * @param keys The key field indexes for the specified driver comparator.
+	 * @param id The ID of the driver comparator.
+	 */
+	public void setDriverKeyInfo(FieldList keys, int id) {
+		this.setDriverKeyInfo(keys, getTrueArray(keys.size()), id);
+	}
+	
+	/**
+	 * Sets the key field information for the specified driver comparator.
+	 * 
+	 * @param keys The key field indexes for the specified driver comparator.
+	 * @param sortOrder The key sort order for the specified driver comparator.
+	 * @param id The ID of the driver comparator.
+	 */
+	public void setDriverKeyInfo(FieldList keys, boolean[] sortOrder, int id) {
+		if(id < 0 || id >= driverKeys.length) {
+			throw new CompilerException("Invalid id for driver key information. DriverStrategy requires only "
+											+super.getDriverStrategy().getNumRequiredComparators()+" comparators.");
+		}
+		this.driverKeys[id] = keys;
+		this.driverSortOrders[id] = sortOrder;
+	}
+	
+	/**
+	 * Gets the key field indexes for the specified driver comparator.
+	 * 
+	 * @param id The id of the driver comparator for which the key field indexes are requested.
+	 * @return The key field indexes of the specified driver comparator.
+	 */
+	public FieldList getKeys(int id) {
+		return this.driverKeys[id];
+	}
+	
+	/**
+	 * Gets the sort order for the specified driver comparator.
+	 * 
+	 * @param id The id of the driver comparator for which the sort order is requested.
+	 * @return The sort order of the specified driver comparator.
+	 */
+	public boolean[] getSortOrders(int id) {
+		return driverSortOrders[id];
+	}
+	
+	/**
+	 * Gets the specified comparator from this PlanNode.
+	 * 
+	 * @param id The ID of the requested comparator.
+	 *
+	 * @return The specified comparator.
+	 */
+	public TypeComparatorFactory<?> getComparator(int id) {
+		return comparators[id];
+	}
+	
+	/**
+	 * Sets the specified comparator for this PlanNode.
+	 *
+	 * @param comparator The comparator to set.
+	 * @param id The ID of the comparator to set.
+	 */
+	public void setComparator(TypeComparatorFactory<?> comparator, int id) {
+		this.comparators[id] = comparator;
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+
+	@Override
+	public void accept(Visitor<PlanNode> visitor) {
+		if (visitor.preVisit(this)) {
+			this.input.getSource().accept(visitor);
+			
+			for (Channel broadcastInput : getBroadcastInputs()) {
+				broadcastInput.getSource().accept(visitor);
+			}
+			
+			visitor.postVisit(this);
+		}
+	}
+
+
+	@Override
+	public Iterable<PlanNode> getPredecessors() {
+		if (getBroadcastInputs() == null || getBroadcastInputs().isEmpty()) {
+			return Collections.singleton(this.input.getSource());
+		}
+		else {
+			List<PlanNode> preds = new ArrayList<PlanNode>();
+			preds.add(input.getSource());
+			
+			for (Channel c : getBroadcastInputs()) {
+				preds.add(c.getSource());
+			}
+			
+			return preds;
+		}
+	}
+
+
+	@Override
+	public Iterable<Channel> getInputs() {
+		return Collections.singleton(this.input);
+	}
+
+	@Override
+	public SourceAndDamReport hasDamOnPathDownTo(PlanNode source) {
+		if (source == this) {
+			return FOUND_SOURCE;
+		}
+		SourceAndDamReport res = this.input.getSource().hasDamOnPathDownTo(source);
+		if (res == FOUND_SOURCE_AND_DAM) {
+			return FOUND_SOURCE_AND_DAM;
+		}
+		else if (res == FOUND_SOURCE) {
+			return (this.input.getLocalStrategy().dams() || this.input.getTempMode().breaksPipeline() ||
+					getDriverStrategy().firstDam() == DamBehavior.FULL_DAM) ?
+				FOUND_SOURCE_AND_DAM : FOUND_SOURCE;
+		}
+		else {
+			// NOT_FOUND
+			// check the broadcast inputs
+			
+			for (NamedChannel nc : getBroadcastInputs()) {
+				SourceAndDamReport bcRes = nc.getSource().hasDamOnPathDownTo(source);
+				if (bcRes != NOT_FOUND) {
+					// broadcast inputs are always dams
+					return FOUND_SOURCE_AND_DAM;
+				}
+			}
+			return NOT_FOUND;
+		}
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	protected static boolean[] getTrueArray(int length) {
+		final boolean[] a = new boolean[length];
+		for (int i = 0; i < length; i++) {
+			a[i] = true;
+		}
+		return a;
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SinkJoinerPlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SinkJoinerPlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SinkJoinerPlanNode.java
new file mode 100644
index 0000000..451484d
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SinkJoinerPlanNode.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.plan;
+
+import java.util.List;
+
+import org.apache.flink.optimizer.CompilerException;
+import org.apache.flink.optimizer.costs.Costs;
+import org.apache.flink.optimizer.dag.SinkJoiner;
+import org.apache.flink.runtime.operators.DriverStrategy;
+
+
+/**
+ *
+ */
+public class SinkJoinerPlanNode extends DualInputPlanNode {
+	
+	public SinkJoinerPlanNode(SinkJoiner template, Channel input1, Channel input2) {
+		super(template, "", input1, input2, DriverStrategy.BINARY_NO_OP);
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	public void setCosts(Costs nodeCosts) {
+		// the plan enumeration logic works as for regular two-input-operators, which is important
+		// because of the branch handling logic. it does pick redistributing network channels
+		// between the sink and the sink joiner, because sinks joiner has a different DOP than the sink.
+		// we discard any cost and simply use the sum of the costs from the two children.
+		
+		Costs totalCosts = getInput1().getSource().getCumulativeCosts().clone();
+		totalCosts.addCosts(getInput2().getSource().getCumulativeCosts());
+		super.setCosts(totalCosts);
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	public void getDataSinks(List<SinkPlanNode> sinks) {
+		final PlanNode in1 = this.input1.getSource();
+		final PlanNode in2 = this.input2.getSource();
+		
+		if (in1 instanceof SinkPlanNode) {
+			sinks.add((SinkPlanNode) in1);
+		} else if (in1 instanceof SinkJoinerPlanNode) {
+			((SinkJoinerPlanNode) in1).getDataSinks(sinks);
+		} else {
+			throw new CompilerException("Illegal child node for a sink joiner utility node: Neither Sink nor Sink Joiner");
+		}
+		
+		if (in2 instanceof SinkPlanNode) {
+			sinks.add((SinkPlanNode) in2);
+		} else if (in2 instanceof SinkJoinerPlanNode) {
+			((SinkJoinerPlanNode) in2).getDataSinks(sinks);
+		} else {
+			throw new CompilerException("Illegal child node for a sink joiner utility node: Neither Sink nor Sink Joiner");
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SinkPlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SinkPlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SinkPlanNode.java
new file mode 100644
index 0000000..656e67f
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SinkPlanNode.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.plan;
+
+import org.apache.flink.optimizer.dag.DataSinkNode;
+import org.apache.flink.runtime.operators.DriverStrategy;
+
+/**
+ * Plan candidate node for data flow sinks.
+ */
+public class SinkPlanNode extends SingleInputPlanNode
+{
+	/**
+	 * Constructs a new sink candidate node that uses <i>NONE</i> as its local strategy. Note that
+	 * local sorting and range partitioning are handled by the incoming channel already.
+	 * 
+	 * @param template The template optimizer node that this candidate is created for.
+	 */
+	public SinkPlanNode(DataSinkNode template, String nodeName, Channel input) {
+		super(template, nodeName, input, DriverStrategy.NONE);
+		
+		this.globalProps = input.getGlobalProperties().clone();
+		this.localProps = input.getLocalProperties().clone();
+	}
+	
+	public DataSinkNode getSinkNode() {
+		if (this.template instanceof DataSinkNode) {
+			return (DataSinkNode) this.template;
+		} else {
+			throw new RuntimeException();
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SolutionSetPlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SolutionSetPlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SolutionSetPlanNode.java
new file mode 100644
index 0000000..63093dd
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SolutionSetPlanNode.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.plan;
+
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE_AND_DAM;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.NOT_FOUND;
+
+import java.util.Collections;
+import java.util.HashMap;
+
+import org.apache.flink.optimizer.costs.Costs;
+import org.apache.flink.optimizer.dag.OptimizerNode;
+import org.apache.flink.optimizer.dag.SolutionSetNode;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.runtime.operators.DriverStrategy;
+import org.apache.flink.util.Visitor;
+
+/**
+ * Plan candidate node for partial solution of a bulk iteration.
+ */
+public class SolutionSetPlanNode extends PlanNode {
+	
+	private static final Costs NO_COSTS = new Costs();
+	
+	private WorksetIterationPlanNode containingIterationNode;
+	
+	private final Channel initialInput;
+	
+	public Object postPassHelper;
+	
+	
+	public SolutionSetPlanNode(SolutionSetNode template, String nodeName,
+			GlobalProperties gProps, LocalProperties lProps,
+			Channel initialInput)
+	{
+		super(template, nodeName, DriverStrategy.NONE);
+		
+		this.globalProps = gProps;
+		this.localProps = lProps;
+		this.initialInput = initialInput;
+		
+		// the node incurs no cost
+		this.nodeCosts = NO_COSTS;
+		this.cumulativeCosts = NO_COSTS;
+		
+		if (initialInput.getSource().branchPlan != null && initialInput.getSource().branchPlan.size() > 0) {
+			if (this.branchPlan == null) {
+				this.branchPlan = new HashMap<OptimizerNode, PlanNode>();
+			}
+			
+			this.branchPlan.putAll(initialInput.getSource().branchPlan);
+		}
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	public SolutionSetNode getSolutionSetNode() {
+		return (SolutionSetNode) this.template;
+	}
+	
+	public WorksetIterationPlanNode getContainingIterationNode() {
+		return this.containingIterationNode;
+	}
+	
+	public void setContainingIterationNode(WorksetIterationPlanNode containingIterationNode) {
+		this.containingIterationNode = containingIterationNode;
+	}
+
+	// --------------------------------------------------------------------------------------------
+	
+
+	@Override
+	public void accept(Visitor<PlanNode> visitor) {
+		if (visitor.preVisit(this)) {
+			visitor.postVisit(this);
+		}
+	}
+
+
+	@Override
+	public Iterable<PlanNode> getPredecessors() {
+		return Collections.<PlanNode>emptyList();
+	}
+
+
+	@Override
+	public Iterable<Channel> getInputs() {
+		return Collections.<Channel>emptyList();
+	}
+
+
+	@Override
+	public SourceAndDamReport hasDamOnPathDownTo(PlanNode source) {
+		if (source == this) {
+			return FOUND_SOURCE_AND_DAM;
+		}
+		
+		SourceAndDamReport res = this.initialInput.getSource().hasDamOnPathDownTo(source);
+		if (res == FOUND_SOURCE_AND_DAM || res == FOUND_SOURCE) {
+			return FOUND_SOURCE_AND_DAM;
+		} else {
+			return NOT_FOUND;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SourcePlanNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SourcePlanNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SourcePlanNode.java
new file mode 100644
index 0000000..11b7cc9
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/SourcePlanNode.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.plan;
+
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.NOT_FOUND;
+
+import java.util.Collections;
+
+import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
+import org.apache.flink.optimizer.dag.DataSourceNode;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.runtime.operators.DriverStrategy;
+import org.apache.flink.util.Visitor;
+
+/**
+ * Plan candidate node for data flow sources that have no input and no special strategies.
+ */
+public class SourcePlanNode extends PlanNode {
+	
+	private TypeSerializerFactory<?> serializer;
+	
+	/**
+	 * Constructs a new source candidate node that uses <i>NONE</i> as its local strategy.
+	 * 
+	 * @param template The template optimizer node that this candidate is created for.
+	 */
+	public SourcePlanNode(DataSourceNode template, String nodeName) {
+		this(template, nodeName, new GlobalProperties(), new LocalProperties());
+	}
+
+	public SourcePlanNode(DataSourceNode template, String nodeName, GlobalProperties gprops, LocalProperties lprops) {
+		super(template, nodeName, DriverStrategy.NONE);
+
+		this.globalProps = gprops;
+		this.localProps = lprops;
+		updatePropertiesWithUniqueSets(template.getUniqueFields());
+	}
+
+	// --------------------------------------------------------------------------------------------
+	
+	public DataSourceNode getDataSourceNode() {
+		return (DataSourceNode) this.template;
+	}
+	
+	/**
+	 * Gets the serializer from this PlanNode.
+	 *
+	 * @return The serializer.
+	 */
+	public TypeSerializerFactory<?> getSerializer() {
+		return serializer;
+	}
+	
+	/**
+	 * Sets the serializer for this PlanNode.
+	 *
+	 * @param serializer The serializer to set.
+	 */
+	public void setSerializer(TypeSerializerFactory<?> serializer) {
+		this.serializer = serializer;
+	}
+
+	// --------------------------------------------------------------------------------------------
+	
+
+	@Override
+	public void accept(Visitor<PlanNode> visitor) {
+		if (visitor.preVisit(this)) {
+			visitor.postVisit(this);
+		}
+	}
+
+
+	@Override
+	public Iterable<PlanNode> getPredecessors() {
+		return Collections.<PlanNode>emptyList();
+	}
+
+
+	@Override
+	public Iterable<Channel> getInputs() {
+		return Collections.<Channel>emptyList();
+	}
+
+
+	@Override
+	public SourceAndDamReport hasDamOnPathDownTo(PlanNode source) {
+		if (source == this) {
+			return FOUND_SOURCE;
+		} else {
+			return NOT_FOUND;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/StreamingPlan.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/StreamingPlan.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/StreamingPlan.java
new file mode 100644
index 0000000..880f2e3
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/StreamingPlan.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.plan;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.flink.runtime.jobgraph.JobGraph;
+
+/**
+ * Abstract class representing Flink Streaming plans
+ * 
+ */
+public abstract class StreamingPlan implements FlinkPlan {
+
+	public abstract JobGraph getJobGraph();
+
+	public abstract String getStreamingPlanAsJSON();
+
+	public abstract void dumpStreamingPlanAsJSON(File file) throws IOException;
+
+}