flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [02/10] flink git commit: [FLINK-2725] Add Max/Min/Sum aggregation for mutable types.
Date Mon, 19 Oct 2015 16:01:08 GMT
[FLINK-2725] Add Max/Min/Sum aggregation for mutable types.

This closes #1191


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/da248b15
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/da248b15
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/da248b15

Branch: refs/heads/master
Commit: da248b15e1b1dbe09345d3bb186dc815a45e9a3c
Parents: 6491559
Author: Greg Hogan <code@greghogan.com>
Authored: Tue Sep 22 13:01:47 2015 -0400
Committer: Fabian Hueske <fhueske@apache.org>
Committed: Mon Oct 19 15:39:28 2015 +0200

----------------------------------------------------------------------
 .../aggregation/MaxAggregationFunction.java     |  83 ++-
 .../aggregation/MinAggregationFunction.java     |  85 ++-
 .../aggregation/SumAggregationFunction.java     | 190 ++++-
 .../flink/api/java/typeutils/ValueTypeInfo.java |  13 +-
 .../test/javaApiOperators/AggregateITCase.java  |  71 ++
 .../util/ValueCollectionDataSets.java           | 730 +++++++++++++++++++
 6 files changed, 1110 insertions(+), 62 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/da248b15/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MaxAggregationFunction.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MaxAggregationFunction.java b/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MaxAggregationFunction.java
index f25ca87..59d9e13 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MaxAggregationFunction.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MaxAggregationFunction.java
@@ -18,35 +18,74 @@
 
 package org.apache.flink.api.java.aggregation;
 
+import org.apache.flink.types.CopyableValue;
+import org.apache.flink.types.ResettableValue;
 
