[22/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-compiler/src/test/java/org/apache/flink/optimizer/dataproperties/RequestedGlobalPropertiesFilteringTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataproperties/RequestedGlobalPropertiesFilteringTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/dataproperties/RequestedGlobalPropertiesFilteringTest.java
deleted file mode 100644
index 9f2c467..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataproperties/RequestedGlobalPropertiesFilteringTest.java
+++ /dev/null
@@ -1,430 +0,0 @@
-/*
- * 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.distributions.DataDistribution;
-import org.apache.flink.api.common.operators.DualInputSemanticProperties;
-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.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.ByteValue;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.LongValue;
-import org.junit.Test;
-
-
-public class RequestedGlobalPropertiesFilteringTest {
-
-	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(expected = NullPointerException.class)
-	public void testNullProps() {
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setAnyPartitioning(new FieldSet(0,1,2));
-
-		rgProps.filterBySemanticProperties(null, 0);
-	}
-
-	@Test
-	public void testEraseAll1() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setAnyPartitioning(new FieldSet(0,1,2));
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testEraseAll2() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"3;4"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setAnyPartitioning(new FieldSet(0, 1, 2));
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testHashPartitioningPreserved1() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"0;3;4"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setHashPartitioned(new FieldSet(0, 3, 4));
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNotNull(filtered);
-		assertEquals(PartitioningProperty.HASH_PARTITIONED, filtered.getPartitioning());
-		assertNotNull(filtered.getPartitionedFields());
-		assertEquals(3, filtered.getPartitionedFields().size());
-		assertTrue(filtered.getPartitionedFields().contains(0));
-		assertTrue(filtered.getPartitionedFields().contains(3));
-		assertTrue(filtered.getPartitionedFields().contains(4));
-		assertNull(filtered.getDataDistribution());
-		assertNull(filtered.getCustomPartitioner());
-		assertNull(filtered.getOrdering());
-	}
-
-	@Test
-	public void testHashPartitioningPreserved2() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"2->0;1->3;7->4"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setHashPartitioned(new FieldSet(0, 3, 4));
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNotNull(filtered);
-		assertEquals(PartitioningProperty.HASH_PARTITIONED, filtered.getPartitioning());
-		assertNotNull(filtered.getPartitionedFields());
-		assertEquals(3, filtered.getPartitionedFields().size());
-		assertTrue(filtered.getPartitionedFields().contains(1));
-		assertTrue(filtered.getPartitionedFields().contains(2));
-		assertTrue(filtered.getPartitionedFields().contains(7));
-		assertNull(filtered.getDataDistribution());
-		assertNull(filtered.getCustomPartitioner());
-		assertNull(filtered.getOrdering());
-	}
-
-	@Test
-	public void testHashPartitioningErased() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"1;2"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setHashPartitioned(new FieldSet(0, 3, 4));
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testAnyPartitioningPreserved1() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"0;3;4"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setAnyPartitioning(new FieldSet(0, 3, 4));
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNotNull(filtered);
-		assertEquals(PartitioningProperty.ANY_PARTITIONING, filtered.getPartitioning());
-		assertNotNull(filtered.getPartitionedFields());
-		assertEquals(3, filtered.getPartitionedFields().size());
-		assertTrue(filtered.getPartitionedFields().contains(0));
-		assertTrue(filtered.getPartitionedFields().contains(3));
-		assertTrue(filtered.getPartitionedFields().contains(4));
-		assertNull(filtered.getDataDistribution());
-		assertNull(filtered.getCustomPartitioner());
-		assertNull(filtered.getOrdering());
-	}
-
-	@Test
-	public void testAnyPartitioningPreserved2() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"2->0;1->3;7->4"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setAnyPartitioning(new FieldSet(0, 3, 4));
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNotNull(filtered);
-		assertEquals(PartitioningProperty.ANY_PARTITIONING, filtered.getPartitioning());
-		assertNotNull(filtered.getPartitionedFields());
-		assertEquals(3, filtered.getPartitionedFields().size());
-		assertTrue(filtered.getPartitionedFields().contains(1));
-		assertTrue(filtered.getPartitionedFields().contains(2));
-		assertTrue(filtered.getPartitionedFields().contains(7));
-		assertNull(filtered.getDataDistribution());
-		assertNull(filtered.getCustomPartitioner());
-		assertNull(filtered.getOrdering());
-	}
-
-	@Test
-	public void testAnyPartitioningErased() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"1;2"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setAnyPartitioning(new FieldSet(0, 3, 4));
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testRangePartitioningPreserved1() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"1;3;6"}, null, null, tupleInfo, tupleInfo);
-
-		Ordering o = new Ordering();
-		o.appendOrdering(3, LongValue.class, Order.DESCENDING);
-		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
-		o.appendOrdering(6, ByteValue.class, Order.DESCENDING);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setRangePartitioned(o);
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNotNull(filtered);
-		assertEquals(PartitioningProperty.RANGE_PARTITIONED, filtered.getPartitioning());
-		assertNotNull(filtered.getOrdering());
-		assertEquals(3, filtered.getOrdering().getNumberOfFields());
-		assertEquals(3, filtered.getOrdering().getFieldNumber(0).intValue());
-		assertEquals(1, filtered.getOrdering().getFieldNumber(1).intValue());
-		assertEquals(6, filtered.getOrdering().getFieldNumber(2).intValue());
-		assertEquals(LongValue.class, filtered.getOrdering().getType(0));
-		assertEquals(IntValue.class, filtered.getOrdering().getType(1));
-		assertEquals(ByteValue.class, filtered.getOrdering().getType(2));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(0));
-		assertEquals(Order.ASCENDING, filtered.getOrdering().getOrder(1));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(2));
-		assertNull(filtered.getPartitionedFields());
-		assertNull(filtered.getDataDistribution());
-		assertNull(filtered.getCustomPartitioner());
-	}
-
-	@Test
-	public void testRangePartitioningPreserved2() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"7->3;1->1;2->6"}, null, null, tupleInfo, tupleInfo);
-
-		Ordering o = new Ordering();
-		o.appendOrdering(3, LongValue.class, Order.DESCENDING);
-		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
-		o.appendOrdering(6, ByteValue.class, Order.DESCENDING);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setRangePartitioned(o);
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNotNull(filtered);
-		assertEquals(PartitioningProperty.RANGE_PARTITIONED, filtered.getPartitioning());
-		assertNotNull(filtered.getOrdering());
-		assertEquals(3, filtered.getOrdering().getNumberOfFields());
-		assertEquals(7, filtered.getOrdering().getFieldNumber(0).intValue());
-		assertEquals(1, filtered.getOrdering().getFieldNumber(1).intValue());
-		assertEquals(2, filtered.getOrdering().getFieldNumber(2).intValue());
-		assertEquals(LongValue.class, filtered.getOrdering().getType(0));
-		assertEquals(IntValue.class, filtered.getOrdering().getType(1));
-		assertEquals(ByteValue.class, filtered.getOrdering().getType(2));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(0));
-		assertEquals(Order.ASCENDING, filtered.getOrdering().getOrder(1));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(2));
-		assertNull(filtered.getPartitionedFields());
-		assertNull(filtered.getDataDistribution());
-		assertNull(filtered.getCustomPartitioner());
-	}
-
-	@Test
-	public void testRangePartitioningPreserved3() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"7->3;1->1;2->6"}, null, null, tupleInfo, tupleInfo);
-
-		DataDistribution dd = new MockDistribution();
-		Ordering o = new Ordering();
-		o.appendOrdering(3, LongValue.class, Order.DESCENDING);
-		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
-		o.appendOrdering(6, ByteValue.class, Order.DESCENDING);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setRangePartitioned(o, dd);
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNotNull(filtered);
-		assertEquals(PartitioningProperty.RANGE_PARTITIONED, filtered.getPartitioning());
-		assertNotNull(filtered.getOrdering());
-		assertEquals(3, filtered.getOrdering().getNumberOfFields());
-		assertEquals(7, filtered.getOrdering().getFieldNumber(0).intValue());
-		assertEquals(1, filtered.getOrdering().getFieldNumber(1).intValue());
-		assertEquals(2, filtered.getOrdering().getFieldNumber(2).intValue());
-		assertEquals(LongValue.class, filtered.getOrdering().getType(0));
-		assertEquals(IntValue.class, filtered.getOrdering().getType(1));
-		assertEquals(ByteValue.class, filtered.getOrdering().getType(2));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(0));
-		assertEquals(Order.ASCENDING, filtered.getOrdering().getOrder(1));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(2));
-		assertNotNull(filtered.getDataDistribution());
-		assertEquals(dd, filtered.getDataDistribution());
-		assertNull(filtered.getPartitionedFields());
-		assertNull(filtered.getCustomPartitioner());
-	}
-
-	@Test
-	public void testRangePartitioningErased() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"1;2"}, null, null, tupleInfo, tupleInfo);
-
-		Ordering o = new Ordering();
-		o.appendOrdering(3, LongValue.class, Order.DESCENDING);
-		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
-		o.appendOrdering(6, ByteValue.class, Order.DESCENDING);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setRangePartitioned(o);
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testCustomPartitioningErased() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"0;1;2"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setCustomPartitioned(new FieldSet(0, 1, 2), new MockPartitioner());
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testRandomDistributionErased() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"0;1;2"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setRandomPartitioning();
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testReplicationErased() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"0;1;2"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setFullyReplicated();
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testRebalancingErased() {
-
-		SingleInputSemanticProperties sProp = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProp, new String[]{"0;1;2"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties rgProps = new RequestedGlobalProperties();
-		rgProps.setForceRebalancing();
-
-		RequestedGlobalProperties filtered = rgProps.filterBySemanticProperties(sProp, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testDualHashPartitioningPreserved() {
-
-		DualInputSemanticProperties dprops = new DualInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsDualFromString(dprops, new String[]{"0;2;4"}, new String[]{"1->3;4->6;3->7"},
-				null, null, null, null, tupleInfo, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties gprops1 = new RequestedGlobalProperties();
-		RequestedGlobalProperties gprops2 = new RequestedGlobalProperties();
-		gprops1.setHashPartitioned(new FieldSet(2, 0, 4));
-		gprops2.setHashPartitioned(new FieldSet(3, 6, 7));
-		RequestedGlobalProperties filtered1 = gprops1.filterBySemanticProperties(dprops, 0);
-		RequestedGlobalProperties filtered2 = gprops2.filterBySemanticProperties(dprops, 1);
-
-		assertNotNull(filtered1);
-		assertEquals(PartitioningProperty.HASH_PARTITIONED, filtered1.getPartitioning());
-		assertNotNull(filtered1.getPartitionedFields());
-		assertEquals(3, filtered1.getPartitionedFields().size());
-		assertTrue(filtered1.getPartitionedFields().contains(0));
-		assertTrue(filtered1.getPartitionedFields().contains(2));
-		assertTrue(filtered1.getPartitionedFields().contains(4));
-		assertNull(filtered1.getOrdering());
-		assertNull(filtered1.getCustomPartitioner());
-		assertNull(filtered1.getDataDistribution());
-
-		assertNotNull(filtered2);
-		assertEquals(PartitioningProperty.HASH_PARTITIONED, filtered2.getPartitioning());
-		assertNotNull(filtered2.getPartitionedFields());
-		assertEquals(3, filtered2.getPartitionedFields().size());
-		assertTrue(filtered2.getPartitionedFields().contains(1));
-		assertTrue(filtered2.getPartitionedFields().contains(3));
-		assertTrue(filtered2.getPartitionedFields().contains(4));
-		assertNull(filtered2.getOrdering());
-		assertNull(filtered2.getCustomPartitioner());
-		assertNull(filtered2.getDataDistribution());
-	}
-
-	@Test(expected = IndexOutOfBoundsException.class)
-	public void testInvalidInputIndex() {
-		SingleInputSemanticProperties sprops = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sprops, new String[]{"0;1"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedGlobalProperties gprops = new RequestedGlobalProperties();
-		gprops.setHashPartitioned(new FieldSet(0,1));
-
-		gprops.filterBySemanticProperties(sprops, 1);
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataproperties/RequestedLocalPropertiesFilteringTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataproperties/RequestedLocalPropertiesFilteringTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/dataproperties/RequestedLocalPropertiesFilteringTest.java
deleted file mode 100644
index 0cede0e..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/dataproperties/RequestedLocalPropertiesFilteringTest.java
+++ /dev/null
@@ -1,246 +0,0 @@
-/*
- * 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.DualInputSemanticProperties;
-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.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.ByteValue;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.LongValue;
-import org.junit.Test;
-
-public class RequestedLocalPropertiesFilteringTest {
-
-	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(expected = NullPointerException.class)
-	public void testNullProps() {
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setGroupedFields(new FieldSet(0, 2, 3));
-
-		rlProp.filterBySemanticProperties(null, 0);
-	}
-
-	@Test
-	public void testAllErased() {
-
-		SingleInputSemanticProperties sProps = new SingleInputSemanticProperties();
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setGroupedFields(new FieldSet(0, 2, 3));
-
-		RequestedLocalProperties filtered = rlProp.filterBySemanticProperties(sProps, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testGroupingPreserved1() {
-
-		SingleInputSemanticProperties sProps = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProps, new String[]{"0;2;3"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setGroupedFields(new FieldSet(0, 2, 3));
-
-		RequestedLocalProperties filtered = rlProp.filterBySemanticProperties(sProps, 0);
-
-		assertNotNull(filtered);
-		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());
-	}
-
-	@Test
-	public void testGroupingPreserved2() {
-
-		SingleInputSemanticProperties sProps = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProps, new String[]{"3->0;5->2;1->3"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setGroupedFields(new FieldSet(0, 2, 3));
-
-		RequestedLocalProperties filtered = rlProp.filterBySemanticProperties(sProps, 0);
-
-		assertNotNull(filtered);
-		assertNotNull(filtered.getGroupedFields());
-		assertEquals(3, filtered.getGroupedFields().size());
-		assertTrue(filtered.getGroupedFields().contains(3));
-		assertTrue(filtered.getGroupedFields().contains(5));
-		assertTrue(filtered.getGroupedFields().contains(1));
-		assertNull(filtered.getOrdering());
-	}
-
-	@Test
-	public void testGroupingErased() {
-
-		SingleInputSemanticProperties sProps = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProps, new String[]{"0;2"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setGroupedFields(new FieldSet(0, 2, 3));
-
-		RequestedLocalProperties filtered = rlProp.filterBySemanticProperties(sProps, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testOrderPreserved1() {
-
-		SingleInputSemanticProperties sProps = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProps, new String[]{"1;4;6"}, null, null, tupleInfo, tupleInfo);
-
-		Ordering o = new Ordering();
-		o.appendOrdering(4, LongValue.class, Order.DESCENDING);
-		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
-		o.appendOrdering(6, ByteValue.class, Order.DESCENDING);
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setOrdering(o);
-
-		RequestedLocalProperties filtered = rlProp.filterBySemanticProperties(sProps, 0);
-
-		assertNotNull(filtered);
-		assertNotNull(filtered.getOrdering());
-		assertEquals(3, filtered.getOrdering().getNumberOfFields());
-		assertEquals(4, filtered.getOrdering().getFieldNumber(0).intValue());
-		assertEquals(1, filtered.getOrdering().getFieldNumber(1).intValue());
-		assertEquals(6, filtered.getOrdering().getFieldNumber(2).intValue());
-		assertEquals(LongValue.class, filtered.getOrdering().getType(0));
-		assertEquals(IntValue.class, filtered.getOrdering().getType(1));
-		assertEquals(ByteValue.class, filtered.getOrdering().getType(2));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(0));
-		assertEquals(Order.ASCENDING, filtered.getOrdering().getOrder(1));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(2));
-		assertNull(filtered.getGroupedFields());
-	}
-
-	@Test
-	public void testOrderPreserved2() {
-
-		SingleInputSemanticProperties sProps = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProps, new String[]{"5->1;0->4;2->6"}, null, null, tupleInfo, tupleInfo);
-
-		Ordering o = new Ordering();
-		o.appendOrdering(4, LongValue.class, Order.DESCENDING);
-		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
-		o.appendOrdering(6, ByteValue.class, Order.DESCENDING);
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setOrdering(o);
-
-		RequestedLocalProperties filtered = rlProp.filterBySemanticProperties(sProps, 0);
-
-		assertNotNull(filtered);
-		assertNotNull(filtered.getOrdering());
-		assertEquals(3, filtered.getOrdering().getNumberOfFields());
-		assertEquals(0, filtered.getOrdering().getFieldNumber(0).intValue());
-		assertEquals(5, filtered.getOrdering().getFieldNumber(1).intValue());
-		assertEquals(2, filtered.getOrdering().getFieldNumber(2).intValue());
-		assertEquals(LongValue.class, filtered.getOrdering().getType(0));
-		assertEquals(IntValue.class, filtered.getOrdering().getType(1));
-		assertEquals(ByteValue.class, filtered.getOrdering().getType(2));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(0));
-		assertEquals(Order.ASCENDING, filtered.getOrdering().getOrder(1));
-		assertEquals(Order.DESCENDING, filtered.getOrdering().getOrder(2));
-		assertNull(filtered.getGroupedFields());
-	}
-
-	@Test
-	public void testOrderErased() {
-
-		SingleInputSemanticProperties sProps = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProps, new String[]{"1; 4"}, null, null, tupleInfo, tupleInfo);
-
-		Ordering o = new Ordering();
-		o.appendOrdering(4, LongValue.class, Order.DESCENDING);
-		o.appendOrdering(1, IntValue.class, Order.ASCENDING);
-		o.appendOrdering(6, ByteValue.class, Order.DESCENDING);
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setOrdering(o);
-
-		RequestedLocalProperties filtered = rlProp.filterBySemanticProperties(sProps, 0);
-
-		assertNull(filtered);
-	}
-
-	@Test
-	public void testDualGroupingPreserved() {
-
-		DualInputSemanticProperties dprops = new DualInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsDualFromString(dprops, new String[]{"1->0;3;2->4"}, new String[]{"0->7;1"},
-				null, null, null, null, tupleInfo, tupleInfo, tupleInfo);
-
-		RequestedLocalProperties lprops1 = new RequestedLocalProperties();
-		lprops1.setGroupedFields(new FieldSet(0,3,4));
-
-		RequestedLocalProperties lprops2 = new RequestedLocalProperties();
-		lprops2.setGroupedFields(new FieldSet(7, 1));
-
-		RequestedLocalProperties filtered1 = lprops1.filterBySemanticProperties(dprops, 0);
-		RequestedLocalProperties filtered2 = lprops2.filterBySemanticProperties(dprops, 1);
-
-		assertNotNull(filtered1);
-		assertNotNull(filtered1.getGroupedFields());
-		assertEquals(3, filtered1.getGroupedFields().size());
-		assertTrue(filtered1.getGroupedFields().contains(1));
-		assertTrue(filtered1.getGroupedFields().contains(2));
-		assertTrue(filtered1.getGroupedFields().contains(3));
-		assertNull(filtered1.getOrdering());
-
-		assertNotNull(filtered2);
-		assertNotNull(filtered2.getGroupedFields());
-		assertEquals(2, filtered2.getGroupedFields().size());
-		assertTrue(filtered2.getGroupedFields().contains(0));
-		assertTrue(filtered2.getGroupedFields().contains(1));
-		assertNull(filtered2.getOrdering());
-	}
-
-	@Test(expected = IndexOutOfBoundsException.class)
-	public void testInvalidInputIndex() {
-
-		SingleInputSemanticProperties sProps = new SingleInputSemanticProperties();
-		SemanticPropUtil.getSemanticPropsSingleFromString(sProps, new String[]{"1; 4"}, null, null, tupleInfo, tupleInfo);
-
-		RequestedLocalProperties rlProp = new RequestedLocalProperties();
-		rlProp.setGroupedFields(new FieldSet(1, 4));
-
-		rlProp.filterBySemanticProperties(sProps, 1);
-	}
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/DeltaIterationDependenciesTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/DeltaIterationDependenciesTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/DeltaIterationDependenciesTest.java
deleted file mode 100644
index b359e6b..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/DeltaIterationDependenciesTest.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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.java;
-
-import static org.junit.Assert.fail;
-
-import org.junit.Test;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.operators.DeltaIteration;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.optimizer.CompilerException;
-import org.apache.flink.optimizer.CompilerTestBase;
-
-
-@SuppressWarnings({"serial", "unchecked"})
-public class DeltaIterationDependenciesTest extends CompilerTestBase {
-
-	@Test
-	public void testExceptionWhenNewWorksetNotDependentOnWorkset() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
-			DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<Long, Long>(0L, 0L));
-
-			DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> deltaIteration = input.iterateDelta(input, 10,0);
-
-			DataSet<Tuple2<Long, Long>> delta = deltaIteration.getSolutionSet().join(deltaIteration.getWorkset())
-														.where(0).equalTo(0)
-														.projectFirst(1).projectSecond(1);
-
-			DataSet<Tuple2<Long, Long>> nextWorkset = deltaIteration.getSolutionSet().join(input)
-														.where(0).equalTo(0)
-														.projectFirst(1).projectSecond(1);
-			
-
-			DataSet<Tuple2<Long, Long>> result = deltaIteration.closeWith(delta, nextWorkset);
-
-			result.print();
-			
-			Plan p = env.createProgramPlan();
-			try {
-				compileNoStats(p);
-				fail("Should not be able to compile, since the next workset does not depend on the workset");
-			}
-			catch (CompilerException e) {
-				// good
-			}
-			catch (Exception e) {
-				fail("wrong exception type");
-			}
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/DistinctAndGroupingOptimizerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/DistinctAndGroupingOptimizerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/DistinctAndGroupingOptimizerTest.java
deleted file mode 100644
index de02836..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/DistinctAndGroupingOptimizerTest.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * 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.java;
-
-import static org.junit.Assert.*;
-
-import org.junit.Test;
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.tuple.Tuple2;
-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.IdentityMapper;
-import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
-
-@SuppressWarnings("serial")
-public class DistinctAndGroupingOptimizerTest extends CompilerTestBase {
-	
-	@Test
-	public void testDistinctPreservesPartitioningOfDistinctFields() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(4);
-			
-			@SuppressWarnings("unchecked")
-			DataSet<Tuple2<Long, Long>> data = env.fromElements(new Tuple2<Long, Long>(0L, 0L), new Tuple2<Long, Long>(1L, 1L))
-					.map(new IdentityMapper<Tuple2<Long,Long>>()).setParallelism(4);
-			
-			data.distinct(0)
-				.groupBy(0)
-				.sum(1)
-				.print();
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			SinkPlanNode sink = op.getDataSinks().iterator().next();
-			SingleInputPlanNode reducer = (SingleInputPlanNode) sink.getInput().getSource();
-			SingleInputPlanNode distinctReducer = (SingleInputPlanNode) reducer.getInput().getSource();
-			
-			assertEquals(ShipStrategyType.FORWARD, sink.getInput().getShipStrategy());
-			
-			// reducer can be forward, reuses partitioning from distinct
-			assertEquals(ShipStrategyType.FORWARD, reducer.getInput().getShipStrategy());
-			
-			// distinct reducer is partitioned
-			assertEquals(ShipStrategyType.PARTITION_HASH, distinctReducer.getInput().getShipStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testDistinctDestroysPartitioningOfNonDistinctFields() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(4);
-			
-			@SuppressWarnings("unchecked")
-			DataSet<Tuple2<Long, Long>> data = env.fromElements(new Tuple2<Long, Long>(0L, 0L), new Tuple2<Long, Long>(1L, 1L))
-					.map(new IdentityMapper<Tuple2<Long,Long>>()).setParallelism(4);
-			
-			data.distinct(1)
-				.groupBy(0)
-				.sum(1)
-				.print();
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			SinkPlanNode sink = op.getDataSinks().iterator().next();
-			SingleInputPlanNode reducer = (SingleInputPlanNode) sink.getInput().getSource();
-			SingleInputPlanNode combiner = (SingleInputPlanNode) reducer.getInput().getSource();
-			SingleInputPlanNode distinctReducer = (SingleInputPlanNode) combiner.getInput().getSource();
-			
-			assertEquals(ShipStrategyType.FORWARD, sink.getInput().getShipStrategy());
-			
-			// reducer must repartition, because it works on a different field
-			assertEquals(ShipStrategyType.PARTITION_HASH, reducer.getInput().getShipStrategy());
-
-			assertEquals(ShipStrategyType.FORWARD, combiner.getInput().getShipStrategy());
-			
-			// distinct reducer is partitioned
-			assertEquals(ShipStrategyType.PARTITION_HASH, distinctReducer.getInput().getShipStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java
deleted file mode 100644
index a683968..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/GroupReduceCompilationTest.java
+++ /dev/null
@@ -1,368 +0,0 @@
-/*
- * 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.java;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.operators.util.FieldList;
-import org.apache.flink.api.common.functions.RichGroupReduceFunction;
-import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.api.java.operators.GroupReduceOperator;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.runtime.operators.DriverStrategy;
-import org.apache.flink.util.Collector;
-import org.junit.Test;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-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.plan.SourcePlanNode;
-
-import static org.junit.Assert.*;
-
-@SuppressWarnings("serial")
-public class GroupReduceCompilationTest extends CompilerTestBase implements java.io.Serializable {
-
-	@Test
-	public void testAllGroupReduceNoCombiner() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Double> data = env.fromElements(0.2, 0.3, 0.4, 0.5).name("source");
-			
-			data.reduceGroup(new RichGroupReduceFunction<Double, Double>() {
-				public void reduce(Iterable<Double> values, Collector<Double> out) {}
-			}).name("reducer")
-			.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			
-			// the all-reduce has no combiner, when the DOP of the input is one
-			
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// check wiring
-			assertEquals(sourceNode, reduceNode.getInput().getSource());
-			assertEquals(reduceNode, sinkNode.getInput().getSource());
-			
-			// check that reduce has the right strategy
-			assertEquals(DriverStrategy.ALL_GROUP_REDUCE, reduceNode.getDriverStrategy());
-			
-			// check DOP
-			assertEquals(1, sourceNode.getParallelism());
-			assertEquals(1, reduceNode.getParallelism());
-			assertEquals(1, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testAllReduceWithCombiner() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Long> data = env.generateSequence(1, 8000000).name("source");
-			
-			GroupReduceOperator<Long, Long> reduced = data.reduceGroup(new RichGroupReduceFunction<Long, Long>() {
-				public void reduce(Iterable<Long> values, Collector<Long> out) {}
-			}).name("reducer");
-			
-			reduced.setCombinable(true);
-			reduced.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			// get the original nodes
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// get the combiner
-			SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
-			
-			// check wiring
-			assertEquals(sourceNode, combineNode.getInput().getSource());
-			assertEquals(reduceNode, sinkNode.getInput().getSource());
-			
-			// check that both reduce and combiner have the same strategy
-			assertEquals(DriverStrategy.ALL_GROUP_REDUCE, reduceNode.getDriverStrategy());
-			assertEquals(DriverStrategy.ALL_GROUP_REDUCE_COMBINE, combineNode.getDriverStrategy());
-			
-			// check DOP
-			assertEquals(8, sourceNode.getParallelism());
-			assertEquals(8, combineNode.getParallelism());
-			assertEquals(1, reduceNode.getParallelism());
-			assertEquals(1, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-	
-	
-	@Test
-	public void testGroupedReduceWithFieldPositionKeyNonCombinable() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
-				.name("source").setParallelism(6);
-			
-			data
-				.groupBy(1)
-				.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
-				public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
-			}).name("reducer")
-			.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			// get the original nodes
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// check wiring
-			assertEquals(sourceNode, reduceNode.getInput().getSource());
-			assertEquals(reduceNode, sinkNode.getInput().getSource());
-			
-			// check that both reduce and combiner have the same strategy
-			assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
-			
-			// check the keys
-			assertEquals(new FieldList(1), reduceNode.getKeys(0));
-			assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
-			
-			// check DOP
-			assertEquals(6, sourceNode.getParallelism());
-			assertEquals(8, reduceNode.getParallelism());
-			assertEquals(8, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testGroupedReduceWithFieldPositionKeyCombinable() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
-				.name("source").setParallelism(6);
-			
-			GroupReduceOperator<Tuple2<String, Double>, Tuple2<String, Double>> reduced = data
-					.groupBy(1)
-					.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
-				public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
-			}).name("reducer");
-			
-			reduced.setCombinable(true);
-			reduced.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			// get the original nodes
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// get the combiner
-			SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
-			
-			// check wiring
-			assertEquals(sourceNode, combineNode.getInput().getSource());
-			assertEquals(reduceNode, sinkNode.getInput().getSource());
-			
-			// check that both reduce and combiner have the same strategy
-			assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
-			assertEquals(DriverStrategy.SORTED_GROUP_COMBINE, combineNode.getDriverStrategy());
-			
-			// check the keys
-			assertEquals(new FieldList(1), reduceNode.getKeys(0));
-			assertEquals(new FieldList(1), combineNode.getKeys(0));
-			assertEquals(new FieldList(1), combineNode.getKeys(1));
-			assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
-			
-			// check DOP
-			assertEquals(6, sourceNode.getParallelism());
-			assertEquals(6, combineNode.getParallelism());
-			assertEquals(8, reduceNode.getParallelism());
-			assertEquals(8, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testGroupedReduceWithSelectorFunctionKeyNoncombinable() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
-				.name("source").setParallelism(6);
-			
-			data
-				.groupBy(new KeySelector<Tuple2<String,Double>, String>() { 
-					public String getKey(Tuple2<String, Double> value) { return value.f0; }
-				})
-				.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
-				public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
-			}).name("reducer")
-			.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			// get the original nodes
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// get the key extractors and projectors
-			SingleInputPlanNode keyExtractor = (SingleInputPlanNode) reduceNode.getInput().getSource();
-			SingleInputPlanNode keyProjector = (SingleInputPlanNode) sinkNode.getInput().getSource();
-			
-			// check wiring
-			assertEquals(sourceNode, keyExtractor.getInput().getSource());
-			assertEquals(keyProjector, sinkNode.getInput().getSource());
-			
-			// check that both reduce and combiner have the same strategy
-			assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
-			
-			// check the keys
-			assertEquals(new FieldList(0), reduceNode.getKeys(0));
-			assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
-			
-			// check DOP
-			assertEquals(6, sourceNode.getParallelism());
-			assertEquals(6, keyExtractor.getParallelism());
-			
-			assertEquals(8, reduceNode.getParallelism());
-			assertEquals(8, keyProjector.getParallelism());
-			assertEquals(8, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testGroupedReduceWithSelectorFunctionKeyCombinable() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
-				.name("source").setParallelism(6);
-			
-			GroupReduceOperator<Tuple2<String, Double>, Tuple2<String, Double>> reduced = data
-				.groupBy(new KeySelector<Tuple2<String,Double>, String>() { 
-					public String getKey(Tuple2<String, Double> value) { return value.f0; }
-				})
-				.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
-				public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
-			}).name("reducer");
-			
-			reduced.setCombinable(true);
-			reduced.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			// get the original nodes
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// get the combiner
-			SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
-			
-			// get the key extractors and projectors
-			SingleInputPlanNode keyExtractor = (SingleInputPlanNode) combineNode.getInput().getSource();
-			SingleInputPlanNode keyProjector = (SingleInputPlanNode) sinkNode.getInput().getSource();
-			
-			// check wiring
-			assertEquals(sourceNode, keyExtractor.getInput().getSource());
-			assertEquals(keyProjector, sinkNode.getInput().getSource());
-			
-			// check that both reduce and combiner have the same strategy
-			assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
-			assertEquals(DriverStrategy.SORTED_GROUP_COMBINE, combineNode.getDriverStrategy());
-			
-			// check the keys
-			assertEquals(new FieldList(0), reduceNode.getKeys(0));
-			assertEquals(new FieldList(0), combineNode.getKeys(0));
-			assertEquals(new FieldList(0), combineNode.getKeys(1));
-			assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
-			
-			// check DOP
-			assertEquals(6, sourceNode.getParallelism());
-			assertEquals(6, keyExtractor.getParallelism());
-			assertEquals(6, combineNode.getParallelism());
-			
-			assertEquals(8, reduceNode.getParallelism());
-			assertEquals(8, keyProjector.getParallelism());
-			assertEquals(8, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java
deleted file mode 100644
index 37a8e81..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/IterationCompilerTest.java
+++ /dev/null
@@ -1,189 +0,0 @@
-/*
- * 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.java;
-
-import static org.junit.Assert.*;
-
-import org.apache.flink.api.common.Plan;
-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.operators.DeltaIteration;
-import org.apache.flink.api.java.operators.IterativeDataSet;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.optimizer.CompilerTestBase;
-import org.apache.flink.optimizer.plan.BulkIterationPlanNode;
-import org.apache.flink.optimizer.plan.NAryUnionPlanNode;
-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.plan.WorksetIterationPlanNode;
-import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
-import org.apache.flink.optimizer.testfunctions.IdentityMapper;
-import org.junit.Test;
-
-@SuppressWarnings("serial")
-public class IterationCompilerTest extends CompilerTestBase {
-
-	@Test
-	public void testIdentityIteration() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(43);
-			
-			IterativeDataSet<Long> iteration = env.generateSequence(-4, 1000).iterate(100);
-			iteration.closeWith(iteration).print();
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			new JobGraphGenerator().compileJobGraph(op);
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testEmptyWorksetIteration() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(43);
-			
-			DataSet<Tuple2<Long, Long>> input = env.generateSequence(1, 20)
-					.map(new MapFunction<Long, Tuple2<Long, Long>>() {
-						@Override
-						public Tuple2<Long, Long> map(Long value){ return null; }
-					});
-					
-					
-			DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iter = input.iterateDelta(input, 100, 0);
-			iter.closeWith(iter.getWorkset(), iter.getWorkset())
-				.print();
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			new JobGraphGenerator().compileJobGraph(op);
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testIterationWithUnionRoot() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(43);
-			
-			IterativeDataSet<Long> iteration = env.generateSequence(-4, 1000).iterate(100);
-			
-			iteration.closeWith(
-					iteration.map(new IdentityMapper<Long>()).union(iteration.map(new IdentityMapper<Long>())))
-					.print();
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			SinkPlanNode sink = op.getDataSinks().iterator().next();
-			BulkIterationPlanNode iterNode = (BulkIterationPlanNode) sink.getInput().getSource();
-			
-			// make sure that the root is part of the dynamic path
-			
-			// the "NoOp" that comes after the union.
-			SingleInputPlanNode noop = (SingleInputPlanNode) iterNode.getRootOfStepFunction();
-			NAryUnionPlanNode union = (NAryUnionPlanNode) noop.getInput().getSource();
-			
-			assertTrue(noop.isOnDynamicPath());
-			assertTrue(noop.getCostWeight() >= 1);
-			
-			assertTrue(union.isOnDynamicPath());
-			assertTrue(union.getCostWeight() >= 1);
-			
-			// see that the jobgraph generator can translate this
-			new JobGraphGenerator().compileJobGraph(op);
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testWorksetIterationWithUnionRoot() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(43);
-			
-			DataSet<Tuple2<Long, Long>> input = env.generateSequence(1, 20)
-					.map(new MapFunction<Long, Tuple2<Long, Long>>() {
-						@Override
-						public Tuple2<Long, Long> map(Long value){ return null; }
-					});
-					
-					
-			DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iter = input.iterateDelta(input, 100, 0);
-			iter.closeWith(
-					iter.getWorkset().map(new IdentityMapper<Tuple2<Long,Long>>())
-				.union(
-					iter.getWorkset().map(new IdentityMapper<Tuple2<Long,Long>>()))
-				, iter.getWorkset().map(new IdentityMapper<Tuple2<Long,Long>>())
-				.union(
-						iter.getWorkset().map(new IdentityMapper<Tuple2<Long,Long>>()))
-				)
-			.print();
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			SinkPlanNode sink = op.getDataSinks().iterator().next();
-			WorksetIterationPlanNode iterNode = (WorksetIterationPlanNode) sink.getInput().getSource();
-			
-			// make sure that the root is part of the dynamic path
-			
-			// the "NoOp"a that come after the union.
-			SingleInputPlanNode nextWorksetNoop = (SingleInputPlanNode) iterNode.getNextWorkSetPlanNode();
-			SingleInputPlanNode solutionDeltaNoop = (SingleInputPlanNode) iterNode.getSolutionSetDeltaPlanNode();
-			
-			NAryUnionPlanNode nextWorksetUnion = (NAryUnionPlanNode) nextWorksetNoop.getInput().getSource();
-			NAryUnionPlanNode solutionDeltaUnion = (NAryUnionPlanNode) solutionDeltaNoop.getInput().getSource();
-			
-			assertTrue(nextWorksetNoop.isOnDynamicPath());
-			assertTrue(nextWorksetNoop.getCostWeight() >= 1);
-			
-			assertTrue(solutionDeltaNoop.isOnDynamicPath());
-			assertTrue(solutionDeltaNoop.getCostWeight() >= 1);
-			
-			assertTrue(nextWorksetUnion.isOnDynamicPath());
-			assertTrue(nextWorksetUnion.getCostWeight() >= 1);
-			
-			assertTrue(solutionDeltaUnion.isOnDynamicPath());
-			assertTrue(solutionDeltaUnion.getCostWeight() >= 1);
-			
-			new JobGraphGenerator().compileJobGraph(op);
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java
deleted file mode 100644
index 0a62132..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/JoinTranslationTest.java
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- * 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.java;
-
-import static org.junit.Assert.*;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.operators.GenericDataSourceBase;
-import org.apache.flink.api.common.operators.Operator;
-import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.functions.KeySelector;
-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.SinkPlanNode;
-import org.apache.flink.runtime.operators.DriverStrategy;
-import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
-import org.apache.flink.util.Visitor;
-import org.junit.Test;
-
-@SuppressWarnings("serial")
-public class JoinTranslationTest extends CompilerTestBase {
-
-	@Test
-	public void testBroadcastHashFirstTest() {
-		try {
-			DualInputPlanNode node = createPlanAndGetJoinNode(JoinHint.BROADCAST_HASH_FIRST);
-			assertEquals(ShipStrategyType.BROADCAST, node.getInput1().getShipStrategy());
-			assertEquals(ShipStrategyType.FORWARD, node.getInput2().getShipStrategy());
-			assertEquals(DriverStrategy.HYBRIDHASH_BUILD_FIRST, node.getDriverStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + ": " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testBroadcastHashSecondTest() {
-		try {
-			DualInputPlanNode node = createPlanAndGetJoinNode(JoinHint.BROADCAST_HASH_SECOND);
-			assertEquals(ShipStrategyType.FORWARD, node.getInput1().getShipStrategy());
-			assertEquals(ShipStrategyType.BROADCAST, node.getInput2().getShipStrategy());
-			assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, node.getDriverStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + ": " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testPartitionHashFirstTest() {
-		try {
-			DualInputPlanNode node = createPlanAndGetJoinNode(JoinHint.REPARTITION_HASH_FIRST);
-			assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput1().getShipStrategy());
-			assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput2().getShipStrategy());
-			assertEquals(DriverStrategy.HYBRIDHASH_BUILD_FIRST, node.getDriverStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + ": " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testPartitionHashSecondTest() {
-		try {
-			DualInputPlanNode node = createPlanAndGetJoinNode(JoinHint.REPARTITION_HASH_SECOND);
-			assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput1().getShipStrategy());
-			assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput2().getShipStrategy());
-			assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, node.getDriverStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + ": " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testPartitionSortMergeTest() {
-		try {
-			DualInputPlanNode node = createPlanAndGetJoinNode(JoinHint.REPARTITION_SORT_MERGE);
-			assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput1().getShipStrategy());
-			assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput2().getShipStrategy());
-			assertEquals(DriverStrategy.MERGE, node.getDriverStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + ": " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testOptimizerChoosesTest() {
-		try {
-			DualInputPlanNode node = createPlanAndGetJoinNode(JoinHint.OPTIMIZER_CHOOSES);
-			assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput1().getShipStrategy());
-			assertEquals(ShipStrategyType.PARTITION_HASH, node.getInput2().getShipStrategy());
-			assertTrue(DriverStrategy.HYBRIDHASH_BUILD_FIRST == node.getDriverStrategy() ||
-					DriverStrategy.HYBRIDHASH_BUILD_SECOND == node.getDriverStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + ": " + e.getMessage());
-		}
-	}
-	
-	
-	private DualInputPlanNode createPlanAndGetJoinNode(JoinHint hint) {
-		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		DataSet<Long> i1 = env.generateSequence(1, 1000);
-		DataSet<Long> i2 = env.generateSequence(1, 1000);
-		
-		i1.join(i2, hint).where(new IdentityKeySelector<Long>()).equalTo(new IdentityKeySelector<Long>()).print();
-		
-		Plan plan = env.createProgramPlan();
-		
-		// set statistics to the sources
-		plan.accept(new Visitor<Operator<?>>() {
-			@Override
-			public boolean preVisit(Operator<?> visitable) {
-				if (visitable instanceof GenericDataSourceBase) {
-					GenericDataSourceBase<?, ?> source = (GenericDataSourceBase<?, ?>) visitable;
-					setSourceStatistics(source, 10000000, 1000);
-				}
-				
-				return true;
-			}
-			
-			@Override
-			public void postVisit(Operator<?> visitable) {}
-		});
-		
-		OptimizedPlan op = compileWithStats(plan);
-		
-		return (DualInputPlanNode) ((SinkPlanNode) op.getDataSinks().iterator().next()).getInput().getSource();
-	}
-	
-	
-	
-	private static final class IdentityKeySelector<T> implements KeySelector<T, T> {
-		
-		@Override
-		public T getKey(T value) {
-			return value;
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/OpenIterationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/OpenIterationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/OpenIterationTest.java
deleted file mode 100644
index cd63b72..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/OpenIterationTest.java
+++ /dev/null
@@ -1,183 +0,0 @@
-/*
- * 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.java;
-
-import static org.junit.Assert.*;
-
-import org.apache.flink.api.common.InvalidProgramException;
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.operators.DeltaIteration;
-import org.apache.flink.api.java.operators.IterativeDataSet;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.optimizer.CompilerTestBase;
-import org.apache.flink.optimizer.testfunctions.IdentityMapper;
-import org.junit.Test;
-
-@SuppressWarnings("serial")
-public class OpenIterationTest extends CompilerTestBase {
-
-	@Test
-	public void testSinkInOpenBulkIteration() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			
-			DataSet<Long> input = env.generateSequence(1, 10);
-			
-			IterativeDataSet<Long> iteration = input.iterate(10);
-			
-			DataSet<Long> mapped = iteration.map(new IdentityMapper<Long>());
-			
-			mapped.print();
-			
-			try {
-				env.createProgramPlan();
-				fail("should throw an exception");
-			}
-			catch (InvalidProgramException e) {
-				// expected
-			}
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testSinkInClosedBulkIteration() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			
-			DataSet<Long> input = env.generateSequence(1, 10);
-			
-			IterativeDataSet<Long> iteration = input.iterate(10);
-			
-			DataSet<Long> mapped = iteration.map(new IdentityMapper<Long>());
-			
-			iteration.closeWith(mapped).print();
-			
-			mapped.print();
-			
-			Plan p = env.createProgramPlan();
-			
-			try {
-				compileNoStats(p);
-				fail("should throw an exception");
-			}
-			catch (InvalidProgramException e) {
-				// expected
-			}
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testSinkOnSolutionSetDeltaIteration() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			
-			@SuppressWarnings("unchecked")
-			DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<Long, Long>(0L,0L));
-			
-			DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration = input.iterateDelta(input, 10, 0);
-			
-			DataSet<Tuple2<Long, Long>> mapped = iteration.getSolutionSet().map(new IdentityMapper<Tuple2<Long, Long>>());
-			
-			mapped.print();
-			
-			try {
-				env.createProgramPlan();
-				fail("should throw an exception");
-			}
-			catch (InvalidProgramException e) {
-				// expected
-			}
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testSinkOnWorksetDeltaIteration() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			
-			@SuppressWarnings("unchecked")
-			DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<Long, Long>(0L,0L));
-			
-			DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration = input.iterateDelta(input, 10, 0);
-			
-			DataSet<Tuple2<Long, Long>> mapped = iteration.getWorkset().map(new IdentityMapper<Tuple2<Long, Long>>());
-			
-			mapped.print();
-			
-			try {
-				env.createProgramPlan();
-				fail("should throw an exception");
-			}
-			catch (InvalidProgramException e) {
-				// expected
-			}
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testOperationOnSolutionSet() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			
-			@SuppressWarnings("unchecked")
-			DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<Long, Long>(0L,0L));
-			
-			DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration = input.iterateDelta(input, 10, 0);
-			
-			DataSet<Tuple2<Long, Long>> mapped = iteration.getSolutionSet().map(new IdentityMapper<Tuple2<Long, Long>>());
-			
-			DataSet<Tuple2<Long, Long>> joined = iteration.getWorkset().join(mapped)
-												.where(0).equalTo(0).projectFirst(1).projectSecond(0);
-			
-			iteration.closeWith(joined, joined)
-				.print();
-			
-			Plan p = env.createProgramPlan();
-			try {
-				compileNoStats(p);
-				fail("should throw an exception");
-			}
-			catch (InvalidProgramException e) {
-				// expected
-			}
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/PartitionOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/PartitionOperatorTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/PartitionOperatorTest.java
deleted file mode 100644
index 8bb9a76..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/PartitionOperatorTest.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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.java;
-
-import static org.junit.Assert.*;
-
-import java.util.Collections;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.functions.Partitioner;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.tuple.Tuple2;
-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.IdentityGroupReducer;
-import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
-import org.junit.Test;
-
-@SuppressWarnings("serial")
-public class PartitionOperatorTest extends CompilerTestBase {
-
-	@Test
-	public void testPartitionOperatorPreservesFields() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			
-			DataSet<Tuple2<Long, Long>> data = env.fromCollection(Collections.singleton(new Tuple2<Long, Long>(0L, 0L)));
-			
-			data.partitionCustom(new Partitioner<Long>() {
-					public int partition(Long key, int numPartitions) { return key.intValue(); }
-				}, 1)
-				.groupBy(1)
-				.reduceGroup(new IdentityGroupReducer<Tuple2<Long,Long>>())
-				.print();
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			SinkPlanNode sink = op.getDataSinks().iterator().next();
-			SingleInputPlanNode reducer = (SingleInputPlanNode) sink.getInput().getSource();
-			SingleInputPlanNode partitioner = (SingleInputPlanNode) reducer.getInput().getSource();
-
-			assertEquals(ShipStrategyType.FORWARD, reducer.getInput().getShipStrategy());
-			assertEquals(ShipStrategyType.PARTITION_CUSTOM, partitioner.getInput().getShipStrategy());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java b/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java
deleted file mode 100644
index 0724a9f..0000000
--- a/flink-compiler/src/test/java/org/apache/flink/optimizer/java/ReduceCompilationTest.java
+++ /dev/null
@@ -1,261 +0,0 @@
-/*
- * 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.java;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.operators.util.FieldList;
-import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.api.common.functions.RichReduceFunction;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.junit.Test;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-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.plan.SourcePlanNode;
-import org.apache.flink.runtime.operators.DriverStrategy;
-
-import static org.junit.Assert.*;
-
-@SuppressWarnings("serial")
-public class ReduceCompilationTest extends CompilerTestBase implements java.io.Serializable {
-
-	@Test
-	public void testAllReduceNoCombiner() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Double> data = env.fromElements(0.2, 0.3, 0.4, 0.5).name("source");
-			
-			data.reduce(new RichReduceFunction<Double>() {
-				
-				@Override
-				public Double reduce(Double value1, Double value2){
-					return value1 + value2;
-				}
-			}).name("reducer")
-			.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			
-			// the all-reduce has no combiner, when the DOP of the input is one
-			
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// check wiring
-			assertEquals(sourceNode, reduceNode.getInput().getSource());
-			assertEquals(reduceNode, sinkNode.getInput().getSource());
-			
-			// check DOP
-			assertEquals(1, sourceNode.getParallelism());
-			assertEquals(1, reduceNode.getParallelism());
-			assertEquals(1, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testAllReduceWithCombiner() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Long> data = env.generateSequence(1, 8000000).name("source");
-			
-			data.reduce(new RichReduceFunction<Long>() {
-				
-				@Override
-				public Long reduce(Long value1, Long value2){
-					return value1 + value2;
-				}
-			}).name("reducer")
-			.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			// get the original nodes
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// get the combiner
-			SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
-			
-			// check wiring
-			assertEquals(sourceNode, combineNode.getInput().getSource());
-			assertEquals(reduceNode, sinkNode.getInput().getSource());
-			
-			// check that both reduce and combiner have the same strategy
-			assertEquals(DriverStrategy.ALL_REDUCE, reduceNode.getDriverStrategy());
-			assertEquals(DriverStrategy.ALL_REDUCE, combineNode.getDriverStrategy());
-			
-			// check DOP
-			assertEquals(8, sourceNode.getParallelism());
-			assertEquals(8, combineNode.getParallelism());
-			assertEquals(1, reduceNode.getParallelism());
-			assertEquals(1, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testGroupedReduceWithFieldPositionKey() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
-				.name("source").setParallelism(6);
-			
-			data
-				.groupBy(1)
-				.reduce(new RichReduceFunction<Tuple2<String,Double>>() {
-				@Override
-				public Tuple2<String, Double> reduce(Tuple2<String, Double> value1, Tuple2<String, Double> value2){
-					return null;
-				}
-			}).name("reducer")
-			.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			// get the original nodes
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// get the combiner
-			SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
-			
-			// check wiring
-			assertEquals(sourceNode, combineNode.getInput().getSource());
-			assertEquals(reduceNode, sinkNode.getInput().getSource());
-			
-			// check that both reduce and combiner have the same strategy
-			assertEquals(DriverStrategy.SORTED_REDUCE, reduceNode.getDriverStrategy());
-			assertEquals(DriverStrategy.SORTED_PARTIAL_REDUCE, combineNode.getDriverStrategy());
-			
-			// check the keys
-			assertEquals(new FieldList(1), reduceNode.getKeys(0));
-			assertEquals(new FieldList(1), combineNode.getKeys(0));
-			assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
-			
-			// check DOP
-			assertEquals(6, sourceNode.getParallelism());
-			assertEquals(6, combineNode.getParallelism());
-			assertEquals(8, reduceNode.getParallelism());
-			assertEquals(8, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-	
-	@Test
-	public void testGroupedReduceWithSelectorFunctionKey() {
-		try {
-			ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-			env.setDegreeOfParallelism(8);
-			
-			DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
-				.name("source").setParallelism(6);
-			
-			data
-				.groupBy(new KeySelector<Tuple2<String,Double>, String>() { 
-					public String getKey(Tuple2<String, Double> value) { return value.f0; }
-				})
-				.reduce(new RichReduceFunction<Tuple2<String,Double>>() {
-				@Override
-				public Tuple2<String, Double> reduce(Tuple2<String, Double> value1, Tuple2<String, Double> value2){
-					return null;
-				}
-			}).name("reducer")
-			.print().name("sink");
-			
-			Plan p = env.createProgramPlan();
-			OptimizedPlan op = compileNoStats(p);
-			
-			OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
-			
-			// get the original nodes
-			SourcePlanNode sourceNode = resolver.getNode("source");
-			SingleInputPlanNode reduceNode = resolver.getNode("reducer");
-			SinkPlanNode sinkNode = resolver.getNode("sink");
-			
-			// get the combiner
-			SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
-			
-			// get the key extractors and projectors
-			SingleInputPlanNode keyExtractor = (SingleInputPlanNode) combineNode.getInput().getSource();
-			SingleInputPlanNode keyProjector = (SingleInputPlanNode) sinkNode.getInput().getSource();
-			
-			// check wiring
-			assertEquals(sourceNode, keyExtractor.getInput().getSource());
-			assertEquals(keyProjector, sinkNode.getInput().getSource());
-			
-			// check that both reduce and combiner have the same strategy
-			assertEquals(DriverStrategy.SORTED_REDUCE, reduceNode.getDriverStrategy());
-			assertEquals(DriverStrategy.SORTED_PARTIAL_REDUCE, combineNode.getDriverStrategy());
-			
-			// check the keys
-			assertEquals(new FieldList(0), reduceNode.getKeys(0));
-			assertEquals(new FieldList(0), combineNode.getKeys(0));
-			assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
-			
-			// check DOP
-			assertEquals(6, sourceNode.getParallelism());
-			assertEquals(6, keyExtractor.getParallelism());
-			assertEquals(6, combineNode.getParallelism());
-			
-			assertEquals(8, reduceNode.getParallelism());
-			assertEquals(8, keyProjector.getParallelism());
-			assertEquals(8, sinkNode.getParallelism());
-		}
-		catch (Exception e) {
-			System.err.println(e.getMessage());
-			e.printStackTrace();
-			fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
-		}
-	}
-}