Move operator ITCases into correct package

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CoGroupGroupSortITCase.java

+/*
+ * 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.test.javaApiOperators;
+
+import org.apache.flink.api.common.functions.CoGroupFunction;
+import org.apache.flink.api.common.operators.Order;
+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.test.util.JavaProgramTestBase;
+import org.apache.flink.types.NullValue;
+import org.apache.flink.util.Collector;
+import org.junit.Assert;
+
+@SuppressWarnings({"serial", "unchecked"})
+public class CoGroupGroupSortITCase extends JavaProgramTestBase {
+
+	@Override
+	protected void testProgram() throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		
+		
+		DataSet<Tuple2<Long, Long>> input1 = env.fromElements(
+				new Tuple2<Long, Long>(0L, 5L),
+				new Tuple2<Long, Long>(0L, 4L),
+				new Tuple2<Long, Long>(0L, 3L),
+				new Tuple2<Long, Long>(0L, 2L),
+				new Tuple2<Long, Long>(0L, 1L),
+				new Tuple2<Long, Long>(1L, 10L),
+				new Tuple2<Long, Long>(1L, 8L),
+				new Tuple2<Long, Long>(1L, 9L),
+				new Tuple2<Long, Long>(1L, 7L));
+		
+		DataSet<TestPojo> input2 = env.fromElements(
+				new TestPojo(0L, 10L, 3L),
+				new TestPojo(0L, 8L, 3L),
+				new TestPojo(0L, 10L, 1L),
+				new TestPojo(0L, 9L, 0L),
+				new TestPojo(0L, 8L, 2L),
+				new TestPojo(0L, 8L, 4L),
+				new TestPojo(1L, 10L, 3L),
+				new TestPojo(1L, 8L, 3L),
+				new TestPojo(1L, 10L, 1L),
+				new TestPojo(1L, 9L, 0L),
+				new TestPojo(1L, 8L, 2L),
+				new TestPojo(1L, 8L, 4L));
+		
+		input1.coGroup(input2)
+		.where(1).equalTo("b")
+		.sortFirstGroup(0, Order.DESCENDING)
+		.sortSecondGroup("c", Order.ASCENDING).sortSecondGroup("a", Order.DESCENDING)
+		
+		.with(new ValidatingCoGroup())
+		.output(new DiscardingOutputFormat<NullValue>());
+		
+		env.execute();
+	}
+	
+	
+	private static class ValidatingCoGroup implements CoGroupFunction<Tuple2<Long, Long>, TestPojo, NullValue> {
+
+		@Override
+		public void coGroup(Iterable<Tuple2<Long, Long>> first, Iterable<TestPojo> second, Collector<NullValue> out) throws Exception {
+			// validate the tuple input, field 1, descending
+			{
+				long lastValue = Long.MAX_VALUE;
+				
+				for (Tuple2<Long, Long> t : first) {
+					long current = t.f1;
+					Assert.assertTrue(current <= lastValue);
+					lastValue = current;
+				}
+			}
+			
+			
+			// validate the pojo input
+			{
+				TestPojo lastValue = new TestPojo(Long.MAX_VALUE, 0, Long.MIN_VALUE);
+				
+				for (TestPojo current : second) {
+					Assert.assertTrue(current.c >= lastValue.c);
+					Assert.assertTrue(current.c != lastValue.c || current.a <= lastValue.a);
+					
+					lastValue = current;
+				}
+			}
+			
+		}
+	}
+	
+	public static class TestPojo implements Cloneable {
+		public long a;
+		public long b;
+		public long c;
+		
+		
+		public TestPojo() {}
+		
+		public TestPojo(long a, long b, long c) {
+			this.a = a;
+			this.b = b;
+			this.c = c;
+		}
+	}
+}

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/MapPartitionITCase.java

+/*
+ * 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.test.javaApiOperators;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+
+@SuppressWarnings("serial")
+public class MapPartitionITCase extends JavaProgramTestBase {
+
+	private static final String IN = "1 1\n2 2\n2 8\n4 4\n4 4\n6 6\n7 7\n8 8\n"
+			+ "1 1\n2 2\n2 2\n4 4\n4 4\n6 3\n5 9\n8 8\n1 1\n2 2\n2 2\n3 0\n4 4\n"
+			+ "5 9\n7 7\n8 8\n1 1\n9 1\n5 9\n4 4\n4 4\n6 6\n7 7\n8 8\n";
+
+	private static final String RESULT = "1 11\n2 12\n4 14\n4 14\n1 11\n2 12\n2 12\n4 14\n4 14\n3 16\n1 11\n2 12\n2 12\n0 13\n4 14\n1 11\n4 14\n4 14\n";
+	
+	
+	private List<Tuple2<String, String>> input = new ArrayList<Tuple2<String,String>>();
+	
+	private List<Tuple2<String, Integer>> expected = new ArrayList<Tuple2<String,Integer>>();
+	
+	private List<Tuple2<String, Integer>> result = new ArrayList<Tuple2<String,Integer>>();
+	
+
+	@Override
+	protected void preSubmit() throws Exception {
+
+		// create input
+		for (String s :IN.split("\n")) {
+			String[] fields = s.split(" ");
+			input.add(new Tuple2<String, String>(fields[0], fields[1]));
+		}
+		
+		// create expected
+		for (String s : RESULT.split("\n")) {
+			String[] fields = s.split(" ");
+			expected.add(new Tuple2<String, Integer>(fields[0], Integer.parseInt(fields[1])));
+		}
+		
+	}
+	
+	@Override
+	protected void postSubmit() {
+		compareResultCollections(expected, result, new TupleComparator<Tuple2<String, Integer>>());
+	}
+
+	@Override
+	protected void testProgram() throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		
+		DataSet<Tuple2<String, String>> data = env.fromCollection(input);
+		
+		data.mapPartition(new TestMapPartition()).output(new LocalCollectionOutputFormat<Tuple2<String,Integer>>(result));
+		
+		env.execute();
+	}
+	
+	
+	public static class TestMapPartition implements MapPartitionFunction<Tuple2<String, String>, Tuple2<String, Integer>> {
+
+		@Override
+		public void mapPartition(Iterable<Tuple2<String, String>> values, Collector<Tuple2<String, Integer>> out) {
+			for (Tuple2<String, String> value : values) {
+				String keyString = value.f0;
+				String valueString = value.f1;
+				
+				int keyInt = Integer.parseInt(keyString);
+				int valueInt = Integer.parseInt(valueString);
+
+				if (keyInt + valueInt < 10) {
+					out.collect(new Tuple2<String, Integer>(valueString, keyInt + 10));
+				}
+			}
+		}
+	}
+}

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/ObjectReuseITCase.java

+/*
+ * 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.test.javaApiOperators;
+
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+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.configuration.Configuration;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * These check whether the object-reuse execution mode does really reuse objects.
+ */
+@RunWith(Parameterized.class)
+public class ObjectReuseITCase extends JavaProgramTestBase {
+
+	private static int NUM_PROGRAMS = 4;
+
+	private int curProgId = config.getInteger("ProgramId", -1);
+	private String resultPath;
+	private String expectedResult;
+
+	private static String inReducePath;
+	private static String inGroupReducePath;
+
+	private String IN_REDUCE = "a,1\na,2\na,3\na,4\na,50\n";
+	private String IN_GROUP_REDUCE = "a,1\na,2\na,3\na,4\na,5\n";
+
+	public ObjectReuseITCase(Configuration config) {
+		super(config);
+	}
+	
+	@Override
+	protected void preSubmit() throws Exception {
+		inReducePath = createTempFile("in_reduce.txt", IN_REDUCE);
+		inGroupReducePath = createTempFile("in_group_reduce.txt", IN_GROUP_REDUCE);
+		resultPath = getTempDirPath("result");
+	}
+
+	@Override
+	protected void testProgram() throws Exception {
+		expectedResult = Progs.runProgram(curProgId, resultPath);
+	}
+	
+	@Override
+	protected void postSubmit() throws Exception {
+		compareResultsByLinesInMemory(expectedResult, resultPath);
+	}
+
+	@Override
+	protected boolean skipCollectionExecution() {
+		return true;
+	}
+
+	
+	@Parameters
+	public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
+
+		LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+
+		for(int i=1; i <= NUM_PROGRAMS; i++) {
+			Configuration config = new Configuration();
+			config.setInteger("ProgramId", i);
+			tConfigs.add(config);
+		}
+		
+		return toParameterList(tConfigs);
+	}
+	
+	@SuppressWarnings({"unchecked", "serial"})
+	private static class Progs {
+		
+		public static String runProgram(int progId, String resultPath) throws Exception {
+			
+			switch(progId) {
+
+			case 1: {
+
+				final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+				env.getConfig().enableObjectReuse();
+
+				DataSet<Tuple2<String, Integer>> input = env.readCsvFile(inReducePath).types(String.class, Integer.class).setParallelism(1);
+				DataSet<Tuple2<String, Integer>> result = input.groupBy(0).reduce(new ReduceFunction<Tuple2<String, Integer>>() {
+
+					@Override
+					public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws
+							Exception {
+						value2.f1 += value1.f1;
+						return value2;
+					}
+
+				});
+
+				result.writeAsCsv(resultPath);
+				env.execute();
+
+				// return expected result
+				return "a,100\n";
+
+			}
+
+			case 2: {
+
+				final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+				env.getConfig().enableObjectReuse();
+
+				DataSet<Tuple2<String, Integer>> input = env.readCsvFile(inReducePath).types(String.class, Integer.class).setParallelism(1);
+
+				DataSet<Tuple2<String, Integer>> result = input
+						.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
+
+							@Override
+							public Tuple2<String, Integer> reduce(
+									Tuple2<String, Integer> value1,
+									Tuple2<String, Integer> value2) throws Exception {
+								value2.f1 += value1.f1;
+								return value2;
+							}
+
+						});
+
+				result.writeAsCsv(resultPath);
+				env.execute();
+
+				// return expected result
+				return "a,100\n";
+
+			}
+
+			case 3: {
+
+				final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+				env.getConfig().enableObjectReuse();
+
+				DataSet<Tuple2<String, Integer>> input = env.readCsvFile(inGroupReducePath).types(String.class, Integer.class).setParallelism(1);
+
+				DataSet<Tuple2<String, Integer>> result = input.reduceGroup(new GroupReduceFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
+
+					@Override
+					public void reduce(Iterable<Tuple2<String, Integer>> values, Collector<Tuple2<String, Integer>> out) throws Exception {
+						List<Tuple2<String, Integer>> list = new ArrayList<Tuple2<String, Integer>>();
+						for (Tuple2<String, Integer> val : values) {
+							list.add(val);
+						}
+
+						for (Tuple2<String, Integer> val : list) {
+							out.collect(val);
+						}
+					}
+
+				});
+
+				result.writeAsCsv(resultPath);
+				env.execute();
+
+				// return expected result
+				return "a,4\n" +
+						"a,4\n" +
+						"a,5\n" +
+						"a,5\n" +
+						"a,5\n";
+
+			}
+
+			case 4: {
+
+				final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+				env.getConfig().enableObjectReuse();
+
+				DataSet<Tuple2<String, Integer>> input = env.readCsvFile(inGroupReducePath).types(String.class, Integer.class).setParallelism(1);
+
+				DataSet<Tuple2<String, Integer>> result = input.reduceGroup(new GroupReduceFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
+
+					@Override
+					public void reduce(Iterable<Tuple2<String, Integer>> values, Collector<Tuple2<String, Integer>> out) throws Exception {
+						List<Tuple2<String, Integer>> list = new ArrayList<Tuple2<String, Integer>>();
+						for (Tuple2<String, Integer> val : values) {
+							list.add(val);
+						}
+
+						for (Tuple2<String, Integer> val : list) {
+							out.collect(val);
+						}
+					}
+
+				});
+
+				result.writeAsCsv(resultPath);
+				env.execute();
+
+				// return expected result
+				return "a,4\n" +
+						"a,4\n" +
+						"a,5\n" +
+						"a,5\n" +
+						"a,5\n";
+
+			}
+
+			default:
+				throw new IllegalArgumentException("Invalid program id");
+			}
+			
+		}
+	
+	}
+}