-public class MaxAggregationFunction<T extends Comparable<T>> extends AggregationFunction<T> {
-	private static final long serialVersionUID = 1L;
 
-	private T value;
+public abstract class MaxAggregationFunction<T extends Comparable<T>> extends AggregationFunction<T> {
+	private static final long serialVersionUID = 1L;
 
 	@Override
-	public void initializeAggregate() {
-		value = null;
+	public String toString() {
+		return "MAX";
 	}
 
-	@Override
-	public void aggregate(T val) {
-		if (value != null) {
-			int cmp = value.compareTo(val);
-			value = (cmp > 0) ? value : val;
-		} else {
-			value = val;
+	// --------------------------------------------------------------------------------------------
+
+	public static final class ImmutableMaxAgg<U extends Comparable<U>> extends MaxAggregationFunction<U> {
+		private static final long serialVersionUID = 1L;
+
+		private U value;
+
+		@Override
+		public void initializeAggregate() {
+			value = null;
 		}
-	}
 
-	@Override
-	public T getAggregate() {
-		return value;
+		@Override
+		public void aggregate(U val) {
+			if (value != null) {
+				int cmp = value.compareTo(val);
+				value = (cmp > 0) ? value : val;
+			} else {
+				value = val;
+			}
+		}
+
+		@Override
+		public U getAggregate() {
+			return value;
+		}
 	}
 	
-	@Override
-	public String toString() {
-		return "MAX";
+	// --------------------------------------------------------------------------------------------
+
+	public static final class MutableMaxAgg<U extends Comparable<U> & ResettableValue<U> & CopyableValue<U>> extends MaxAggregationFunction<U> {
+		private static final long serialVersionUID = 1L;
+
+		private U value;
+
+		@Override
+		public void initializeAggregate() {
+			value = null;
+		}
+
+		@Override
+		public void aggregate(U val) {
+			if (value != null) {
+				int cmp = value.compareTo(val);
+				if (cmp < 0) {
+					value.setValue(val);
+				}
+			} else {
+				value = val.copy();
+			}
+		}
+
+		@Override
+		public U getAggregate() {
+			return value;
+		}
 	}
 	
 	// --------------------------------------------------------------------------------------------
@@ -58,7 +97,11 @@ public class MaxAggregationFunction<T extends Comparable<T>> extends Aggregation
 		@Override
 		public <T> AggregationFunction<T> createAggregationFunction(Class<T> type) {
 			if (Comparable.class.isAssignableFrom(type)) {
-				return (AggregationFunction<T>) new MaxAggregationFunction();
+				if (ResettableValue.class.isAssignableFrom(type) & CopyableValue.class.isAssignableFrom(type)) {
+					return (AggregationFunction<T>) new MutableMaxAgg();
+				} else {
+					return (AggregationFunction<T>) new ImmutableMaxAgg();
+				}
 			} else {
 				throw new UnsupportedAggregationTypeException("The type " + type.getName() + 
 					" is not supported for maximum aggregation. " +

http://git-wip-us.apache.org/repos/asf/flink/blob/da248b15/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MinAggregationFunction.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MinAggregationFunction.java b/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MinAggregationFunction.java
index faf28a7..b72b0f4 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MinAggregationFunction.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/aggregation/MinAggregationFunction.java
@@ -18,35 +18,74 @@
 
 package org.apache.flink.api.java.aggregation;
 
+import org.apache.flink.types.CopyableValue;
+import org.apache.flink.types.ResettableValue;
 
-public class MinAggregationFunction<T extends Comparable<T>> extends AggregationFunction<T> {
-	private static final long serialVersionUID = 1L;
 
-	private T value;
+public abstract class MinAggregationFunction<T extends Comparable<T>> extends AggregationFunction<T> {
+	private static final long serialVersionUID = 1L;
 
 	@Override
-	public void initializeAggregate() {
-		value = null;
+	public String toString() {
+		return "MIN";
 	}
 
-	@Override
-	public void aggregate(T val) {
-		if (value != null) {
-			int cmp = value.compareTo(val);
-			value = (cmp < 0) ? value : val;
-		} else {
-			value = val;
+	// --------------------------------------------------------------------------------------------
+
+	public static final class ImmutableMinAgg<U extends Comparable<U>> extends MinAggregationFunction<U> {
+		private static final long serialVersionUID = 1L;
+
+		private U value;
+
+		@Override
+		public void initializeAggregate() {
+			value = null;
 		}
-	}
 
-	@Override
-	public T getAggregate() {
-		return value;
+		@Override
+		public void aggregate(U val) {
+			if (value != null) {
+				int cmp = value.compareTo(val);
+				value = (cmp < 0) ? value : val;
+			} else {
+				value = val;
+			}
+		}
+
+		@Override
+		public U getAggregate() {
+			return value;
+		}
 	}
-	
-	@Override
-	public String toString() {
-		return "MIN";
+
+	// --------------------------------------------------------------------------------------------
+
+	public static final class MutableMinAgg<U extends Comparable<U> & ResettableValue<U> & CopyableValue<U>> extends MinAggregationFunction<U> {
+		private static final long serialVersionUID = 1L;
+
+		private U value;
+
+		@Override
+		public void initializeAggregate() {
+			value = null;
+		}
+
+		@Override
+		public void aggregate(U val) {
+			if (value != null) {
+				int cmp = value.compareTo(val);
+				if (cmp > 0) {
+					value.setValue(val);
+				}
+			} else {
+				value = val.copy();
+			}
+		}
+
+		@Override
+		public U getAggregate() {
+			return value;
+		}
 	}
 	
 	// --------------------------------------------------------------------------------------------
@@ -58,7 +97,11 @@ public class MinAggregationFunction<T extends Comparable<T>> extends Aggregation
 		@Override
 		public <T> AggregationFunction<T> createAggregationFunction(Class<T> type) {
 			if (Comparable.class.isAssignableFrom(type)) {
-				return (AggregationFunction<T>) new MinAggregationFunction();
+				if (ResettableValue.class.isAssignableFrom(type) & CopyableValue.class.isAssignableFrom(type)) {
+					return (AggregationFunction<T>) new MutableMinAgg();
+				} else {
+					return (AggregationFunction<T>) new ImmutableMinAgg();
+				}
 			} else {
 				throw new UnsupportedAggregationTypeException("The type " + type.getName() + 
 					" is not supported for minimum aggregation. " +

http://git-wip-us.apache.org/repos/asf/flink/blob/da248b15/flink-java/src/main/java/org/apache/flink/api/java/aggregation/SumAggregationFunction.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/aggregation/SumAggregationFunction.java b/flink-java/src/main/java/org/apache/flink/api/java/aggregation/SumAggregationFunction.java
index 24e8f31..ad4644b 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/aggregation/SumAggregationFunction.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/aggregation/SumAggregationFunction.java
@@ -18,21 +18,27 @@
 
 package org.apache.flink.api.java.aggregation;
 
+import org.apache.flink.types.ByteValue;
+import org.apache.flink.types.DoubleValue;
+import org.apache.flink.types.FloatValue;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.ShortValue;
 
 public abstract class SumAggregationFunction<T> extends AggregationFunction<T> {
-	
+
 	private static final long serialVersionUID = 1L;
 
 	@Override
 	public String toString() {
 		return "SUM";
 	}
-	
+
 	// --------------------------------------------------------------------------------------------
-	
+
 	public static final class ByteSumAgg extends SumAggregationFunction<Byte> {
 		private static final long serialVersionUID = 1L;
-		
+
 		private long agg;
 
 		@Override
@@ -50,10 +56,31 @@ public abstract class SumAggregationFunction<T> extends AggregationFunction<T> {
 			return (byte) agg;
 		}
 	}
-	
+
+	public static final class ByteValueSumAgg extends SumAggregationFunction<ByteValue> {
+		private static final long serialVersionUID = 1L;
+
+		private long agg;
+
+		@Override
+		public void initializeAggregate() {
+			agg = 0;
+		}
+
+		@Override
+		public void aggregate(ByteValue value) {
+			agg += value.getValue();
+		}
+
+		@Override
+		public ByteValue getAggregate() {
+			return new ByteValue((byte) agg);
+		}
+	}
+
 	public static final class ShortSumAgg extends SumAggregationFunction<Short> {
 		private static final long serialVersionUID = 1L;
-		
+
 		private long agg;
 
 		@Override
@@ -71,10 +98,31 @@ public abstract class SumAggregationFunction<T> extends AggregationFunction<T> {
 			return (short) agg;
 		}
 	}
-	
+
+	public static final class ShortValueSumAgg extends SumAggregationFunction<ShortValue> {
+		private static final long serialVersionUID = 1L;
+
+		private long agg;
+
+		@Override
+		public void initializeAggregate() {
+			agg = 0;
+		}
+
+		@Override
+		public void aggregate(ShortValue value) {
+			agg += value.getValue();
+		}
+
+		@Override
+		public ShortValue getAggregate() {
+			return new ShortValue((short) agg);
+		}
+	}
+
 	public static final class IntSumAgg extends SumAggregationFunction<Integer> {
 		private static final long serialVersionUID = 1L;
-		
+
 		private long agg;
 
 		@Override
@@ -92,10 +140,31 @@ public abstract class SumAggregationFunction<T> extends AggregationFunction<T> {
 			return (int) agg;
 		}
 	}
-	
+
+	public static final class IntValueSumAgg extends SumAggregationFunction<IntValue> {
+		private static final long serialVersionUID = 1L;
+
+		private long agg;
+
+		@Override
+		public void initializeAggregate() {
+			agg = 0;
+		}
+
+		@Override
+		public void aggregate(IntValue value) {
+			agg += value.getValue();
+		}
+
+		@Override
+		public IntValue getAggregate() {
+			return new IntValue((int) agg);
+		}
+	}
+
 	public static final class LongSumAgg extends SumAggregationFunction<Long> {
 		private static final long serialVersionUID = 1L;
-		
+
 		private long agg;
 
 		@Override
@@ -113,11 +182,32 @@ public abstract class SumAggregationFunction<T> extends AggregationFunction<T> {
 			return agg;
 		}
 	}
-	
+
+	public static final class LongValueSumAgg extends SumAggregationFunction<LongValue> {
+		private static final long serialVersionUID = 1L;
+
+		private long agg;
+
+		@Override
+		public void initializeAggregate() {
+			agg = 0L;
+		}
+
+		@Override
+		public void aggregate(LongValue value) {
+			agg += value.getValue();
+		}
+
+		@Override
+		public LongValue getAggregate() {
+			return new LongValue(agg);
+		}
+	}
+
 	public static final class FloatSumAgg extends SumAggregationFunction<Float> {
 		private static final long serialVersionUID = 1L;
-		
-		private float agg;
+
+		private double agg;
 
 		@Override
 		public void initializeAggregate() {
@@ -131,13 +221,34 @@ public abstract class SumAggregationFunction<T> extends AggregationFunction<T> {
 
 		@Override
 		public Float getAggregate() {
-			return agg;
+			return (float) agg;
+		}
+	}
+
+	public static final class FloatValueSumAgg extends SumAggregationFunction<FloatValue> {
+		private static final long serialVersionUID = 1L;
+
+		private double agg;
+
+		@Override
+		public void initializeAggregate() {
+			agg = 0.0f;
+		}
+
+		@Override
+		public void aggregate(FloatValue value) {
+			agg += value.getValue();
+		}
+
+		@Override
+		public FloatValue getAggregate() {
+			return new FloatValue((float) agg);
 		}
 	}
-	
+
 	public static final class DoubleSumAgg extends SumAggregationFunction<Double> {
 		private static final long serialVersionUID = 1L;
-		
+
 		private double agg;
 
 		@Override
@@ -155,36 +266,75 @@ public abstract class SumAggregationFunction<T> extends AggregationFunction<T> {
 			return agg;
 		}
 	}
-	
+
+	public static final class DoubleValueSumAgg extends SumAggregationFunction<DoubleValue> {
+		private static final long serialVersionUID = 1L;
+
+		private double agg;
+
+		@Override
+		public void initializeAggregate() {
+			agg = 0.0;
+		}
+
+		@Override
+		public void aggregate(DoubleValue value) {
+			agg += value.getValue();
+		}
+
+		@Override
+		public DoubleValue getAggregate() {
+			return new DoubleValue(agg);
+		}
+	}
+
 	// --------------------------------------------------------------------------------------------
-	
+
 	public static final class SumAggregationFunctionFactory implements AggregationFunctionFactory {
 		private static final long serialVersionUID = 1L;
-		
+
 		@SuppressWarnings("unchecked")
 		@Override
 		public <T> AggregationFunction<T> createAggregationFunction(Class<T> type) {
 			if (type == Long.class) {
 				return (AggregationFunction<T>) new LongSumAgg();
 			}
+			else if (type == LongValue.class) {
+				return (AggregationFunction<T>) new LongValueSumAgg();
+			}
 			else if (type == Integer.class) {
 				return (AggregationFunction<T>) new IntSumAgg();
 			}
+			else if (type == IntValue.class) {
+				return (AggregationFunction<T>) new IntValueSumAgg();
+			}
 			else if (type == Double.class) {
 				return (AggregationFunction<T>) new DoubleSumAgg();
 			}
+			else if (type == DoubleValue.class) {
+				return (AggregationFunction<T>) new DoubleValueSumAgg();
+			}
 			else if (type == Float.class) {
 				return (AggregationFunction<T>) new FloatSumAgg();
 			}
+			else if (type == FloatValue.class) {
+				return (AggregationFunction<T>) new FloatValueSumAgg();
+			}
 			else if (type == Byte.class) {
 				return (AggregationFunction<T>) new ByteSumAgg();
 			}
+			else if (type == ByteValue.class) {
+				return (AggregationFunction<T>) new ByteValueSumAgg();
+			}
 			else if (type == Short.class) {
 				return (AggregationFunction<T>) new ShortSumAgg();
 			}
+			else if (type == ShortValue.class) {
+				return (AggregationFunction<T>) new ShortValueSumAgg();
+			}
 			else {
 				throw new UnsupportedAggregationTypeException("The type " + type.getName() + 
-					" has currently not supported for built-in sum aggregations.");
+					" is currently not supported for built-in sum aggregations.");
 			}
 		}
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/da248b15/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java
index 0b4823e..5187de7 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java
@@ -51,7 +51,18 @@ import org.apache.flink.types.Value;
 public class ValueTypeInfo<T extends Value> extends TypeInformation<T> implements AtomicType<T> {
 
 	private static final long serialVersionUID = 1L;
-	
+
+	public static final ValueTypeInfo<BooleanValue> BOOLEAN_VALUE_TYPE_INFO = new ValueTypeInfo<>(BooleanValue.class);
+	public static final ValueTypeInfo<ByteValue> BYTE_VALUE_TYPE_INFO = new ValueTypeInfo<>(ByteValue.class);
+	public static final ValueTypeInfo<CharValue> CHAR_VALUE_TYPE_INFO = new ValueTypeInfo<>(CharValue.class);
+	public static final ValueTypeInfo<DoubleValue> DOUBLE_VALUE_TYPE_INFO = new ValueTypeInfo<>(DoubleValue.class);
+	public static final ValueTypeInfo<FloatValue> FLOAT_VALUE_TYPE_INFO = new ValueTypeInfo<>(FloatValue.class);
+	public static final ValueTypeInfo<IntValue> INT_VALUE_TYPE_INFO = new ValueTypeInfo<>(IntValue.class);
+	public static final ValueTypeInfo<LongValue> LONG_VALUE_TYPE_INFO = new ValueTypeInfo<>(LongValue.class);
+	public static final ValueTypeInfo<NullValue> NULL_VALUE_TYPE_INFO = new ValueTypeInfo<>(NullValue.class);
+	public static final ValueTypeInfo<ShortValue> SHORT_VALUE_TYPE_INFO = new ValueTypeInfo<>(ShortValue.class);
+	public static final ValueTypeInfo<StringValue> STRING_VALUE_TYPE_INFO = new ValueTypeInfo<>(StringValue.class);
+
 	private final Class<T> type;
 	
 	public ValueTypeInfo(Class<T> type) {

http://git-wip-us.apache.org/repos/asf/flink/blob/da248b15/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/AggregateITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/AggregateITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/AggregateITCase.java
index d02f228..fc01ce7 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/AggregateITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/AggregateITCase.java
@@ -25,7 +25,11 @@ import org.apache.flink.api.java.tuple.Tuple1;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
+import org.apache.flink.test.javaApiOperators.util.ValueCollectionDataSets;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.StringValue;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -62,6 +66,27 @@ public class AggregateITCase extends MultipleProgramsTestBase {
 	}
 
 	@Test
+	public void testFullAggregateOfMutableValueTypes() throws Exception {
+		/*
+		 * Full Aggregate of mutable value types
+		 */
+
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+		DataSet<Tuple3<IntValue, LongValue, StringValue>> ds = ValueCollectionDataSets.get3TupleDataSet(env);
+		DataSet<Tuple2<IntValue, LongValue>> aggregateDs = ds
+				.aggregate(Aggregations.SUM, 0)
+				.and(Aggregations.MAX, 1)
+				.project(0, 1);
+
+		List<Tuple2<IntValue, LongValue>> result = aggregateDs.collect();
+
+		String expected = "231,6\n";
+
+		compareResultAsTuples(result, expected);
+	}
+
+	@Test
 	public void testGroupedAggregate() throws Exception {
 		/*
 		 * Grouped Aggregate
@@ -87,6 +112,31 @@ public class AggregateITCase extends MultipleProgramsTestBase {
 	}
 
 	@Test
+	public void testGroupedAggregateOfMutableValueTypes() throws Exception {
+		/*
+		 * Grouped Aggregate of mutable value types
+		 */
+
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+		DataSet<Tuple3<IntValue, LongValue, StringValue>> ds = ValueCollectionDataSets.get3TupleDataSet(env);
+		DataSet<Tuple2<IntValue, LongValue>> aggregateDs = ds.groupBy(1)
+				.aggregate(Aggregations.SUM, 0)
+				.project(1, 0);
+
+		List<Tuple2<IntValue, LongValue>> result = aggregateDs.collect();
+
+		String expected = "1,1\n" +
+				"2,5\n" +
+				"3,15\n" +
+				"4,34\n" +
+				"5,65\n" +
+				"6,111\n";
+
+		compareResultAsTuples(result, expected);
+	}
+
+	@Test
 	public void testNestedAggregate() throws Exception {
 		/*
 		 * Nested Aggregate
@@ -106,4 +156,25 @@ public class AggregateITCase extends MultipleProgramsTestBase {
 
 		compareResultAsTuples(result, expected);
 	}
+
+	@Test
+	public void testNestedAggregateOfMutableValueTypes() throws Exception {
+		/*
+		 * Nested Aggregate of mutable value types
+		 */
+
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+		DataSet<Tuple3<IntValue, LongValue, StringValue>> ds = ValueCollectionDataSets.get3TupleDataSet(env);
+		DataSet<Tuple1<IntValue>> aggregateDs = ds.groupBy(1)
+				.aggregate(Aggregations.MIN, 0)
+				.aggregate(Aggregations.MIN, 0)
+				.project(0);
+
+		List<Tuple1<IntValue>> result = aggregateDs.collect();
+
+		String expected = "1\n";
+
+		compareResultAsTuples(result, expected);
+	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/da248b15/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/ValueCollectionDataSets.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/ValueCollectionDataSets.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/ValueCollectionDataSets.java
new file mode 100644
index 0000000..04a7bc5
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/ValueCollectionDataSets.java
@@ -0,0 +1,730 @@
+/*
+ * 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.util;
+
+import java.io.File;
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.GregorianCalendar;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.api.java.tuple.Tuple7;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.ValueTypeInfo;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.StringValue;
+import org.apache.hadoop.io.IntWritable;
+
+import scala.math.BigInt;
+
+/**
+ * #######################################################################################################
+ * 
+ * 			BE AWARE THAT OTHER TESTS DEPEND ON THIS TEST DATA. 
+ * 			IF YOU MODIFY THE DATA MAKE SURE YOU CHECK THAT ALL TESTS ARE STILL WORKING!
+ * 
+ * #######################################################################################################
+ */
+public class ValueCollectionDataSets {
+
+	public static DataSet<Tuple3<IntValue, LongValue, StringValue>> get3TupleDataSet(ExecutionEnvironment env) {
+		List<Tuple3<IntValue, LongValue, StringValue>> data = new ArrayList<>();
+
+		data.add(new Tuple3<>(new IntValue(1), new LongValue(1l), new StringValue("Hi")));
+		data.add(new Tuple3<>(new IntValue(2), new LongValue(2l), new StringValue("Hello")));
+		data.add(new Tuple3<>(new IntValue(3), new LongValue(2l), new StringValue("Hello world")));
+		data.add(new Tuple3<>(new IntValue(4), new LongValue(3l), new StringValue("Hello world, how are you?")));
+		data.add(new Tuple3<>(new IntValue(5), new LongValue(3l), new StringValue("I am fine.")));
+		data.add(new Tuple3<>(new IntValue(6), new LongValue(3l), new StringValue("Luke Skywalker")));
+		data.add(new Tuple3<>(new IntValue(7), new LongValue(4l), new StringValue("Comment#1")));
+		data.add(new Tuple3<>(new IntValue(8), new LongValue(4l), new StringValue("Comment#2")));
+		data.add(new Tuple3<>(new IntValue(9), new LongValue(4l), new StringValue("Comment#3")));
+		data.add(new Tuple3<>(new IntValue(10), new LongValue(4l), new StringValue("Comment#4")));
+		data.add(new Tuple3<>(new IntValue(11), new LongValue(5l), new StringValue("Comment#5")));
+		data.add(new Tuple3<>(new IntValue(12), new LongValue(5l), new StringValue("Comment#6")));
+		data.add(new Tuple3<>(new IntValue(13), new LongValue(5l), new StringValue("Comment#7")));
+		data.add(new Tuple3<>(new IntValue(14), new LongValue(5l), new StringValue("Comment#8")));
+		data.add(new Tuple3<>(new IntValue(15), new LongValue(5l), new StringValue("Comment#9")));
+		data.add(new Tuple3<>(new IntValue(16), new LongValue(6l), new StringValue("Comment#10")));
+		data.add(new Tuple3<>(new IntValue(17), new LongValue(6l), new StringValue("Comment#11")));
+		data.add(new Tuple3<>(new IntValue(18), new LongValue(6l), new StringValue("Comment#12")));
+		data.add(new Tuple3<>(new IntValue(19), new LongValue(6l), new StringValue("Comment#13")));
+		data.add(new Tuple3<>(new IntValue(20), new LongValue(6l), new StringValue("Comment#14")));
+		data.add(new Tuple3<>(new IntValue(21), new LongValue(6l), new StringValue("Comment#15")));
+
+		Collections.shuffle(data);
+
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<Tuple3<IntValue, LongValue, StringValue>> getSmall3TupleDataSet(ExecutionEnvironment env) {
+		List<Tuple3<IntValue, LongValue, StringValue>> data = new ArrayList<>();
+
+		data.add(new Tuple3<>(new IntValue(1), new LongValue(1l), new StringValue("Hi")));
+		data.add(new Tuple3<>(new IntValue(2), new LongValue(2l), new StringValue("Hello")));
+		data.add(new Tuple3<>(new IntValue(3), new LongValue(2l), new StringValue("Hello world")));
+
+		Collections.shuffle(data);
+
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<Tuple5<IntValue, LongValue, IntValue, StringValue, LongValue>> get5TupleDataSet(ExecutionEnvironment env) {
+		List<Tuple5<IntValue, LongValue, IntValue, StringValue, LongValue>> data = new ArrayList<>();
+
+		data.add(new Tuple5<>(new IntValue(1), new LongValue(1l), new IntValue(0), new StringValue("Hallo"), new LongValue(1l)));
+		data.add(new Tuple5<>(new IntValue(2), new LongValue(2l), new IntValue(1), new StringValue("Hallo Welt"), new LongValue(2l)));
+		data.add(new Tuple5<>(new IntValue(2), new LongValue(3l), new IntValue(2), new StringValue("Hallo Welt wie"), new LongValue(1l)));
+		data.add(new Tuple5<>(new IntValue(3), new LongValue(4l), new IntValue(3), new StringValue("Hallo Welt wie gehts?"), new LongValue(2l)));
+		data.add(new Tuple5<>(new IntValue(3), new LongValue(5l), new IntValue(4), new StringValue("ABC"), new LongValue(2l)));
+		data.add(new Tuple5<>(new IntValue(3), new LongValue(6l), new IntValue(5), new StringValue("BCD"), new LongValue(3l)));
+		data.add(new Tuple5<>(new IntValue(4), new LongValue(7l), new IntValue(6), new StringValue("CDE"), new LongValue(2l)));
+		data.add(new Tuple5<>(new IntValue(4), new LongValue(8l), new IntValue(7), new StringValue("DEF"), new LongValue(1l)));
+		data.add(new Tuple5<>(new IntValue(4), new LongValue(9l), new IntValue(8), new StringValue("EFG"), new LongValue(1l)));
+		data.add(new Tuple5<>(new IntValue(4), new LongValue(10l), new IntValue(9), new StringValue("FGH"), new LongValue(2l)));
+		data.add(new Tuple5<>(new IntValue(5), new LongValue(11l), new IntValue(10), new StringValue("GHI"), new LongValue(1l)));
+		data.add(new Tuple5<>(new IntValue(5), new LongValue(12l), new IntValue(11), new StringValue("HIJ"), new LongValue(3l)));
+		data.add(new Tuple5<>(new IntValue(5), new LongValue(13l), new IntValue(12), new StringValue("IJK"), new LongValue(3l)));
+		data.add(new Tuple5<>(new IntValue(5), new LongValue(14l), new IntValue(13), new StringValue("JKL"), new LongValue(2l)));
+		data.add(new Tuple5<>(new IntValue(5), new LongValue(15l), new IntValue(14), new StringValue("KLM"), new LongValue(2l)));
+
+		Collections.shuffle(data);
+
+		TupleTypeInfo<Tuple5<IntValue, LongValue, IntValue, StringValue, LongValue>> type = new
+			TupleTypeInfo<>(
+				ValueTypeInfo.INT_VALUE_TYPE_INFO,
+				ValueTypeInfo.LONG_VALUE_TYPE_INFO,
+				ValueTypeInfo.INT_VALUE_TYPE_INFO,
+				ValueTypeInfo.STRING_VALUE_TYPE_INFO,
+				ValueTypeInfo.LONG_VALUE_TYPE_INFO
+		);
+
+		return env.fromCollection(data, type);
+	}
+
+	public static DataSet<Tuple5<IntValue, LongValue, IntValue, StringValue, LongValue>> getSmall5TupleDataSet(ExecutionEnvironment env) {
+		List<Tuple5<IntValue, LongValue, IntValue, StringValue, LongValue>> data = new ArrayList<>();
+
+		data.add(new Tuple5<>(new IntValue(1), new LongValue(1l), new IntValue(0), new StringValue("Hallo"), new LongValue(1l)));
+		data.add(new Tuple5<>(new IntValue(2), new LongValue(2l), new IntValue(1), new StringValue("Hallo Welt"), new LongValue(2l)));
+		data.add(new Tuple5<>(new IntValue(2), new LongValue(3l), new IntValue(2), new StringValue("Hallo Welt wie"), new LongValue(1l)));
+
+		Collections.shuffle(data);
+
+		TupleTypeInfo<Tuple5<IntValue, LongValue, IntValue, StringValue, LongValue>> type = new
+			TupleTypeInfo<>(
+				ValueTypeInfo.INT_VALUE_TYPE_INFO,
+				ValueTypeInfo.LONG_VALUE_TYPE_INFO,
+				ValueTypeInfo.INT_VALUE_TYPE_INFO,
+				ValueTypeInfo.STRING_VALUE_TYPE_INFO,
+				ValueTypeInfo.LONG_VALUE_TYPE_INFO
+		);
+
+		return env.fromCollection(data, type);
+	}
+
+	public static DataSet<Tuple2<Tuple2<IntValue, IntValue>, StringValue>> getSmallNestedTupleDataSet(ExecutionEnvironment env) {
+		List<Tuple2<Tuple2<IntValue, IntValue>, StringValue>> data = new ArrayList<>();
+
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(1), new IntValue(1)), new StringValue("one")));
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(2), new IntValue(2)), new StringValue("two")));
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(3), new IntValue(3)), new StringValue("three")));
+
+		TupleTypeInfo<Tuple2<Tuple2<IntValue, IntValue>, StringValue>> type = new
+			TupleTypeInfo<>(
+				new TupleTypeInfo<Tuple2<IntValue, IntValue>>(ValueTypeInfo.INT_VALUE_TYPE_INFO, ValueTypeInfo.INT_VALUE_TYPE_INFO),
+				ValueTypeInfo.STRING_VALUE_TYPE_INFO
+		);
+
+		return env.fromCollection(data, type);
+	}
+
+	public static DataSet<Tuple2<Tuple2<IntValue, IntValue>, StringValue>> getGroupSortedNestedTupleDataSet(ExecutionEnvironment env) {
+		List<Tuple2<Tuple2<IntValue, IntValue>, StringValue>> data = new ArrayList<>();
+
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(1), new IntValue(3)), new StringValue("a")));
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(1), new IntValue(2)), new StringValue("a")));
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(2), new IntValue(1)), new StringValue("a")));
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(2), new IntValue(2)), new StringValue("b")));
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(3), new IntValue(3)), new StringValue("c")));
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(3), new IntValue(6)), new StringValue("c")));
+		data.add(new Tuple2<>(new Tuple2<>(new IntValue(4), new IntValue(9)), new StringValue("c")));
+
+		TupleTypeInfo<Tuple2<Tuple2<IntValue, IntValue>, StringValue>> type = new
+			TupleTypeInfo<>(
+				new TupleTypeInfo<Tuple2<IntValue, IntValue>>(ValueTypeInfo.INT_VALUE_TYPE_INFO, ValueTypeInfo.INT_VALUE_TYPE_INFO),
+				ValueTypeInfo.STRING_VALUE_TYPE_INFO
+		);
+
+		return env.fromCollection(data, type);
+	}
+
+	public static DataSet<Tuple3<Tuple2<IntValue, IntValue>, StringValue, IntValue>> getGroupSortedNestedTupleDataSet2(ExecutionEnvironment env) {
+		List<Tuple3<Tuple2<IntValue, IntValue>, StringValue, IntValue>> data = new ArrayList<>();
+
+		data.add(new Tuple3<>(new Tuple2<IntValue, IntValue>(new IntValue(1), new IntValue(3)), new StringValue("a"), new IntValue(2)));
+		data.add(new Tuple3<>(new Tuple2<IntValue, IntValue>(new IntValue(1), new IntValue(2)), new StringValue("a"), new IntValue(1)));
+		data.add(new Tuple3<>(new Tuple2<IntValue, IntValue>(new IntValue(2), new IntValue(1)), new StringValue("a"), new IntValue(3)));
+		data.add(new Tuple3<>(new Tuple2<IntValue, IntValue>(new IntValue(2), new IntValue(2)), new StringValue("b"), new IntValue(4)));
+		data.add(new Tuple3<>(new Tuple2<IntValue, IntValue>(new IntValue(3), new IntValue(3)), new StringValue("c"), new IntValue(5)));
+		data.add(new Tuple3<>(new Tuple2<IntValue, IntValue>(new IntValue(3), new IntValue(6)), new StringValue("c"), new IntValue(6)));
+		data.add(new Tuple3<>(new Tuple2<IntValue, IntValue>(new IntValue(4), new IntValue(9)), new StringValue("c"), new IntValue(7)));
+
+		TupleTypeInfo<Tuple3<Tuple2<IntValue, IntValue>, StringValue, IntValue>> type = new
+			TupleTypeInfo<>(
+				new TupleTypeInfo<Tuple2<IntValue, IntValue>>(ValueTypeInfo.INT_VALUE_TYPE_INFO, ValueTypeInfo.INT_VALUE_TYPE_INFO),
+				ValueTypeInfo.STRING_VALUE_TYPE_INFO,
+				ValueTypeInfo.INT_VALUE_TYPE_INFO
+		);
+
+		return env.fromCollection(data, type);
+	}
+	
+	public static DataSet<StringValue> getStringDataSet(ExecutionEnvironment env) {
+		List<StringValue> data = new ArrayList<>();
+
+		data.add(new StringValue("Hi"));
+		data.add(new StringValue("Hello"));
+		data.add(new StringValue("Hello world"));
+		data.add(new StringValue("Hello world, how are you?"));
+		data.add(new StringValue("I am fine."));
+		data.add(new StringValue("Luke Skywalker"));
+		data.add(new StringValue("Random comment"));
+		data.add(new StringValue("LOL"));
+
+		Collections.shuffle(data);
+
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<IntValue> getIntDataSet(ExecutionEnvironment env) {
+		List<IntValue> data = new ArrayList<>();
+
+		data.add(new IntValue(1));
+		data.add(new IntValue(2));
+		data.add(new IntValue(2));
+		data.add(new IntValue(3));
+		data.add(new IntValue(3));
+		data.add(new IntValue(3));
+		data.add(new IntValue(4));
+		data.add(new IntValue(4));
+		data.add(new IntValue(4));
+		data.add(new IntValue(4));
+		data.add(new IntValue(5));
+		data.add(new IntValue(5));
+		data.add(new IntValue(5));
+		data.add(new IntValue(5));
+		data.add(new IntValue(5));
+
+		Collections.shuffle(data);
+
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<CustomType> getCustomTypeDataSet(ExecutionEnvironment env) {
+		List<CustomType> data = new ArrayList<CustomType>();
+
+		data.add(new CustomType(1, 0l, "Hi"));
+		data.add(new CustomType(2, 1l, "Hello"));
+		data.add(new CustomType(2, 2l, "Hello world"));
+		data.add(new CustomType(3, 3l, "Hello world, how are you?"));
+		data.add(new CustomType(3, 4l, "I am fine."));
+		data.add(new CustomType(3, 5l, "Luke Skywalker"));
+		data.add(new CustomType(4, 6l, "Comment#1"));
+		data.add(new CustomType(4, 7l, "Comment#2"));
+		data.add(new CustomType(4, 8l, "Comment#3"));
+		data.add(new CustomType(4, 9l, "Comment#4"));
+		data.add(new CustomType(5, 10l, "Comment#5"));
+		data.add(new CustomType(5, 11l, "Comment#6"));
+		data.add(new CustomType(5, 12l, "Comment#7"));
+		data.add(new CustomType(5, 13l, "Comment#8"));
+		data.add(new CustomType(5, 14l, "Comment#9"));
+		data.add(new CustomType(6, 15l, "Comment#10"));
+		data.add(new CustomType(6, 16l, "Comment#11"));
+		data.add(new CustomType(6, 17l, "Comment#12"));
+		data.add(new CustomType(6, 18l, "Comment#13"));
+		data.add(new CustomType(6, 19l, "Comment#14"));
+		data.add(new CustomType(6, 20l, "Comment#15"));
+
+		Collections.shuffle(data);
+
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<CustomType> getSmallCustomTypeDataSet(ExecutionEnvironment env) {
+		List<CustomType> data = new ArrayList<CustomType>();
+
+		data.add(new CustomType(1, 0l, "Hi"));
+		data.add(new CustomType(2, 1l, "Hello"));
+		data.add(new CustomType(2, 2l, "Hello world"));
+
+		Collections.shuffle(data);
+
+		return env.fromCollection(data);
+	}
+
+	public static class CustomType implements Serializable {
+
+		private static final long serialVersionUID = 1L;
+
+		public IntValue myInt;
+		public LongValue myLong;
+		public StringValue myString;
+
+		public CustomType() {
+		}
+
+		public CustomType(int i, long l, String s) {
+			myInt = new IntValue(i);
+			myLong = new LongValue(l);
+			myString = new StringValue(s);
+		}
+
+		@Override
+		public String toString() {
+			return myInt + "," + myLong + "," + myString;
+		}
+	}
+
+	public static class CustomTypeComparator implements Comparator<CustomType> {
+
+		@Override
+		public int compare(CustomType o1, CustomType o2) {
+			int diff = o1.myInt.getValue() - o2.myInt.getValue();
+			if (diff != 0) {
+				return diff;
+			}
+			diff = (int) (o1.myLong.getValue() - o2.myLong.getValue());
+			return diff != 0 ? diff : o1.myString.getValue().compareTo(o2.myString.getValue());
+		}
+
+	}
+
+	public static DataSet<Tuple7<IntValue, StringValue, IntValue, IntValue, LongValue, StringValue, LongValue>> getSmallTuplebasedDataSet(ExecutionEnvironment env) {
+		List<Tuple7<IntValue, StringValue, IntValue, IntValue, LongValue, StringValue, LongValue>> data = new ArrayList<>();
+		
+		data.add(new Tuple7<>(new IntValue(1), new StringValue("First"), new IntValue(10), new IntValue(100), new LongValue(1000L), new StringValue("One"), new LongValue(10000L)));
+		data.add(new Tuple7<>(new IntValue(2), new StringValue("Second"), new IntValue(20), new IntValue(200), new LongValue(2000L), new StringValue("Two"), new LongValue(20000L)));
+		data.add(new Tuple7<>(new IntValue(3), new StringValue("Third"), new IntValue(30), new IntValue(300), new LongValue(3000L), new StringValue("Three"), new LongValue(30000L)));
+
+		return env.fromCollection(data);
+	}
+	
+	public static DataSet<Tuple7<LongValue, IntValue, IntValue, LongValue, StringValue, IntValue, StringValue>> getSmallTuplebasedDataSetMatchingPojo(ExecutionEnvironment env) {
+		List<Tuple7<LongValue, IntValue, IntValue, LongValue, StringValue, IntValue, StringValue>> data = new ArrayList<>();
+		
+		data.add(new Tuple7<>(new LongValue(10000L), new IntValue(10), new IntValue(100), new LongValue(1000L), new StringValue("One"), new IntValue(1), new StringValue("First")));
+		data.add(new Tuple7<>(new LongValue(20000L), new IntValue(20), new IntValue(200), new LongValue(2000L), new StringValue("Two"), new IntValue(2), new StringValue("Second")));
+		data.add(new Tuple7<>(new LongValue(30000L), new IntValue(30), new IntValue(300), new LongValue(3000L), new StringValue("Three"), new IntValue(3), new StringValue("Third")));
+		
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<POJO> getSmallPojoDataSet(ExecutionEnvironment env) {
+		List<POJO> data = new ArrayList<POJO>();
+
+		data.add(new POJO(1 /*number*/, "First" /*str*/, 10 /*f0*/, 100/*f1.myInt*/, 1000L/*f1.myLong*/, "One" /*f1.myString*/, 10000L /*nestedPojo.longNumber*/));
+		data.add(new POJO(2, "Second", 20, 200, 2000L, "Two", 20000L));
+		data.add(new POJO(3, "Third", 30, 300, 3000L, "Three", 30000L));
+
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<POJO> getDuplicatePojoDataSet(ExecutionEnvironment env) {
+		List<POJO> data = new ArrayList<POJO>();
+
+		data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10000L)); // 5x
+		data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10000L));
+		data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10000L));
+		data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10000L));
+		data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10000L));
+		data.add(new POJO(2, "Second", 20, 200, 2000L, "Two", 20000L));
+		data.add(new POJO(3, "Third", 30, 300, 3000L, "Three", 30000L)); // 2x
+		data.add(new POJO(3, "Third", 30, 300, 3000L, "Three", 30000L));
+
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<POJO> getMixedPojoDataSet(ExecutionEnvironment env) {
+		List<POJO> data = new ArrayList<POJO>();
+
+		data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10100L)); // 5x
+		data.add(new POJO(2, "First_", 10, 105, 1000L, "One", 10200L));
+		data.add(new POJO(3, "First", 11, 102, 3000L, "One", 10200L));
+		data.add(new POJO(4, "First_", 11, 106, 1000L, "One", 10300L));
+		data.add(new POJO(5, "First", 11, 102, 2000L, "One", 10100L));
+		data.add(new POJO(6, "Second_", 20, 200, 2000L, "Two", 10100L));
+		data.add(new POJO(7, "Third", 31, 301, 2000L, "Three", 10200L)); // 2x
+		data.add(new POJO(8, "Third_", 30, 300, 1000L, "Three", 10100L));
+
+		return env.fromCollection(data);
+	}
+
+	public static class POJO {
+		public IntValue number;
+		public StringValue str;
+		public Tuple2<IntValue, CustomType> nestedTupleWithCustom;
+		public NestedPojo nestedPojo;
+		public transient LongValue ignoreMe;
+
+		public POJO(int i0, String s0,
+					int i1, int i2, long l0, String s1,
+					long l1) {
+			this.number = new IntValue(i0);
+			this.str = new StringValue(s0);
+			this.nestedTupleWithCustom = new Tuple2<>(new IntValue(i1), new CustomType(i2, l0, s1));
+			this.nestedPojo = new NestedPojo();
+			this.nestedPojo.longNumber = new LongValue(l1);
+		}
+
+		public POJO() {
+		}
+
+		@Override
+		public String toString() {
+			return number + " " + str + " " + nestedTupleWithCustom + " " + nestedPojo.longNumber;
+		}
+	}
+
+	public static class NestedPojo {
+		public static Object ignoreMe;
+		public LongValue longNumber;
+
+		public NestedPojo() {
+		}
+	}
+
+	public static DataSet<CrazyNested> getCrazyNestedDataSet(ExecutionEnvironment env) {
+		List<CrazyNested> data = new ArrayList<CrazyNested>();
+
+		data.add(new CrazyNested("aa"));
+		data.add(new CrazyNested("bb"));
+		data.add(new CrazyNested("bb"));
+		data.add(new CrazyNested("cc"));
+		data.add(new CrazyNested("cc"));
+		data.add(new CrazyNested("cc"));
+
+		return env.fromCollection(data);
+	}
+
+	public static class CrazyNested {
+		public CrazyNestedL1 nest_Lvl1;
+		public LongValue something; // test proper null-value handling
+
+		public CrazyNested() {
+		}
+
+		public CrazyNested(String set, String second, long s) { // additional CTor to set all fields to non-null values
+			this(set);
+			something = new LongValue(s);
+			nest_Lvl1.a = new StringValue(second);
+		}
+
+		public CrazyNested(String set) {
+			nest_Lvl1 = new CrazyNestedL1();
+			nest_Lvl1.nest_Lvl2 = new CrazyNestedL2();
+			nest_Lvl1.nest_Lvl2.nest_Lvl3 = new CrazyNestedL3();
+			nest_Lvl1.nest_Lvl2.nest_Lvl3.nest_Lvl4 = new CrazyNestedL4();
+			nest_Lvl1.nest_Lvl2.nest_Lvl3.nest_Lvl4.f1nal = new StringValue(set);
+		}
+	}
+
+	public static class CrazyNestedL1 {
+		public StringValue a;
+		public IntValue b;
+		public CrazyNestedL2 nest_Lvl2;
+	}
+
+	public static class CrazyNestedL2 {
+		public CrazyNestedL3 nest_Lvl3;
+	}
+
+	public static class CrazyNestedL3 {
+		public CrazyNestedL4 nest_Lvl4;
+	}
+
+	public static class CrazyNestedL4 {
+		public StringValue f1nal;
+	}
+
+	// Copied from TypeExtractorTest
+	public static class FromTuple extends Tuple3<StringValue, StringValue, LongValue> {
+		private static final long serialVersionUID = 1L;
+		public IntValue special;
+	}
+
+	public static class FromTupleWithCTor extends FromTuple {
+
+		private static final long serialVersionUID = 1L;
+
+		public FromTupleWithCTor() {}
+
+		public FromTupleWithCTor(int special, long tupleField) {
+			this.special = new IntValue(special);
+			this.setField(new LongValue(tupleField), 2);
+		}
+	}
+
+	public static DataSet<FromTupleWithCTor> getPojoExtendingFromTuple(ExecutionEnvironment env) {
+		List<FromTupleWithCTor> data = new ArrayList<>();
+		data.add(new FromTupleWithCTor(1, 10L)); // 3x
+		data.add(new FromTupleWithCTor(1, 10L));
+		data.add(new FromTupleWithCTor(1, 10L));
+		data.add(new FromTupleWithCTor(2, 20L)); // 2x
+		data.add(new FromTupleWithCTor(2, 20L));
+		return env.fromCollection(data);
+	}
+
+	public static class PojoContainingTupleAndWritable {
+		public IntValue someInt;
+		public StringValue someString;
+		public IntWritable hadoopFan;
+		public Tuple2<LongValue, LongValue> theTuple;
+
+		public PojoContainingTupleAndWritable() {
+		}
+
+		public PojoContainingTupleAndWritable(int i, long l1, long l2) {
+			hadoopFan = new IntWritable(i);
+			someInt = new IntValue(i);
+			theTuple = new Tuple2<>(new LongValue(l1), new LongValue(l2));
+		}
+	}
+
+	public static DataSet<PojoContainingTupleAndWritable> getPojoContainingTupleAndWritable(ExecutionEnvironment env) {
+		List<PojoContainingTupleAndWritable> data = new ArrayList<>();
+		data.add(new PojoContainingTupleAndWritable(1, 10L, 100L)); // 1x
+		data.add(new PojoContainingTupleAndWritable(2, 20L, 200L)); // 5x
+		data.add(new PojoContainingTupleAndWritable(2, 20L, 200L));
+		data.add(new PojoContainingTupleAndWritable(2, 20L, 200L));
+		data.add(new PojoContainingTupleAndWritable(2, 20L, 200L));
+		data.add(new PojoContainingTupleAndWritable(2, 20L, 200L));
+		return env.fromCollection(data);
+	}
+
+
+
+	public static DataSet<PojoContainingTupleAndWritable> getGroupSortedPojoContainingTupleAndWritable(ExecutionEnvironment env) {
+		List<PojoContainingTupleAndWritable> data = new ArrayList<>();
+		data.add(new PojoContainingTupleAndWritable(1, 10L, 100L)); // 1x
+		data.add(new PojoContainingTupleAndWritable(2, 20L, 200L)); // 5x
+		data.add(new PojoContainingTupleAndWritable(2, 20L, 201L));
+		data.add(new PojoContainingTupleAndWritable(2, 30L, 200L));
+		data.add(new PojoContainingTupleAndWritable(2, 30L, 600L));
+		data.add(new PojoContainingTupleAndWritable(2, 30L, 400L));
+		return env.fromCollection(data);
+	}
+
+	public static DataSet<Tuple3<IntValue, CrazyNested, POJO>> getTupleContainingPojos(ExecutionEnvironment env) {
+		List<Tuple3<IntValue, CrazyNested, POJO>> data = new ArrayList<>();
+		data.add(new Tuple3<IntValue, CrazyNested, POJO>(new IntValue(1), new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); // 3x
+		data.add(new Tuple3<IntValue, CrazyNested, POJO>(new IntValue(1), new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L)));
+		data.add(new Tuple3<IntValue, CrazyNested, POJO>(new IntValue(1), new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L)));
+		// POJO is not initialized according to the first two fields.
+		data.add(new Tuple3<IntValue, CrazyNested, POJO>(new IntValue(2), new CrazyNested("two", "duo", 2L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); // 1x
+		return env.fromCollection(data);
+	}
+
+	public static class Pojo1 {
+		public StringValue a;
+		public StringValue b;
+
+		public Pojo1() {}
+
+		public Pojo1(String a, String b) {
+			this.a = new StringValue(a);
+			this.b = new StringValue(b);
+		}
+	}
+
+	public static class Pojo2 {
+		public StringValue a2;
+		public StringValue b2;
+	}
+
+	public static class PojoWithMultiplePojos {
+		public Pojo1 p1;
+		public Pojo2 p2;
+		public IntValue i0;
+
+		public PojoWithMultiplePojos() {
+		}
+
+		public PojoWithMultiplePojos(String a, String b, String a1, String b1, int i0) {
+			p1 = new Pojo1();
+			p1.a = new StringValue(a);
+			p1.b = new StringValue(b);
+			p2 = new Pojo2();
+			p2.a2 = new StringValue(a1);
+			p2.b2 = new StringValue(b1);
+			this.i0 = new IntValue(i0);
+		}
+	}
+
+	public static DataSet<PojoWithMultiplePojos> getPojoWithMultiplePojos(ExecutionEnvironment env) {
+		List<PojoWithMultiplePojos> data = new ArrayList<>();
+		data.add(new PojoWithMultiplePojos("a", "aa", "b", "bb", 1));
+		data.add(new PojoWithMultiplePojos("b", "bb", "c", "cc", 2));
+		data.add(new PojoWithMultiplePojos("b", "bb", "c", "cc", 2));
+		data.add(new PojoWithMultiplePojos("b", "bb", "c", "cc", 2));
+		data.add(new PojoWithMultiplePojos("d", "dd", "e", "ee", 3));
+		data.add(new PojoWithMultiplePojos("d", "dd", "e", "ee", 3));
+		return env.fromCollection(data);
+	}
+
+	public enum Category {
+		CAT_A, CAT_B;
+	}
+
+	public static class PojoWithDateAndEnum {
+		public StringValue group;
+		public Date date;
+		public Category cat;
+	}
+	
+	public static DataSet<PojoWithDateAndEnum> getPojoWithDateAndEnum(ExecutionEnvironment env) {
+		List<PojoWithDateAndEnum> data = new ArrayList<PojoWithDateAndEnum>();
+		
+		PojoWithDateAndEnum one = new PojoWithDateAndEnum();
+		one.group = new StringValue("a");
+		one.date = new Date(666);
+		one.cat = Category.CAT_A;
+		data.add(one);
+		
+		PojoWithDateAndEnum two = new PojoWithDateAndEnum();
+		two.group = new StringValue("a");
+		two.date = new Date(666);
+		two.cat = Category.CAT_A;
+		data.add(two);
+		
+		PojoWithDateAndEnum three = new PojoWithDateAndEnum();
+		three.group = new StringValue("b");
+		three.date = new Date(666);
+		three.cat = Category.CAT_B;
+		data.add(three);
+		
+		return env.fromCollection(data);
+	}
+
+	public static class PojoWithCollection {
+		public List<Pojo1> pojos;
+		public IntValue key;
+		public java.sql.Date sqlDate;
+		public BigInteger bigInt;
+		public BigDecimal bigDecimalKeepItNull;
+		public BigInt scalaBigInt;
+		public List<Object> mixed;
+
+		@Override
+		public String toString() {
+			return "PojoWithCollection{" +
+					"pojos.size()=" + pojos.size() +
+					", key=" + key +
+					", sqlDate=" + sqlDate +
+					", bigInt=" + bigInt +
+					", bigDecimalKeepItNull=" + bigDecimalKeepItNull +
+					", scalaBigInt=" + scalaBigInt +
+					", mixed=" + mixed +
+					'}';
+		}
+	}
+
+	public static class PojoWithCollectionGeneric {
+		public List<Pojo1> pojos;
+		public IntValue key;
+		public java.sql.Date sqlDate;
+		public BigInteger bigInt;
+		public BigDecimal bigDecimalKeepItNull;
+		public BigInt scalaBigInt;
+		public List<Object> mixed;
+		private PojoWithDateAndEnum makeMeGeneric;
+
+		@Override
+		public String toString() {
+			return "PojoWithCollection{" +
+					"pojos.size()=" + pojos.size() +
+					", key=" + key +
+					", sqlDate=" + sqlDate +
+					", bigInt=" + bigInt +
+					", bigDecimalKeepItNull=" + bigDecimalKeepItNull +
+					", scalaBigInt=" + scalaBigInt +
+					", mixed=" + mixed +
+					'}';
+		}
+	}
+
+	public static DataSet<PojoWithCollection> getPojoWithCollection(ExecutionEnvironment env) {
+		List<PojoWithCollection> data = new ArrayList<>();
+
+		List<Pojo1> pojosList1 = new ArrayList<>();
+		pojosList1.add(new Pojo1("a", "aa"));
+		pojosList1.add(new Pojo1("b", "bb"));
+
+		List<Pojo1> pojosList2 = new ArrayList<>();
+		pojosList2.add(new Pojo1("a2", "aa2"));
+		pojosList2.add(new Pojo1("b2", "bb2"));
+
+		PojoWithCollection pwc1 = new PojoWithCollection();
+		pwc1.pojos = pojosList1;
+		pwc1.key = new IntValue(0);
+		pwc1.bigInt = BigInteger.valueOf(Long.MAX_VALUE).multiply(BigInteger.TEN);
+		pwc1.scalaBigInt = BigInt.int2bigInt(10);
+		pwc1.bigDecimalKeepItNull = null;
+		
+		// use calendar to make it stable across time zones
+		GregorianCalendar gcl1 = new GregorianCalendar(2033, 04, 18);
+		pwc1.sqlDate = new java.sql.Date(gcl1.getTimeInMillis());
+		pwc1.mixed = new ArrayList<Object>();
+		Map<StringValue, IntValue> map = new HashMap<>();
+		map.put(new StringValue("someKey"), new IntValue(1));
+		pwc1.mixed.add(map);
+		pwc1.mixed.add(new File("/this/is/wrong"));
+		pwc1.mixed.add("uhlala");
+
+		PojoWithCollection pwc2 = new PojoWithCollection();
+		pwc2.pojos = pojosList2;
+		pwc2.key = new IntValue(0);
+		pwc2.bigInt = BigInteger.valueOf(Long.MAX_VALUE).multiply(BigInteger.TEN);
+		pwc2.scalaBigInt = BigInt.int2bigInt(31104000);
+		pwc2.bigDecimalKeepItNull = null;
+		
+		GregorianCalendar gcl2 = new GregorianCalendar(1976, 4, 3);
+		pwc2.sqlDate = new java.sql.Date(gcl2.getTimeInMillis()); // 1976
+
+		data.add(pwc1);
+		data.add(pwc2);
+
+		return env.fromCollection(data);
+	}
+
+}
+


Mime
View raw message