@Test
public void testDeepNesting() throws Exception {
	/*
	 * Deep nesting test
	 * + null value in pojo
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	DataSet<CrazyNested> ds = CollectionDataSets.getCrazyNestedDataSet(env);
	DataSet<Tuple2<String, Integer>> reduceDs = ds.groupBy("nestLvl1.nestLvl2.nestLvl3.nestLvl4.f1nal")
			.reduceGroup(new GroupReducer1());

	List<Tuple2<String, Integer>> result = reduceDs.collect();

	String expected = "aa,1\nbb,2\ncc,3\n";

	compareResultAsTuples(result, expected);
}
 

@Test
public void testTupleContainingPojosAndRegularFields() throws Exception {
	/*
	 * Test Tuple containing pojos and regular fields
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	DataSet<Tuple3<Integer, CrazyNested, POJO>> ds = CollectionDataSets.getTupleContainingPojos(env);

	DataSet<Integer> reduceDs = ds.groupBy("f0", "f1.*") // nested full tuple selection
			.reduceGroup(new GroupReducer4());

	List<Integer> result = reduceDs.collect();

	String expected = "3\n1\n";

	compareResultAsText(result, expected);
}
 

@Test
public void testDeepNesting() throws Exception {
	/*
	 * Deep nesting test
	 * + null value in pojo
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	DataSet<CrazyNested> ds = CollectionDataSets.getCrazyNestedDataSet(env);
	DataSet<Tuple2<String, Integer>> reduceDs = ds.groupBy("nestLvl1.nestLvl2.nestLvl3.nestLvl4.f1nal")
			.reduceGroup(new GroupReducer1());

	List<Tuple2<String, Integer>> result = reduceDs.collect();

	String expected = "aa,1\nbb,2\ncc,3\n";

	compareResultAsTuples(result, expected);
}
 

@Test
public void testTupleContainingPojosAndRegularFields() throws Exception {
	/*
	 * Test Tuple containing pojos and regular fields
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	DataSet<Tuple3<Integer, CrazyNested, POJO>> ds = CollectionDataSets.getTupleContainingPojos(env);

	DataSet<Integer> reduceDs = ds.groupBy("f0", "f1.*") // nested full tuple selection
			.reduceGroup(new GroupReducer4());

	List<Integer> result = reduceDs.collect();

	String expected = "3\n1\n";

	compareResultAsText(result, expected);
}
 

@Test
public void testDeepNesting() throws Exception {
	/*
	 * Deep nesting test
	 * + null value in pojo
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	DataSet<CrazyNested> ds = CollectionDataSets.getCrazyNestedDataSet(env);
	DataSet<Tuple2<String, Integer>> reduceDs = ds.groupBy("nestLvl1.nestLvl2.nestLvl3.nestLvl4.f1nal")
			.reduceGroup(new GroupReducer1());

	List<Tuple2<String, Integer>> result = reduceDs.collect();

	String expected = "aa,1\nbb,2\ncc,3\n";

	compareResultAsTuples(result, expected);
}
 

@Test
public void testTupleContainingPojosAndRegularFields() throws Exception {
	/*
	 * Test Tuple containing pojos and regular fields
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	DataSet<Tuple3<Integer, CrazyNested, POJO>> ds = CollectionDataSets.getTupleContainingPojos(env);

	DataSet<Integer> reduceDs = ds.groupBy("f0", "f1.*") // nested full tuple selection
			.reduceGroup(new GroupReducer4());

	List<Integer> result = reduceDs.collect();

	String expected = "3\n1\n";

	compareResultAsText(result, expected);
}
 

@Override
public void reduce(Iterable<CrazyNested> values,
		Collector<Tuple2<String, Integer>> out)
				throws Exception {
	int c = 0; String n = null;
	for (CrazyNested v : values) {
		c++; // haha
		n = v.nestLvl1.nestLvl2.nestLvl3.nestLvl4.f1nal;
	}
	out.collect(new Tuple2<>(n, c));
}
 

@Override
public void reduce(Iterable<CrazyNested> values,
		Collector<Tuple2<String, Integer>> out)
				throws Exception {
	int c = 0; String n = null;
	for (CrazyNested v : values) {
		c++; // haha
		n = v.nestLvl1.nestLvl2.nestLvl3.nestLvl4.f1nal;
	}
	out.collect(new Tuple2<>(n, c));
}
 

@Override
public void reduce(Iterable<CrazyNested> values,
		Collector<Tuple2<String, Integer>> out)
				throws Exception {
	int c = 0; String n = null;
	for (CrazyNested v : values) {
		c++; // haha
		n = v.nestLvl1.nestLvl2.nestLvl3.nestLvl4.f1nal;
	}
	out.collect(new Tuple2<>(n, c));
}
 

@Override
public void reduce(Iterable<Tuple3<Integer, CrazyNested, POJO>> values, Collector<Integer> out) {
	out.collect(countElements(values));
}
 

@Override
public void reduce(Iterable<Tuple3<Integer, CrazyNested, POJO>> values, Collector<Integer> out) {
	out.collect(countElements(values));
}
 

@Override
public void reduce(Iterable<Tuple3<Integer, CrazyNested, POJO>> values, Collector<Integer> out) {
	out.collect(countElements(values));
}