[03/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/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeForwardTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeForwardTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeForwardTest.java
new file mode 100644
index 0000000..5175d8c
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeForwardTest.java
@@ -0,0 +1,139 @@
+/*
+ * 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.dataexchange;
+
+import org.apache.flink.api.common.ExecutionMode;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.optimizer.CompilerTestBase;
+import org.apache.flink.optimizer.plan.OptimizedPlan;
+import org.apache.flink.optimizer.plan.SingleInputPlanNode;
+import org.apache.flink.optimizer.plan.SinkPlanNode;
+import org.apache.flink.optimizer.testfunctions.IdentityKeyExtractor;
+import org.apache.flink.optimizer.testfunctions.Top1GroupReducer;
+import org.apache.flink.runtime.io.network.DataExchangeMode;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * This test verifies that the optimizer assigns the correct
+ * data exchange mode to a simple forward / shuffle plan.
+ *
+ * <pre>
+ *     (source) -> (map) -> (filter) -> (groupBy / reduce)
+ * </pre>
+ */
+@SuppressWarnings("serial")
+public class DataExchangeModeForwardTest extends CompilerTestBase {
+
+
+	@Test
+	public void testPipelinedForced() {
+		// PIPELINED_FORCED should result in pipelining all the way
+		verifySimpleForwardPlan(ExecutionMode.PIPELINED_FORCED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED);
+	}
+
+	@Test
+	public void testPipelined() {
+		// PIPELINED should result in pipelining all the way
+		verifySimpleForwardPlan(ExecutionMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED);
+	}
+
+	@Test
+	public void testBatch() {
+		// BATCH should result in batching the shuffle all the way
+		verifySimpleForwardPlan(ExecutionMode.BATCH,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.BATCH, DataExchangeMode.PIPELINED);
+	}
+
+	@Test
+	public void testBatchForced() {
+		// BATCH_FORCED should result in batching all the way
+		verifySimpleForwardPlan(ExecutionMode.BATCH_FORCED,
+				DataExchangeMode.BATCH, DataExchangeMode.BATCH,
+				DataExchangeMode.BATCH, DataExchangeMode.PIPELINED,
+				DataExchangeMode.BATCH, DataExchangeMode.BATCH);
+	}
+
+	private void verifySimpleForwardPlan(ExecutionMode execMode,
+										DataExchangeMode toMap,
+										DataExchangeMode toFilter,
+										DataExchangeMode toKeyExtractor,
+										DataExchangeMode toCombiner,
+										DataExchangeMode toReduce,
+										DataExchangeMode toSink)
+	{
+		try {
+			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+			env.getConfig().setExecutionMode(execMode);
+
+			DataSet<String> dataSet = env.readTextFile("/never/accessed");
+			dataSet
+				.map(new MapFunction<String, Integer>() {
+					@Override
+					public Integer map(String value) {
+						return 0;
+					}
+				})
+				.filter(new FilterFunction<Integer>() {
+					@Override
+					public boolean filter(Integer value) {
+						return false;
+					}
+				})
+				.groupBy(new IdentityKeyExtractor<Integer>())
+				.reduceGroup(new Top1GroupReducer<Integer>())
+				.output(new DiscardingOutputFormat<Integer>());
+
+			OptimizedPlan optPlan = compileNoStats(env.createProgramPlan());
+			SinkPlanNode sinkNode = optPlan.getDataSinks().iterator().next();
+
+			SingleInputPlanNode reduceNode = (SingleInputPlanNode) sinkNode.getPredecessor();
+			SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getPredecessor();
+			SingleInputPlanNode keyExtractorNode = (SingleInputPlanNode) combineNode.getPredecessor();
+
+			SingleInputPlanNode filterNode = (SingleInputPlanNode) keyExtractorNode.getPredecessor();
+			SingleInputPlanNode mapNode = (SingleInputPlanNode) filterNode.getPredecessor();
+
+			assertEquals(toMap, mapNode.getInput().getDataExchangeMode());
+			assertEquals(toFilter, filterNode.getInput().getDataExchangeMode());
+			assertEquals(toKeyExtractor, keyExtractorNode.getInput().getDataExchangeMode());
+			assertEquals(toCombiner, combineNode.getInput().getDataExchangeMode());
+			assertEquals(toReduce, reduceNode.getInput().getDataExchangeMode());
+			assertEquals(toSink, sinkNode.getInput().getDataExchangeMode());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java
new file mode 100644
index 0000000..6b2691a
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/DataExchangeModeOpenBranchingTest.java
@@ -0,0 +1,182 @@
+/*
+ * 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.dataexchange;
+
+import org.apache.flink.api.common.ExecutionMode;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.optimizer.CompilerTestBase;
+import org.apache.flink.optimizer.plan.DualInputPlanNode;
+import org.apache.flink.optimizer.plan.OptimizedPlan;
+import org.apache.flink.optimizer.plan.SingleInputPlanNode;
+import org.apache.flink.optimizer.plan.SinkPlanNode;
+import org.apache.flink.runtime.io.network.DataExchangeMode;
+import org.junit.Test;
+
+import java.util.Collection;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * This test checks the correct assignment of the DataExchangeMode to
+ * connections for programs that branch, but do not re-join the branches.
+ *
+ * <pre>
+ *                      /---> (filter) -> (sink)
+ *                     /
+ *                    /
+ * (source) -> (map) -----------------\
+ *                    \               (join) -> (sink)
+ *                     \   (source) --/
+ *                      \
+ *                       \
+ *                        \-> (sink)
+ * </pre>
+ */
+@SuppressWarnings({"serial", "unchecked"})
+public class DataExchangeModeOpenBranchingTest extends CompilerTestBase {
+
+	@Test
+	public void testPipelinedForced() {
+		// PIPELINED_FORCED should result in pipelining all the way
+		verifyBranchigPlan(ExecutionMode.PIPELINED_FORCED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED);
+	}
+
+	@Test
+	public void testPipelined() {
+		// PIPELINED should result in pipelining all the way
+		verifyBranchigPlan(ExecutionMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED);
+	}
+
+	@Test
+	public void testBatch() {
+		// BATCH should result in batching the shuffle all the way
+		verifyBranchigPlan(ExecutionMode.BATCH,
+				DataExchangeMode.PIPELINED, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED, DataExchangeMode.BATCH,
+				DataExchangeMode.BATCH, DataExchangeMode.PIPELINED,
+				DataExchangeMode.PIPELINED);
+	}
+
+	@Test
+	public void testBatchForced() {
+		// BATCH_FORCED should result in batching all the way
+		verifyBranchigPlan(ExecutionMode.BATCH_FORCED,
+				DataExchangeMode.BATCH, DataExchangeMode.BATCH,
+				DataExchangeMode.BATCH, DataExchangeMode.BATCH,
+				DataExchangeMode.BATCH, DataExchangeMode.BATCH,
+				DataExchangeMode.BATCH);
+	}
+
+	private void verifyBranchigPlan(ExecutionMode execMode,
+									DataExchangeMode toMap,
+									DataExchangeMode toFilter,
+									DataExchangeMode toFilterSink,
+									DataExchangeMode toJoin1,
+									DataExchangeMode toJoin2,
+									DataExchangeMode toJoinSink,
+									DataExchangeMode toDirectSink)
+	{
+		try {
+			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+			env.getConfig().setExecutionMode(execMode);
+
+			DataSet<Tuple2<Long, Long>> data = env.generateSequence(1, 100000)
+					.map(new MapFunction<Long, Tuple2<Long, Long>>() {
+						@Override
+						public Tuple2<Long, Long> map(Long value) {
+							return new Tuple2<Long, Long>(value, value);
+						}
+					});
+
+			// output 1
+			data
+					.filter(new FilterFunction<Tuple2<Long, Long>>() {
+						@Override
+						public boolean filter(Tuple2<Long, Long> value) {
+							return false;
+						}
+					})
+					.output(new DiscardingOutputFormat<Tuple2<Long, Long>>()).name("sink1");
+
+			// output 2 does a join before a join
+			data
+					.join(env.fromElements(new Tuple2<Long, Long>(1L, 2L)))
+					.where(1)
+					.equalTo(0)
+					.output(new DiscardingOutputFormat<Tuple2<Tuple2<Long, Long>, Tuple2<Long, Long>>>()).name("sink2");
+
+			// output 3 is direct
+			data
+					.output(new DiscardingOutputFormat<Tuple2<Long, Long>>()).name("sink3");
+
+			OptimizedPlan optPlan = compileNoStats(env.createProgramPlan());
+
+			SinkPlanNode filterSink = findSink(optPlan.getDataSinks(), "sink1");
+			SinkPlanNode joinSink = findSink(optPlan.getDataSinks(), "sink2");
+			SinkPlanNode directSink = findSink(optPlan.getDataSinks(), "sink3");
+
+			SingleInputPlanNode filterNode = (SingleInputPlanNode) filterSink.getPredecessor();
+			SingleInputPlanNode mapNode = (SingleInputPlanNode) filterNode.getPredecessor();
+
+			DualInputPlanNode joinNode = (DualInputPlanNode) joinSink.getPredecessor();
+			assertEquals(mapNode, joinNode.getInput1().getSource());
+
+			assertEquals(mapNode, directSink.getPredecessor());
+
+			assertEquals(toFilterSink, filterSink.getInput().getDataExchangeMode());
+			assertEquals(toJoinSink, joinSink.getInput().getDataExchangeMode());
+			assertEquals(toDirectSink, directSink.getInput().getDataExchangeMode());
+
+			assertEquals(toMap, mapNode.getInput().getDataExchangeMode());
+			assertEquals(toFilter, filterNode.getInput().getDataExchangeMode());
+
+			assertEquals(toJoin1, joinNode.getInput1().getDataExchangeMode());
+			assertEquals(toJoin2, joinNode.getInput2().getDataExchangeMode());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+
+	private SinkPlanNode findSink(Collection<SinkPlanNode> collection, String name) {
+		for (SinkPlanNode node : collection) {
+			String nodeName = node.getOptimizerNode().getOperator().getName();
+			if (nodeName != null && nodeName.equals(name)) {
+				return node;
+			}
+		}
+
+		throw new IllegalArgumentException("No node with that name was found.");
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java
new file mode 100644
index 0000000..1a14be5
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataexchange/PipelineBreakingTest.java
@@ -0,0 +1,322 @@
+/*
+ * 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.dataexchange;
+
+import org.apache.flink.api.common.Plan;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.optimizer.dag.DataSinkNode;
+import org.apache.flink.optimizer.dag.OptimizerNode;
+import org.apache.flink.optimizer.dag.SingleInputNode;
+import org.apache.flink.optimizer.dag.SinkJoiner;
+import org.apache.flink.optimizer.dag.TwoInputNode;
+import org.apache.flink.optimizer.testfunctions.DummyCoGroupFunction;
+import org.apache.flink.optimizer.testfunctions.DummyFlatJoinFunction;
+import org.apache.flink.optimizer.testfunctions.IdentityFlatMapper;
+import org.apache.flink.optimizer.testfunctions.IdentityKeyExtractor;
+import org.apache.flink.optimizer.testfunctions.SelectOneReducer;
+import org.apache.flink.optimizer.testfunctions.Top1GroupReducer;
+import org.apache.flink.optimizer.traversals.BranchesVisitor;
+import org.apache.flink.optimizer.traversals.GraphCreatingVisitor;
+import org.apache.flink.optimizer.traversals.IdAndEstimatesVisitor;
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.List;
+
+import static org.junit.Assert.*;
+
+/**
+ * This test checks whether connections are correctly marked as pipelined breaking.
+ */
+@SuppressWarnings("serial")
+public class PipelineBreakingTest {
+
+	/**
+	 * Tests that no pipeline breakers are inserted into a simple forward
+	 * pipeline.
+	 *
+	 * <pre>
+	 *     (source) -> (map) -> (filter) -> (groupBy / reduce)
+	 * </pre>
+	 */
+	@Test
+	public void testSimpleForwardPlan() {
+		try {
+			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+			DataSet<String> dataSet = env.readTextFile("/never/accessed");
+			dataSet
+				.map(new MapFunction<String, Integer>() {
+					@Override
+					public Integer map(String value) {
+						return 0;
+					}
+				})
+				.filter(new FilterFunction<Integer>() {
+					@Override
+					public boolean filter(Integer value) {
+						return false;
+					}
+				})
+				.groupBy(new IdentityKeyExtractor<Integer>())
+				.reduceGroup(new Top1GroupReducer<Integer>())
+				.output(new DiscardingOutputFormat<Integer>());
+
+			DataSinkNode sinkNode = convertPlan(env.createProgramPlan()).get(0);
+
+			SingleInputNode reduceNode = (SingleInputNode) sinkNode.getPredecessorNode();
+			SingleInputNode keyExtractorNode = (SingleInputNode) reduceNode.getPredecessorNode();
+
+			SingleInputNode filterNode = (SingleInputNode) keyExtractorNode.getPredecessorNode();
+			SingleInputNode mapNode = (SingleInputNode) filterNode.getPredecessorNode();
+
+			assertFalse(sinkNode.getInputConnection().isBreakingPipeline());
+			assertFalse(reduceNode.getIncomingConnection().isBreakingPipeline());
+			assertFalse(keyExtractorNode.getIncomingConnection().isBreakingPipeline());
+			assertFalse(filterNode.getIncomingConnection().isBreakingPipeline());
+			assertFalse(mapNode.getIncomingConnection().isBreakingPipeline());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+
+	/**
+	 * Tests that branching plans, where the branches are not re-joined,
+	 * do not place pipeline breakers.
+	 * 
+	 * <pre>
+	 *                      /---> (filter) -> (sink)
+	 *                     /
+	 *                    /
+	 * (source) -> (map) -----------------\
+	 *                    \               (join) -> (sink)
+	 *                     \   (source) --/
+	 *                      \
+	 *                       \
+	 *                        \-> (sink)
+	 * </pre>
+	 */
+	@Test
+	public void testBranchingPlanNotReJoined() {
+		try {
+			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+			DataSet<Integer> data = env.readTextFile("/never/accessed")
+				.map(new MapFunction<String, Integer>() {
+					@Override
+					public Integer map(String value) {
+						return 0;
+					}
+				});
+
+			// output 1
+			data
+				.filter(new FilterFunction<Integer>() {
+					@Override
+					public boolean filter(Integer value) {
+						return false;
+					}
+				})
+				.output(new DiscardingOutputFormat<Integer>());
+
+			// output 2 does a join before a join
+			data
+				.join(env.fromElements(1, 2, 3, 4))
+					.where(new IdentityKeyExtractor<Integer>())
+					.equalTo(new IdentityKeyExtractor<Integer>())
+				.output(new DiscardingOutputFormat<Tuple2<Integer, Integer>>());
+
+			// output 3 is direct
+			data
+				.output(new DiscardingOutputFormat<Integer>());
+
+			List<DataSinkNode> sinks = convertPlan(env.createProgramPlan());
+
+			// gather the optimizer DAG nodes
+
+			DataSinkNode sinkAfterFilter = sinks.get(0);
+			DataSinkNode sinkAfterJoin = sinks.get(1);
+			DataSinkNode sinkDirect = sinks.get(2);
+
+			SingleInputNode filterNode = (SingleInputNode) sinkAfterFilter.getPredecessorNode();
+			SingleInputNode mapNode = (SingleInputNode) filterNode.getPredecessorNode();
+
+			TwoInputNode joinNode = (TwoInputNode) sinkAfterJoin.getPredecessorNode();
+			SingleInputNode joinInput = (SingleInputNode) joinNode.getSecondPredecessorNode();
+
+			// verify the non-pipeline breaking status
+
+			assertFalse(sinkAfterFilter.getInputConnection().isBreakingPipeline());
+			assertFalse(sinkAfterJoin.getInputConnection().isBreakingPipeline());
+			assertFalse(sinkDirect.getInputConnection().isBreakingPipeline());
+
+			assertFalse(filterNode.getIncomingConnection().isBreakingPipeline());
+			assertFalse(mapNode.getIncomingConnection().isBreakingPipeline());
+
+			assertFalse(joinNode.getFirstIncomingConnection().isBreakingPipeline());
+			assertFalse(joinNode.getSecondIncomingConnection().isBreakingPipeline());
+			assertFalse(joinInput.getIncomingConnection().isBreakingPipeline());
+
+			// some other sanity checks on the plan construction (cannot hurt)
+
+			assertEquals(mapNode, ((SingleInputNode) joinNode.getFirstPredecessorNode()).getPredecessorNode());
+			assertEquals(mapNode, sinkDirect.getPredecessorNode());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+
+	/**
+	 * Tests that branches that are re-joined have place pipeline breakers.
+	 * 
+	 * <pre>
+	 *                                         /-> (sink)
+	 *                                        /
+	 *                         /-> (reduce) -+          /-> (flatmap) -> (sink)
+	 *                        /               \        /
+	 *     (source) -> (map) -                (join) -+-----\
+	 *                        \               /              \
+	 *                         \-> (filter) -+                \
+	 *                                       \                (co group) -> (sink)
+	 *                                        \                /
+	 *                                         \-> (reduce) - /
+	 * </pre>
+	 */
+	@Test
+	public void testReJoinedBranches() {
+		try {
+			// build a test program
+
+			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+			DataSet<Tuple2<Long, Long>> data = env.fromElements(33L, 44L)
+					.map(new MapFunction<Long, Tuple2<Long, Long>>() {
+						@Override
+						public Tuple2<Long, Long> map(Long value) {
+							return new Tuple2<Long, Long>(value, value);
+						}
+					});
+
+			DataSet<Tuple2<Long, Long>> reduced = data.groupBy(0).reduce(new SelectOneReducer<Tuple2<Long, Long>>());
+			reduced.output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
+			
+			DataSet<Tuple2<Long, Long>> filtered = data.filter(new FilterFunction<Tuple2<Long, Long>>() {
+				@Override
+				public boolean filter(Tuple2<Long, Long> value) throws Exception {
+					return false;
+				}
+			});
+			
+			DataSet<Tuple2<Long, Long>> joined = reduced.join(filtered)
+					.where(1).equalTo(1)
+					.with(new DummyFlatJoinFunction<Tuple2<Long, Long>>());
+			
+			joined.flatMap(new IdentityFlatMapper<Tuple2<Long, Long>>())
+					.output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
+
+			joined.coGroup(filtered.groupBy(1).reduceGroup(new Top1GroupReducer<Tuple2<Long, Long>>()))
+					.where(0).equalTo(0)
+					.with(new DummyCoGroupFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>())
+					.output(new DiscardingOutputFormat<Tuple2<Tuple2<Long, Long>, Tuple2<Long, Long>>>());
+
+			List<DataSinkNode> sinks = convertPlan(env.createProgramPlan());
+
+			// gather the optimizer DAG nodes
+
+			DataSinkNode sinkAfterReduce = sinks.get(0);
+			DataSinkNode sinkAfterFlatMap = sinks.get(1);
+			DataSinkNode sinkAfterCoGroup = sinks.get(2);
+
+			SingleInputNode reduceNode = (SingleInputNode) sinkAfterReduce.getPredecessorNode();
+			SingleInputNode mapNode = (SingleInputNode) reduceNode.getPredecessorNode();
+
+			SingleInputNode flatMapNode = (SingleInputNode) sinkAfterFlatMap.getPredecessorNode();
+			TwoInputNode joinNode = (TwoInputNode) flatMapNode.getPredecessorNode();
+			SingleInputNode filterNode = (SingleInputNode) joinNode.getSecondPredecessorNode();
+
+			TwoInputNode coGroupNode = (TwoInputNode) sinkAfterCoGroup.getPredecessorNode();
+			SingleInputNode otherReduceNode = (SingleInputNode) coGroupNode.getSecondPredecessorNode();
+
+			// test sanity checks (that we constructed the DAG correctly)
+
+			assertEquals(reduceNode, joinNode.getFirstPredecessorNode());
+			assertEquals(mapNode, filterNode.getPredecessorNode());
+			assertEquals(joinNode, coGroupNode.getFirstPredecessorNode());
+			assertEquals(filterNode, otherReduceNode.getPredecessorNode());
+
+			// verify the pipeline breaking status
+
+			assertFalse(sinkAfterReduce.getInputConnection().isBreakingPipeline());
+			assertFalse(sinkAfterFlatMap.getInputConnection().isBreakingPipeline());
+			assertFalse(sinkAfterCoGroup.getInputConnection().isBreakingPipeline());
+
+			assertFalse(mapNode.getIncomingConnection().isBreakingPipeline());
+			assertFalse(flatMapNode.getIncomingConnection().isBreakingPipeline());
+			assertFalse(joinNode.getFirstIncomingConnection().isBreakingPipeline());
+			assertFalse(coGroupNode.getFirstIncomingConnection().isBreakingPipeline());
+			assertFalse(coGroupNode.getSecondIncomingConnection().isBreakingPipeline());
+
+			// these should be pipeline breakers
+			assertTrue(reduceNode.getIncomingConnection().isBreakingPipeline());
+			assertTrue(filterNode.getIncomingConnection().isBreakingPipeline());
+			assertTrue(otherReduceNode.getIncomingConnection().isBreakingPipeline());
+			assertTrue(joinNode.getSecondIncomingConnection().isBreakingPipeline());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+
+	private static List<DataSinkNode> convertPlan(Plan p) {
+		GraphCreatingVisitor dagCreator =
+				new GraphCreatingVisitor(17, p.getExecutionConfig().getExecutionMode());
+
+		// create the DAG
+		p.accept(dagCreator);
+		List<DataSinkNode> sinks = dagCreator.getSinks();
+
+		// build a single root and run the branch tracking logic
+		OptimizerNode rootNode;
+		if (sinks.size() == 1) {
+			rootNode = sinks.get(0);
+		}
+		else {
+			Iterator<DataSinkNode> iter = sinks.iterator();
+			rootNode = iter.next();
+
+			while (iter.hasNext()) {
+				rootNode = new SinkJoiner(rootNode, iter.next());
+			}
+		}
+		rootNode.accept(new IdAndEstimatesVisitor(null));
+		rootNode.accept(new BranchesVisitor());
+
+		return sinks;
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesFilteringTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesFilteringTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesFilteringTest.java
new file mode 100644
index 0000000..3e32905
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesFilteringTest.java
@@ -0,0 +1,428 @@
+/*
+ * 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.dataproperties;
+
+import static org.junit.Assert.*;
+
+import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.api.common.operators.Order;
+import org.apache.flink.api.common.operators.Ordering;
+import org.apache.flink.api.common.operators.SingleInputSemanticProperties;
+import org.apache.flink.api.common.operators.util.FieldList;
+import org.apache.flink.api.common.operators.util.FieldSet;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.functions.SemanticPropUtil;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple8;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.StringValue;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Set;
+
+public class GlobalPropertiesFilteringTest {
+
+	private TupleTypeInfo<Tuple8<Integer, Integer, Integer, Integer, Integer, Integer, Integer, Integer>> tupleInfo =
+			new TupleTypeInfo<Tuple8<Integer, Integer, Integer, Integer, Integer, Integer, Integer, Integer>>(
+					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO,
+					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO
+			);
+
+	@Test
+	public void testAllErased1() {
+
+		SingleInputSemanticProperties semProps = new SingleInputSemanticProperties();
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setHashPartitioned(new FieldList(0, 1));
+		gprops.addUniqueFieldCombination(new FieldSet(3, 4));
+		gprops.addUniqueFieldCombination(new FieldSet(5, 6));
+
+		GlobalProperties result = gprops.filterBySemanticProperties(semProps, 0);
+
+		assertEquals(PartitioningProperty.RANDOM_PARTITIONED, result.getPartitioning());
+		assertNull(result.getPartitioningFields());
+		assertNull(result.getPartitioningOrdering());
+		assertNull(result.getUniqueFieldCombination());
+	}
+
+	@Test
+	public void testAllErased2() {
+
+		SingleInputSemanticProperties semProps = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(semProps, new String[]{"2"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setHashPartitioned(new FieldList(0, 1));
+		gprops.addUniqueFieldCombination(new FieldSet(3, 4));
+		gprops.addUniqueFieldCombination(new FieldSet(5, 6));
+
+		GlobalProperties result = gprops.filterBySemanticProperties(semProps, 0);
+
+		assertEquals(PartitioningProperty.RANDOM_PARTITIONED, result.getPartitioning());
+		assertNull(result.getPartitioningFields());
+		assertNull(result.getPartitioningOrdering());
+		assertNull(result.getUniqueFieldCombination());
+	}
+
+	@Test
+	public void testHashPartitioningPreserved1() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1;4"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setHashPartitioned(new FieldList(0, 1, 4));
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.HASH_PARTITIONED, result.getPartitioning());
+		FieldList pFields = result.getPartitioningFields();
+		assertEquals(3, pFields.size());
+		assertTrue(pFields.contains(0));
+		assertTrue(pFields.contains(1));
+		assertTrue(pFields.contains(4));
+	}
+
+	@Test
+	public void testHashPartitioningPreserved2() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0->1; 1->2; 4->3"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setHashPartitioned(new FieldList(0, 1, 4));
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.HASH_PARTITIONED, result.getPartitioning());
+		FieldList pFields = result.getPartitioningFields();
+		assertEquals(3, pFields.size());
+		assertTrue(pFields.contains(1));
+		assertTrue(pFields.contains(2));
+		assertTrue(pFields.contains(3));
+	}
+
+	@Test
+	public void testHashPartitioningErased() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setHashPartitioned(new FieldList(0, 1, 4));
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.RANDOM_PARTITIONED, result.getPartitioning());
+		assertNull(result.getPartitioningFields());
+	}
+
+	@Test
+	public void testAnyPartitioningPreserved1() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1;4"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setAnyPartitioning(new FieldList(0, 1, 4));
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.ANY_PARTITIONING, result.getPartitioning());
+		FieldList pFields = result.getPartitioningFields();
+		assertEquals(3, pFields.size());
+		assertTrue(pFields.contains(0));
+		assertTrue(pFields.contains(1));
+		assertTrue(pFields.contains(4));
+	}
+
+	@Test
+	public void testAnyPartitioningPreserved2() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0->1; 1->2; 4->3"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setAnyPartitioning(new FieldList(0, 1, 4));
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.ANY_PARTITIONING, result.getPartitioning());
+		FieldList pFields = result.getPartitioningFields();
+		assertEquals(3, pFields.size());
+		assertTrue(pFields.contains(1));
+		assertTrue(pFields.contains(2));
+		assertTrue(pFields.contains(3));
+	}
+
+	@Test
+	public void testAnyPartitioningErased() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setAnyPartitioning(new FieldList(0, 1, 4));
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.RANDOM_PARTITIONED, result.getPartitioning());
+		assertNull(result.getPartitioningFields());
+	}
+
+	@Test
+	public void testCustomPartitioningPreserved1() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1;4"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		Partitioner<Tuple2<Long, Integer>> myP = new MockPartitioner();
+		gprops.setCustomPartitioned(new FieldList(0, 4), myP);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.CUSTOM_PARTITIONING, result.getPartitioning());
+		FieldList pFields = result.getPartitioningFields();
+		assertEquals(2, pFields.size());
+		assertTrue(pFields.contains(0));
+		assertTrue(pFields.contains(4));
+		assertEquals(myP, result.getCustomPartitioner());
+	}
+
+	@Test
+	public void testCustomPartitioningPreserved2() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0->1; 1->2; 4->3"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		Partitioner<Tuple2<Long, Integer>> myP = new MockPartitioner();
+		gprops.setCustomPartitioned(new FieldList(0, 4), myP);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.CUSTOM_PARTITIONING, result.getPartitioning());
+		FieldList pFields = result.getPartitioningFields();
+		assertEquals(2, pFields.size());
+		assertTrue(pFields.contains(1));
+		assertTrue(pFields.contains(3));
+		assertEquals(myP, result.getCustomPartitioner());
+	}
+
+	@Test
+	public void testCustomPartitioningErased() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		Partitioner<Tuple2<Long, Integer>> myP = new MockPartitioner();
+		gprops.setCustomPartitioned(new FieldList(0, 4), myP);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.RANDOM_PARTITIONED, result.getPartitioning());
+		assertNull(result.getPartitioningFields());
+		assertNull(result.getCustomPartitioner());
+	}
+
+	@Test
+	public void testRangePartitioningPreserved1() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"1;2;5"}, null, null, tupleInfo, tupleInfo);
+
+		Ordering o = new Ordering();
+		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
+		o.appendOrdering(5, LongValue.class, Order.DESCENDING);
+		o.appendOrdering(2, StringValue.class, Order.ASCENDING);
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setRangePartitioned(o);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.RANGE_PARTITIONED, result.getPartitioning());
+		FieldList pFields = result.getPartitioningFields();
+		assertEquals(3, pFields.size());
+		assertEquals(1, pFields.get(0).intValue());
+		assertEquals(5, pFields.get(1).intValue());
+		assertEquals(2, pFields.get(2).intValue());
+		Ordering pOrder = result.getPartitioningOrdering();
+		assertEquals(3, pOrder.getNumberOfFields());
+		assertEquals(1, pOrder.getFieldNumber(0).intValue());
+		assertEquals(5, pOrder.getFieldNumber(1).intValue());
+		assertEquals(2, pOrder.getFieldNumber(2).intValue());
+		assertEquals(Order.ASCENDING, pOrder.getOrder(0));
+		assertEquals(Order.DESCENDING, pOrder.getOrder(1));
+		assertEquals(Order.ASCENDING, pOrder.getOrder(2));
+		assertEquals(IntValue.class, pOrder.getType(0));
+		assertEquals(LongValue.class, pOrder.getType(1));
+		assertEquals(StringValue.class, pOrder.getType(2));
+	}
+
+	@Test
+	public void testRangePartitioningPreserved2() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"1->3; 2->0; 5->1"}, null, null, tupleInfo, tupleInfo);
+
+		Ordering o = new Ordering();
+		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
+		o.appendOrdering(5, LongValue.class, Order.DESCENDING);
+		o.appendOrdering(2, StringValue.class, Order.ASCENDING);
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setRangePartitioned(o);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.RANGE_PARTITIONED, result.getPartitioning());
+		FieldList pFields = result.getPartitioningFields();
+		assertEquals(3, pFields.size());
+		assertEquals(3, pFields.get(0).intValue());
+		assertEquals(1, pFields.get(1).intValue());
+		assertEquals(0, pFields.get(2).intValue());
+		Ordering pOrder = result.getPartitioningOrdering();
+		assertEquals(3, pOrder.getNumberOfFields());
+		assertEquals(3, pOrder.getFieldNumber(0).intValue());
+		assertEquals(1, pOrder.getFieldNumber(1).intValue());
+		assertEquals(0, pOrder.getFieldNumber(2).intValue());
+		assertEquals(Order.ASCENDING, pOrder.getOrder(0));
+		assertEquals(Order.DESCENDING, pOrder.getOrder(1));
+		assertEquals(Order.ASCENDING, pOrder.getOrder(2));
+		assertEquals(IntValue.class, pOrder.getType(0));
+		assertEquals(LongValue.class, pOrder.getType(1));
+		assertEquals(StringValue.class, pOrder.getType(2));
+	}
+
+	@Test
+	public void testRangePartitioningErased() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"1;5"}, null, null, tupleInfo, tupleInfo);
+
+		Ordering o = new Ordering();
+		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
+		o.appendOrdering(5, LongValue.class, Order.DESCENDING);
+		o.appendOrdering(2, StringValue.class, Order.ASCENDING);
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setRangePartitioned(o);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.RANDOM_PARTITIONED, result.getPartitioning());
+		assertNull(result.getPartitioningOrdering());
+		assertNull(result.getPartitioningFields());
+	}
+
+	@Test
+	public void testRebalancingPreserved() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0->1; 1->2; 4->3"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setForcedRebalanced();
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+
+		assertEquals(PartitioningProperty.FORCED_REBALANCED, result.getPartitioning());
+		assertNull(result.getPartitioningFields());
+	}
+
+	@Test
+	public void testUniqueFieldGroupsPreserved1() {
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1;2;3;4"}, null, null, tupleInfo, tupleInfo);
+
+		FieldSet set1 = new FieldSet(0, 1, 2);
+		FieldSet set2 = new FieldSet(3, 4);
+		FieldSet set3 = new FieldSet(4, 5, 6, 7);
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.addUniqueFieldCombination(set1);
+		gprops.addUniqueFieldCombination(set2);
+		gprops.addUniqueFieldCombination(set3);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+		Set<FieldSet> unique = result.getUniqueFieldCombination();
+		FieldSet expected1 = new FieldSet(0, 1, 2);
+		FieldSet expected2 = new FieldSet(3, 4);
+
+		Assert.assertTrue(unique.size() == 2);
+		Assert.assertTrue(unique.contains(expected1));
+		Assert.assertTrue(unique.contains(expected2));
+	}
+
+	@Test
+	public void testUniqueFieldGroupsPreserved2() {
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0->5;1;2;3->6;4"}, null, null, tupleInfo, tupleInfo);
+
+		FieldSet set1 = new FieldSet(0, 1, 2);
+		FieldSet set2 = new FieldSet(3, 4);
+		FieldSet set3 = new FieldSet(4, 5, 6, 7);
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.addUniqueFieldCombination(set1);
+		gprops.addUniqueFieldCombination(set2);
+		gprops.addUniqueFieldCombination(set3);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+		Set<FieldSet> unique = result.getUniqueFieldCombination();
+		FieldSet expected1 = new FieldSet(1, 2, 5);
+		FieldSet expected2 = new FieldSet(4, 6);
+
+		Assert.assertTrue(unique.size() == 2);
+		Assert.assertTrue(unique.contains(expected1));
+		Assert.assertTrue(unique.contains(expected2));
+	}
+
+	@Test
+	public void testUniqueFieldGroupsErased() {
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0; 3; 5; 6; 7"}, null, null, tupleInfo, tupleInfo);
+
+		FieldSet set1 = new FieldSet(0, 1, 2);
+		FieldSet set2 = new FieldSet(3, 4);
+		FieldSet set3 = new FieldSet(4, 5, 6, 7);
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.addUniqueFieldCombination(set1);
+		gprops.addUniqueFieldCombination(set2);
+		gprops.addUniqueFieldCombination(set3);
+
+		GlobalProperties result = gprops.filterBySemanticProperties(sprops, 0);
+		Assert.assertNull(result.getUniqueFieldCombination());
+	}
+
+	@Test(expected = IndexOutOfBoundsException.class)
+	public void testInvalidInputIndex() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1"}, null, null, tupleInfo, tupleInfo);
+
+		GlobalProperties gprops = new GlobalProperties();
+		gprops.setHashPartitioned(new FieldList(0, 1));
+
+		gprops.filterBySemanticProperties(sprops, 1);
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesMatchingTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesMatchingTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesMatchingTest.java
new file mode 100644
index 0000000..52826d6
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesMatchingTest.java
@@ -0,0 +1,294 @@
+/*
+ * 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.dataproperties;
+
+import static org.junit.Assert.*;
+
+import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.api.common.operators.Order;
+import org.apache.flink.api.common.operators.Ordering;
+import org.apache.flink.api.common.operators.util.FieldList;
+import org.apache.flink.api.common.operators.util.FieldSet;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.junit.Test;
+
+public class GlobalPropertiesMatchingTest {
+
+	@Test
+	public void testMatchingAnyPartitioning() {
+		try {
+			
+			RequestedGlobalProperties req = new RequestedGlobalProperties();
+			req.setAnyPartitioning(new FieldSet(6, 2));
+			
+			// match any partitioning
+			{
+				GlobalProperties gp1 = new GlobalProperties();
+				gp1.setAnyPartitioning(new FieldList(2, 6));
+				assertTrue(req.isMetBy(gp1));
+
+				GlobalProperties gp2 = new GlobalProperties();
+				gp2.setAnyPartitioning(new FieldList(6, 2));
+				assertTrue(req.isMetBy(gp2));
+
+				GlobalProperties gp3 = new GlobalProperties();
+				gp3.setAnyPartitioning(new FieldList(6, 2, 4));
+				assertFalse(req.isMetBy(gp3));
+
+				GlobalProperties gp4 = new GlobalProperties();
+				gp4.setAnyPartitioning(new FieldList(6, 1));
+				assertFalse(req.isMetBy(gp4));
+
+				GlobalProperties gp5 = new GlobalProperties();
+				gp5.setAnyPartitioning(new FieldList(2));
+				assertTrue(req.isMetBy(gp5));
+			}
+
+			// match hash partitioning
+			{
+				GlobalProperties gp1 = new GlobalProperties();
+				gp1.setHashPartitioned(new FieldList(2, 6));
+				assertTrue(req.isMetBy(gp1));
+
+				GlobalProperties gp2 = new GlobalProperties();
+				gp2.setHashPartitioned(new FieldList(6, 2));
+				assertTrue(req.isMetBy(gp2));
+
+				GlobalProperties gp3 = new GlobalProperties();
+				gp3.setHashPartitioned(new FieldList(6, 1));
+				assertFalse(req.isMetBy(gp3));
+			}
+			
+			// match range partitioning
+			{
+				GlobalProperties gp1 = new GlobalProperties();
+				gp1.setRangePartitioned(new Ordering(2, null, Order.DESCENDING).appendOrdering(6, null, Order.ASCENDING));
+				assertTrue(req.isMetBy(gp1));
+				
+				GlobalProperties gp2 = new GlobalProperties();
+				gp2.setRangePartitioned(new Ordering(6, null, Order.DESCENDING).appendOrdering(2, null, Order.ASCENDING));
+				assertTrue(req.isMetBy(gp2));
+
+				GlobalProperties gp3 = new GlobalProperties();
+				gp3.setRangePartitioned(new Ordering(6, null, Order.DESCENDING).appendOrdering(1, null, Order.ASCENDING));
+				assertFalse(req.isMetBy(gp3));
+				
+				GlobalProperties gp4 = new GlobalProperties();
+				gp4.setRangePartitioned(new Ordering(6, null, Order.DESCENDING));
+				assertTrue(req.isMetBy(gp4));
+			}
+			
+			// match custom partitioning
+			{
+				GlobalProperties gp1 = new GlobalProperties();
+				gp1.setCustomPartitioned(new FieldList(2, 6), new MockPartitioner());
+				assertTrue(req.isMetBy(gp1));
+				
+				GlobalProperties gp2 = new GlobalProperties();
+				gp2.setCustomPartitioned(new FieldList(6, 2), new MockPartitioner());
+				assertTrue(req.isMetBy(gp2));
+				
+				GlobalProperties gp3 = new GlobalProperties();
+				gp3.setCustomPartitioned(new FieldList(6, 1), new MockPartitioner());
+				assertFalse(req.isMetBy(gp3));
+			}
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+	
+	@Test
+	public void testMatchingCustomPartitioning() {
+		try {
+			final Partitioner<Tuple2<Long, Integer>> partitioner = new MockPartitioner();
+			
+			RequestedGlobalProperties req = new RequestedGlobalProperties();
+			req.setCustomPartitioned(new FieldSet(6, 2), partitioner);
+			
+			// match custom partitionings
+			{
+				GlobalProperties gp1 = new GlobalProperties();
+				gp1.setCustomPartitioned(new FieldList(2, 6), partitioner);
+				assertTrue(req.isMetBy(gp1));
+				
+				GlobalProperties gp2 = new GlobalProperties();
+				gp2.setCustomPartitioned(new FieldList(6, 2), partitioner);
+				assertTrue(req.isMetBy(gp2));
+				
+				GlobalProperties gp3 = new GlobalProperties();
+				gp3.setCustomPartitioned(new FieldList(6, 2), new MockPartitioner());
+				assertFalse(req.isMetBy(gp3));
+			}
+			
+			// cannot match other types of partitionings
+			{
+				GlobalProperties gp1 = new GlobalProperties();
+				gp1.setAnyPartitioning(new FieldList(6, 2));
+				assertFalse(req.isMetBy(gp1));
+				
+				GlobalProperties gp2 = new GlobalProperties();
+				gp2.setHashPartitioned(new FieldList(6, 2));
+				assertFalse(req.isMetBy(gp2));
+				
+				GlobalProperties gp3 = new GlobalProperties();
+				gp3.setRangePartitioned(new Ordering(2, null, Order.DESCENDING).appendOrdering(6, null, Order.ASCENDING));
+				assertFalse(req.isMetBy(gp3));
+			}
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+
+	@Test
+	public void testStrictlyMatchingAnyPartitioning() {
+
+		RequestedGlobalProperties req = new RequestedGlobalProperties();
+		req.setAnyPartitioning(new FieldList(6, 2));
+
+		// match any partitioning
+		{
+			GlobalProperties gp1 = new GlobalProperties();
+			gp1.setAnyPartitioning(new FieldList(6, 2));
+			assertTrue(req.isMetBy(gp1));
+
+			GlobalProperties gp2 = new GlobalProperties();
+			gp2.setAnyPartitioning(new FieldList(2, 6));
+			assertFalse(req.isMetBy(gp2));
+
+			GlobalProperties gp3 = new GlobalProperties();
+			gp3.setAnyPartitioning(new FieldList(6, 2, 3));
+			assertFalse(req.isMetBy(gp3));
+
+			GlobalProperties gp4 = new GlobalProperties();
+			gp3.setAnyPartitioning(new FieldList(6, 1));
+			assertFalse(req.isMetBy(gp3));
+
+			GlobalProperties gp5 = new GlobalProperties();
+			gp4.setAnyPartitioning(new FieldList(2));
+			assertFalse(req.isMetBy(gp4));
+		}
+
+		// match hash partitioning
+		{
+			GlobalProperties gp1 = new GlobalProperties();
+			gp1.setHashPartitioned(new FieldList(6, 2));
+			assertTrue(req.isMetBy(gp1));
+
+			GlobalProperties gp2 = new GlobalProperties();
+			gp2.setHashPartitioned(new FieldList(2, 6));
+			assertFalse(req.isMetBy(gp2));
+
+			GlobalProperties gp3 = new GlobalProperties();
+			gp3.setHashPartitioned(new FieldList(6, 1));
+			assertFalse(req.isMetBy(gp3));
+		}
+
+		// match range partitioning
+		{
+			GlobalProperties gp1 = new GlobalProperties();
+			gp1.setRangePartitioned(new Ordering(6, null, Order.DESCENDING).appendOrdering(2, null, Order.ASCENDING));
+			assertTrue(req.isMetBy(gp1));
+
+			GlobalProperties gp2 = new GlobalProperties();
+			gp2.setRangePartitioned(new Ordering(2, null, Order.DESCENDING).appendOrdering(6, null, Order.ASCENDING));
+			assertFalse(req.isMetBy(gp2));
+
+			GlobalProperties gp3 = new GlobalProperties();
+			gp3.setRangePartitioned(new Ordering(6, null, Order.DESCENDING).appendOrdering(1, null, Order.ASCENDING));
+			assertFalse(req.isMetBy(gp3));
+
+			GlobalProperties gp4 = new GlobalProperties();
+			gp4.setRangePartitioned(new Ordering(6, null, Order.DESCENDING));
+			assertFalse(req.isMetBy(gp4));
+		}
+
+	}
+
+	@Test
+	public void testStrictlyMatchingHashPartitioning() {
+
+		RequestedGlobalProperties req = new RequestedGlobalProperties();
+		req.setHashPartitioned(new FieldList(6, 2));
+
+		// match any partitioning
+		{
+			GlobalProperties gp1 = new GlobalProperties();
+			gp1.setAnyPartitioning(new FieldList(6, 2));
+			assertFalse(req.isMetBy(gp1));
+
+			GlobalProperties gp2 = new GlobalProperties();
+			gp2.setAnyPartitioning(new FieldList(2, 6));
+			assertFalse(req.isMetBy(gp2));
+
+			GlobalProperties gp3 = new GlobalProperties();
+			gp3.setAnyPartitioning(new FieldList(6, 1));
+			assertFalse(req.isMetBy(gp3));
+
+			GlobalProperties gp4 = new GlobalProperties();
+			gp4.setAnyPartitioning(new FieldList(2));
+			assertFalse(req.isMetBy(gp4));
+		}
+
+		// match hash partitioning
+		{
+			GlobalProperties gp1 = new GlobalProperties();
+			gp1.setHashPartitioned(new FieldList(6, 2));
+			assertTrue(req.isMetBy(gp1));
+
+			GlobalProperties gp2 = new GlobalProperties();
+			gp2.setHashPartitioned(new FieldList(2, 6));
+			assertFalse(req.isMetBy(gp2));
+
+			GlobalProperties gp3 = new GlobalProperties();
+			gp3.setHashPartitioned(new FieldList(6, 1));
+			assertFalse(req.isMetBy(gp3));
+
+			GlobalProperties gp4 = new GlobalProperties();
+			gp4.setHashPartitioned(new FieldList(6, 2, 0));
+			assertFalse(req.isMetBy(gp4));
+		}
+
+		// match range partitioning
+		{
+			GlobalProperties gp1 = new GlobalProperties();
+			gp1.setRangePartitioned(new Ordering(6, null, Order.DESCENDING).appendOrdering(2, null, Order.ASCENDING));
+			assertFalse(req.isMetBy(gp1));
+
+			GlobalProperties gp2 = new GlobalProperties();
+			gp2.setRangePartitioned(new Ordering(2, null, Order.DESCENDING).appendOrdering(6, null, Order.ASCENDING));
+			assertFalse(req.isMetBy(gp2));
+
+			GlobalProperties gp3 = new GlobalProperties();
+			gp3.setRangePartitioned(new Ordering(6, null, Order.DESCENDING).appendOrdering(1, null, Order.ASCENDING));
+			assertFalse(req.isMetBy(gp3));
+
+			GlobalProperties gp4 = new GlobalProperties();
+			gp4.setRangePartitioned(new Ordering(6, null, Order.DESCENDING));
+			assertFalse(req.isMetBy(gp4));
+		}
+
+	}
+	
+	// --------------------------------------------------------------------------------------------
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesPushdownTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesPushdownTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesPushdownTest.java
new file mode 100644
index 0000000..0868720
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/GlobalPropertiesPushdownTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.dataproperties;
+
+import static org.junit.Assert.*;
+
+import org.apache.flink.api.common.operators.SemanticProperties;
+import org.apache.flink.api.common.operators.SingleInputSemanticProperties;
+import org.apache.flink.api.common.operators.util.FieldSet;
+import org.junit.Test;
+
+public class GlobalPropertiesPushdownTest {
+
+	@Test
+	public void testAnyPartitioningPushedDown() {
+		try {
+			RequestedGlobalProperties req = new RequestedGlobalProperties();
+			req.setAnyPartitioning(new FieldSet(3, 1));
+			
+			RequestedGlobalProperties preserved = req.filterBySemanticProperties(getAllPreservingSemProps(), 0);
+			assertEquals(PartitioningProperty.ANY_PARTITIONING, preserved.getPartitioning());
+			assertTrue(preserved.getPartitionedFields().isValidSubset(new FieldSet(1, 3)));
+			
+			RequestedGlobalProperties nonPreserved = req.filterBySemanticProperties(getNonePreservingSemProps(), 0);
+			assertTrue(nonPreserved == null || nonPreserved.isTrivial());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+	
+	@Test
+	public void testHashPartitioningPushedDown() {
+		try {
+			RequestedGlobalProperties req = new RequestedGlobalProperties();
+			req.setHashPartitioned(new FieldSet(3, 1));
+			
+			RequestedGlobalProperties preserved = req.filterBySemanticProperties(getAllPreservingSemProps(), 0);
+			assertEquals(PartitioningProperty.HASH_PARTITIONED, preserved.getPartitioning());
+			assertTrue(preserved.getPartitionedFields().isValidSubset(new FieldSet(1, 3)));
+			
+			RequestedGlobalProperties nonPreserved = req.filterBySemanticProperties(getNonePreservingSemProps(), 0);
+			assertTrue(nonPreserved == null || nonPreserved.isTrivial());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+	
+	@Test
+	public void testCustomPartitioningNotPushedDown() {
+		try {
+			RequestedGlobalProperties req = new RequestedGlobalProperties();
+			req.setCustomPartitioned(new FieldSet(3, 1), new MockPartitioner());
+			
+			RequestedGlobalProperties pushedDown = req.filterBySemanticProperties(getAllPreservingSemProps(), 0);
+			assertTrue(pushedDown == null || pushedDown.isTrivial());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+	
+	@Test
+	public void testForcedReblancingNotPushedDown() {
+		try {
+			RequestedGlobalProperties req = new RequestedGlobalProperties();
+			req.setForceRebalancing();
+			
+			RequestedGlobalProperties pushedDown = req.filterBySemanticProperties(getAllPreservingSemProps(), 0);
+			assertTrue(pushedDown == null || pushedDown.isTrivial());
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	private static SemanticProperties getAllPreservingSemProps() {
+		return new SingleInputSemanticProperties.AllFieldsForwardedProperties();
+	}
+	
+	private static SemanticProperties getNonePreservingSemProps() {
+		return new SingleInputSemanticProperties();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/LocalPropertiesFilteringTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/LocalPropertiesFilteringTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/LocalPropertiesFilteringTest.java
new file mode 100644
index 0000000..1ff62ed
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/LocalPropertiesFilteringTest.java
@@ -0,0 +1,373 @@
+/*
+ * 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.dataproperties;
+
+import static org.junit.Assert.*;
+
+import org.apache.flink.api.common.operators.Order;
+import org.apache.flink.api.common.operators.Ordering;
+import org.apache.flink.api.common.operators.SingleInputSemanticProperties;
+import org.apache.flink.api.common.operators.util.FieldList;
+import org.apache.flink.api.common.operators.util.FieldSet;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.functions.SemanticPropUtil;
+import org.apache.flink.api.java.tuple.Tuple8;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.StringValue;
+import org.junit.Test;
+
+public class LocalPropertiesFilteringTest {
+
+	private TupleTypeInfo<Tuple8<Integer, Integer, Integer, Integer, Integer, Integer, Integer, Integer>> tupleInfo =
+			new TupleTypeInfo<Tuple8<Integer, Integer, Integer, Integer, Integer, Integer, Integer, Integer>>(
+					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO,
+					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO
+			);
+
+	@Test
+	public void testAllErased1() {
+
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = LocalProperties.forGrouping(new FieldList(0, 1, 2));
+		lProps = lProps.addUniqueFields(new FieldSet(3,4));
+		lProps = lProps.addUniqueFields(new FieldSet(5,6));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+
+		assertNull(filtered.getGroupedFields());
+		assertNull(filtered.getOrdering());
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testAllErased2() {
+
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"5"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = LocalProperties.forGrouping(new FieldList(0, 1, 2));
+		lProps = lProps.addUniqueFields(new FieldSet(3,4));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+
+		assertNull(filtered.getGroupedFields());
+		assertNull(filtered.getOrdering());
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testGroupingPreserved1() {
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0;2;3"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = LocalProperties.forGrouping(new FieldList(0, 2, 3));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+
+		assertNotNull(filtered.getGroupedFields());
+		assertEquals(3, filtered.getGroupedFields().size());
+		assertTrue(filtered.getGroupedFields().contains(0));
+		assertTrue(filtered.getGroupedFields().contains(2));
+		assertTrue(filtered.getGroupedFields().contains(3));
+		assertNull(filtered.getOrdering());
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testGroupingPreserved2() {
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0->4;2->0;3->7"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = LocalProperties.forGrouping(new FieldList(0, 2, 3));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+
+		assertNotNull(filtered.getGroupedFields());
+		assertEquals(3, filtered.getGroupedFields().size());
+		assertTrue(filtered.getGroupedFields().contains(4));
+		assertTrue(filtered.getGroupedFields().contains(0));
+		assertTrue(filtered.getGroupedFields().contains(7));
+		assertNull(filtered.getOrdering());
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testGroupingErased() {
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0->4;2->0"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = LocalProperties.forGrouping(new FieldList(0, 2, 3));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+
+		assertNull(filtered.getGroupedFields());
+		assertNull(filtered.getOrdering());
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testSortingPreserved1() {
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0;2;5"}, null, null, tupleInfo, tupleInfo);
+
+		Ordering o = new Ordering();
+		o.appendOrdering(2, IntValue.class, Order.ASCENDING);
+		o.appendOrdering(0, StringValue.class, Order.DESCENDING);
+		o.appendOrdering(5, LongValue.class, Order.DESCENDING);
+		LocalProperties lProps = LocalProperties.forOrdering(o);
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+		FieldList gFields = filtered.getGroupedFields();
+		Ordering order = filtered.getOrdering();
+
+		assertNotNull(gFields);
+		assertEquals(3, gFields.size());
+		assertTrue(gFields.contains(0));
+		assertTrue(gFields.contains(2));
+		assertTrue(gFields.contains(5));
+		assertNotNull(order);
+		assertEquals(3, order.getNumberOfFields());
+		assertEquals(2, order.getFieldNumber(0).intValue());
+		assertEquals(0, order.getFieldNumber(1).intValue());
+		assertEquals(5, order.getFieldNumber(2).intValue());
+		assertEquals(Order.ASCENDING, order.getOrder(0));
+		assertEquals(Order.DESCENDING, order.getOrder(1));
+		assertEquals(Order.DESCENDING, order.getOrder(2));
+		assertEquals(IntValue.class, order.getType(0));
+		assertEquals(StringValue.class, order.getType(1));
+		assertEquals(LongValue.class, order.getType(2));
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testSortingPreserved2() {
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0->3;2->7;5->1"}, null, null, tupleInfo, tupleInfo);
+
+		Ordering o = new Ordering();
+		o.appendOrdering(2, IntValue.class, Order.ASCENDING);
+		o.appendOrdering(0, StringValue.class, Order.DESCENDING);
+		o.appendOrdering(5, LongValue.class, Order.DESCENDING);
+		LocalProperties lProps = LocalProperties.forOrdering(o);
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+		FieldList gFields = filtered.getGroupedFields();
+		Ordering order = filtered.getOrdering();
+
+		assertNotNull(gFields);
+		assertEquals(3, gFields.size());
+		assertTrue(gFields.contains(3));
+		assertTrue(gFields.contains(7));
+		assertTrue(gFields.contains(1));
+		assertNotNull(order);
+		assertEquals(3, order.getNumberOfFields());
+		assertEquals(7, order.getFieldNumber(0).intValue());
+		assertEquals(3, order.getFieldNumber(1).intValue());
+		assertEquals(1, order.getFieldNumber(2).intValue());
+		assertEquals(Order.ASCENDING, order.getOrder(0));
+		assertEquals(Order.DESCENDING, order.getOrder(1));
+		assertEquals(Order.DESCENDING, order.getOrder(2));
+		assertEquals(IntValue.class, order.getType(0));
+		assertEquals(StringValue.class, order.getType(1));
+		assertEquals(LongValue.class, order.getType(2));
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testSortingPreserved3() {
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0;2"}, null, null, tupleInfo, tupleInfo);
+
+		Ordering o = new Ordering();
+		o.appendOrdering(2, IntValue.class, Order.ASCENDING);
+		o.appendOrdering(0, StringValue.class, Order.DESCENDING);
+		o.appendOrdering(5, LongValue.class, Order.DESCENDING);
+		LocalProperties lProps = LocalProperties.forOrdering(o);
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+		FieldList gFields = filtered.getGroupedFields();
+		Ordering order = filtered.getOrdering();
+
+		assertNotNull(gFields);
+		assertEquals(2, gFields.size());
+		assertTrue(gFields.contains(0));
+		assertTrue(gFields.contains(2));
+		assertNotNull(order);
+		assertEquals(2, order.getNumberOfFields());
+		assertEquals(2, order.getFieldNumber(0).intValue());
+		assertEquals(0, order.getFieldNumber(1).intValue());
+		assertEquals(Order.ASCENDING, order.getOrder(0));
+		assertEquals(Order.DESCENDING, order.getOrder(1));
+		assertEquals(IntValue.class, order.getType(0));
+		assertEquals(StringValue.class, order.getType(1));
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testSortingPreserved4() {
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"2->7;5"}, null, null, tupleInfo, tupleInfo);
+
+		Ordering o = new Ordering();
+		o.appendOrdering(2, IntValue.class, Order.ASCENDING);
+		o.appendOrdering(0, StringValue.class, Order.DESCENDING);
+		o.appendOrdering(5, LongValue.class, Order.DESCENDING);
+		LocalProperties lProps = LocalProperties.forOrdering(o);
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+		FieldList gFields = filtered.getGroupedFields();
+		Ordering order = filtered.getOrdering();
+
+		assertNotNull(gFields);
+		assertEquals(1, gFields.size());
+		assertTrue(gFields.contains(7));
+		assertNotNull(order);
+		assertEquals(1, order.getNumberOfFields());
+		assertEquals(7, order.getFieldNumber(0).intValue());
+		assertEquals(Order.ASCENDING, order.getOrder(0));
+		assertEquals(IntValue.class, order.getType(0));
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testSortingErased() {
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0;5"}, null, null, tupleInfo, tupleInfo);
+
+		Ordering o = new Ordering();
+		o.appendOrdering(2, IntValue.class, Order.ASCENDING);
+		o.appendOrdering(0, StringValue.class, Order.DESCENDING);
+		o.appendOrdering(5, LongValue.class, Order.DESCENDING);
+		LocalProperties lProps = LocalProperties.forOrdering(o);
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+		FieldList gFields = filtered.getGroupedFields();
+		Ordering order = filtered.getOrdering();
+
+		assertNull(gFields);
+		assertNull(order);
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test
+	public void testUniqueFieldsPreserved1() {
+
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0;1;2;3;4"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = new LocalProperties();
+		lProps = lProps.addUniqueFields(new FieldSet(0,1,2));
+		lProps = lProps.addUniqueFields(new FieldSet(3,4));
+		lProps = lProps.addUniqueFields(new FieldSet(4,5,6));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+		FieldSet expected1 = new FieldSet(0,1,2);
+		FieldSet expected2 = new FieldSet(3,4);
+
+		assertNull(filtered.getGroupedFields());
+		assertNull(filtered.getOrdering());
+		assertNotNull(filtered.getUniqueFields());
+		assertEquals(2, filtered.getUniqueFields().size());
+		assertTrue(filtered.getUniqueFields().contains(expected1));
+		assertTrue(filtered.getUniqueFields().contains(expected2));
+	}
+
+	@Test
+	public void testUniqueFieldsPreserved2() {
+
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0;1;2;3;4"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = LocalProperties.forGrouping(new FieldList(1,2));
+		lProps = lProps.addUniqueFields(new FieldSet(0,1,2));
+		lProps = lProps.addUniqueFields(new FieldSet(3,4));
+		lProps = lProps.addUniqueFields(new FieldSet(4,5,6));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+		FieldSet expected1 = new FieldSet(0,1,2);
+		FieldSet expected2 = new FieldSet(3,4);
+
+		assertNull(filtered.getOrdering());
+		assertNotNull(filtered.getGroupedFields());
+		assertEquals(2, filtered.getGroupedFields().size());
+		assertTrue(filtered.getGroupedFields().contains(1));
+		assertTrue(filtered.getGroupedFields().contains(2));
+		assertNotNull(filtered.getUniqueFields());
+		assertEquals(2, filtered.getUniqueFields().size());
+		assertTrue(filtered.getUniqueFields().contains(expected1));
+		assertTrue(filtered.getUniqueFields().contains(expected2));
+	}
+
+	@Test
+	public void testUniqueFieldsPreserved3() {
+
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0->7;1->6;2->5;3->4;4->3"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = new LocalProperties();
+		lProps = lProps.addUniqueFields(new FieldSet(0,1,2));
+		lProps = lProps.addUniqueFields(new FieldSet(3,4));
+		lProps = lProps.addUniqueFields(new FieldSet(4,5,6));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+		FieldSet expected1 = new FieldSet(5,6,7);
+		FieldSet expected2 = new FieldSet(3,4);
+
+		assertNull(filtered.getGroupedFields());
+		assertNull(filtered.getOrdering());
+		assertNotNull(filtered.getUniqueFields());
+		assertEquals(2, filtered.getUniqueFields().size());
+		assertTrue(filtered.getUniqueFields().contains(expected1));
+		assertTrue(filtered.getUniqueFields().contains(expected2));
+	}
+
+	@Test
+	public void testUniqueFieldsErased() {
+
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, new String[]{"0;1;4"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lProps = new LocalProperties();
+		lProps = lProps.addUniqueFields(new FieldSet(0,1,2));
+		lProps = lProps.addUniqueFields(new FieldSet(3,4));
+		lProps = lProps.addUniqueFields(new FieldSet(4,5,6));
+
+		LocalProperties filtered = lProps.filterBySemanticProperties(sp, 0);
+
+		assertNull(filtered.getGroupedFields());
+		assertNull(filtered.getOrdering());
+		assertNull(filtered.getUniqueFields());
+	}
+
+	@Test(expected = IndexOutOfBoundsException.class)
+	public void testInvalidInputIndex() {
+
+		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1"}, null, null, tupleInfo, tupleInfo);
+
+		LocalProperties lprops = LocalProperties.forGrouping(new FieldList(0,1));
+
+		lprops.filterBySemanticProperties(sprops, 1);
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/MockDistribution.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/MockDistribution.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/MockDistribution.java
new file mode 100644
index 0000000..74126f8
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/MockDistribution.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.dataproperties;
+
+import org.apache.flink.api.common.distributions.DataDistribution;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.types.Key;
+
+import java.io.IOException;
+
+@SuppressWarnings("serial")
+public class MockDistribution implements DataDistribution {
+
+	@Override
+	public Key<?>[] getBucketBoundary(int bucketNum, int totalNumBuckets) {
+		return new Key<?>[0];
+	}
+
+	@Override
+	public int getNumberOfFields() {
+		return 0;
+	}
+
+	@Override
+	public void write(DataOutputView out) throws IOException {
+
+	}
+
+	@Override
+	public void read(DataInputView in) throws IOException {
+
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/MockPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/MockPartitioner.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/MockPartitioner.java
new file mode 100644
index 0000000..2b2ab14
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/dataproperties/MockPartitioner.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.dataproperties;
+
+import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+class MockPartitioner implements Partitioner<Tuple2<Long, Integer>> {
+	
+	private static final long serialVersionUID = 1L;
+
+	@Override
+	public int partition(Tuple2<Long, Integer> key, int numPartitions) {
+		return 0;
+	}
+}
\ No newline at end of file