Add GenericTypeComparator

stratosphere-core/src/test/java/eu/stratosphere/api/common/typeutils/ComparatorTestBase.java

@@ -43,6 +43,13 @@ public abstract class ComparatorTestBase<T> {
 
 	protected abstract TypeSerializer<T> createSerializer();
 
+	/**
+	 * Returns the sorted data set.
+	 * <p>
+	 * Note: every element needs to be *strictly greater* than the previous element.
+	 *
+	 * @return sorted test data set
+	 */
 	protected abstract T[] getSortedTestData();
 
 	// -------------------------------- test duplication ------------------------------------------
@@ -433,7 +440,7 @@ public abstract class ComparatorTestBase<T> {
 		assertTrue("No data available during deserialization.", in.available() > 0);
 
 		T deserialized = serializer.deserialize(serializer.createInstance(), in);
-		deepEquals("Deserialized value if wrong.", value, deserialized);
+		deepEquals("Deserialized value is wrong.", value, deserialized);
 
 	}
 

stratosphere-core/src/test/java/eu/stratosphere/api/common/typeutils/base/ByteComparatorTest.java

@@ -17,8 +17,6 @@ package eu.stratosphere.api.common.typeutils.base;
 import eu.stratosphere.api.common.typeutils.ComparatorTestBase;
 import eu.stratosphere.api.common.typeutils.TypeComparator;
 import eu.stratosphere.api.common.typeutils.TypeSerializer;
-import eu.stratosphere.api.common.typeutils.base.ByteComparator;
-import eu.stratosphere.api.common.typeutils.base.ByteSerializer;
 
 import java.util.Random;
 

stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/GenericTypeInfo.java

@@ -17,6 +17,7 @@ package eu.stratosphere.api.java.typeutils;
 import eu.stratosphere.api.common.typeutils.TypeComparator;
 import eu.stratosphere.api.common.typeutils.TypeSerializer;
 import eu.stratosphere.api.java.typeutils.runtime.AvroSerializer;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeComparator;
 
 
 /**
@@ -60,19 +61,26 @@ public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType
 	public TypeSerializer<T> createSerializer() {
 		return new AvroSerializer<T>(this.typeClass);
 	}
-	
+
 	@Override
 	public TypeComparator<T> createComparator(boolean sortOrderAscending) {
-		throw new UnsupportedOperationException("Generic type comparators are not yet implemented.");
+		if (isKeyType()) {
+			@SuppressWarnings("unchecked")
+			GenericTypeComparator comparator = new GenericTypeComparator(sortOrderAscending, createSerializer(), this.typeClass);
+
+			return comparator;
+		}
+
+		throw new UnsupportedOperationException("Generic types that don't implement java.lang.Comparable cannot be used as keys.");
 	}
-	
+
 	// --------------------------------------------------------------------------------------------
-	
+
 	@Override
 	public int hashCode() {
 		return typeClass.hashCode() ^ 0x165667b1;
 	}
-	
+
 	@Override
 	public boolean equals(Object obj) {
 		if (obj.getClass() == GenericTypeInfo.class) {

stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/GenericTypeComparator.java

+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.api.java.typeutils.runtime;
+
+import com.esotericsoftware.kryo.Kryo;
+import eu.stratosphere.api.common.typeutils.TypeComparator;
+import eu.stratosphere.api.common.typeutils.TypeSerializer;
+import eu.stratosphere.api.common.typeutils.TypeSerializerFactory;
+import eu.stratosphere.core.memory.DataInputView;
+import eu.stratosphere.core.memory.DataOutputView;
+import eu.stratosphere.core.memory.MemorySegment;
+import eu.stratosphere.types.NormalizableKey;
+import eu.stratosphere.util.InstantiationUtil;
+
+import java.io.IOException;
+
+/**
+ * TypeComparator for all types that extend Comparable.
+ */
+public class GenericTypeComparator<T extends Comparable<T>> extends TypeComparator<T> {
+
+	private static final long serialVersionUID = 1L;
+
+	private final boolean ascending;
+
+	private final Class<T> type;
+
+	private final TypeSerializerFactory<T> serializerFactory;
+
+	private transient TypeSerializer<T> serializer;
+
+	private transient T reference;
+
+	private transient T tmpReference;
+
+	private transient Kryo kryo;
+
+	// ------------------------------------------------------------------------
+
+	public GenericTypeComparator(boolean ascending, TypeSerializer<T> serializer, Class<T> type) {
+		this.ascending = ascending;
+		this.serializer = serializer;
+		this.type = type;
+
+		this.serializerFactory = this.serializer.isStateful()
+				? new RuntimeStatefulSerializerFactory<T>(this.serializer, this.type)
+				: new RuntimeStatelessSerializerFactory<T>(this.serializer, this.type);
+	}
+
+	private GenericTypeComparator(GenericTypeComparator<T> toClone) {
+		this.ascending = toClone.ascending;
+		this.serializerFactory = toClone.serializerFactory;
+		this.type = toClone.type;
+	}
+
+	@Override
+	public int hash(T record) {
+		return record.hashCode();
+	}
+
+	@Override
+	public void setReference(T toCompare) {
+		checkKryoInitialized();
+		this.reference = this.kryo.copy(toCompare);
+	}
+
+	@Override
+	public boolean equalToReference(T candidate) {
+		return candidate.equals(this.reference);
+	}
+
+	@Override
+	public int compareToReference(TypeComparator<T> referencedComparator) {
+		T otherRef = ((GenericTypeComparator<T>) referencedComparator).reference;
+		int cmp = otherRef.compareTo(this.reference);
+
+		return this.ascending ? cmp : -cmp;
+	}
+
+	@Override
+	public int compare(T first, T second) {
+		return first.compareTo(second);
+	}
+
+	@Override
+	public int compare(final DataInputView firstSource, final DataInputView secondSource) throws IOException {
+		if (this.serializer == null) {
+			this.serializer = this.serializerFactory.getSerializer();
+		}
+
+		if (this.reference == null) {
+			this.reference = this.serializer.createInstance();
+		}
+
+		if (this.tmpReference == null) {
+			this.tmpReference = this.serializer.createInstance();
+		}
+
+		this.reference = this.serializer.deserialize(this.reference, firstSource);
+		this.tmpReference = this.serializer.deserialize(this.tmpReference, secondSource);
+
+		int cmp = this.reference.compareTo(this.tmpReference);
+		return this.ascending ? cmp : -cmp;
+	}
+
+	@Override
+	public boolean supportsNormalizedKey() {
+		return NormalizableKey.class.isAssignableFrom(this.type);
+	}
+
+	@Override
+	public int getNormalizeKeyLen() {
+		if (this.reference == null) {
+			this.reference = InstantiationUtil.instantiate(this.type);
+		}
+
+		NormalizableKey<?> key = (NormalizableKey<?>) this.reference;
+		return key.getMaxNormalizedKeyLen();
+	}
+
+	@Override
+	public boolean isNormalizedKeyPrefixOnly(int keyBytes) {
+		return keyBytes < getNormalizeKeyLen();
+	}
+
+	@Override
+	public void putNormalizedKey(T record, MemorySegment target, int offset, int numBytes) {
+		NormalizableKey<?> key = (NormalizableKey<?>) record;
+		key.copyNormalizedKey(target, offset, numBytes);
+	}
+
+	@Override
+	public boolean invertNormalizedKey() {
+		return !ascending;
+	}
+
+	@Override
+	public TypeComparator<T> duplicate() {
+		return new GenericTypeComparator<T>(this);
+	}
+
+	private final void checkKryoInitialized() {
+		if (this.kryo == null) {
+			this.kryo = new Kryo();
+			this.kryo.setAsmEnabled(true);
+			this.kryo.register(this.type);
+		}
+	}
+
+	// ------------------------------------------------------------------------
+
+	@Override
+	public boolean supportsSerializationWithKeyNormalization() {
+		return false;
+	}
+
+	@Override
+	public void writeWithKeyNormalization(T record, DataOutputView target) throws IOException {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public T readWithKeyDenormalization(T reuse, DataInputView source) throws IOException {
+		throw new UnsupportedOperationException();
+	}
+}

stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/RuntimeStatefulSerializerFactory.java

@@ -23,8 +23,7 @@ import eu.stratosphere.util.InstantiationUtil;
 public final class RuntimeStatefulSerializerFactory<T> implements TypeSerializerFactory<T>, java.io.Serializable {
 
 	private static final long serialVersionUID = 1L;
-	
-	
+
 	private static final String CONFIG_KEY_SER = "SER_DATA";
 
 	private static final String CONFIG_KEY_CLASS = "CLASS_DATA";
@@ -37,7 +36,6 @@ public final class RuntimeStatefulSerializerFactory<T> implements TypeSerializer
 
 	private Class<T> clazz;
 
-
 	public RuntimeStatefulSerializerFactory() {}
 
 	public RuntimeStatefulSerializerFactory(TypeSerializer<T> serializer, Class<T> clazz) {

stratosphere-java/src/test/java/eu/stratosphere/api/java/typeutils/runtime/GenericArraySerializerTest.java

@@ -25,11 +25,11 @@ import org.junit.Test;
 import eu.stratosphere.api.common.typeutils.SerializerTestInstance;
 import eu.stratosphere.api.common.typeutils.TypeSerializer;
 import eu.stratosphere.api.common.typeutils.base.StringSerializer;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.Book;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.BookAuthor;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.ComplexNestedObject1;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.ComplexNestedObject2;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.SimpleTypes;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.Book;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.BookAuthor;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.ComplexNestedObject1;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.ComplexNestedObject2;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.SimpleTypes;
 import eu.stratosphere.util.StringUtils;
 
 public class GenericArraySerializerTest {

stratosphere-java/src/test/java/eu/stratosphere/api/java/typeutils/runtime/GenericTypeComparatorTest.java

+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.api.java.typeutils.runtime;
+
+import eu.stratosphere.api.common.typeutils.ComparatorTestBase;
+import eu.stratosphere.api.common.typeutils.TypeComparator;
+import eu.stratosphere.api.common.typeutils.TypeSerializer;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class GenericTypeComparatorTest {
+
+	@Test
+	public void testString() {
+		runTests(new String[]{
+				"",
+				"Lorem Ipsum Dolor Omit Longer",
+				"aaaa",
+				"abcd",
+				"abce",
+				"abdd",
+				"accd",
+				"bbcd"
+		});
+	}
+
+	@Test
+	public void testSimpleTypesObjects() {
+		runTests(
+				new SimpleTypes(0, 1, (byte) 2, "", (short) 3, 4.0),
+				new SimpleTypes(1, 1, (byte) 2, "", (short) 3, 4.0),
+				new SimpleTypes(1, 2, (byte) 2, "", (short) 3, 4.0),
+				new SimpleTypes(1, 2, (byte) 3, "", (short) 3, 4.0),
+				new SimpleTypes(1, 2, (byte) 3, "a", (short) 3, 4.0),
+				new SimpleTypes(1, 2, (byte) 3, "b", (short) 3, 4.0),
+				new SimpleTypes(1, 2, (byte) 3, "b", (short) 4, 4.0),
+				new SimpleTypes(1, 2, (byte) 3, "b", (short) 4, 6.0)
+		);
+	}
+
+	@Test
+	public void testCompositeObject() {
+		ComplexNestedObject1 o1 = new ComplexNestedObject1(-1100);
+		ComplexNestedObject1 o2 = new ComplexNestedObject1(0);
+		ComplexNestedObject1 o3 = new ComplexNestedObject1(44);
+		ComplexNestedObject1 o4 = new ComplexNestedObject1(76923, "A");
+		ComplexNestedObject1 o5 = new ComplexNestedObject1(5626435, "A somewhat random collection");
+
+		runTests(o1, o2, o3, o4, o5);
+	}
+
+	@Test
+	public void testBeanStyleObjects() {
+		{
+			Book b111 = new Book(-1L, "A Low level interfaces", 0xC);
+			Book b122 = new Book(-1L, "Low level interfaces", 0xC);
+			Book b123 = new Book(-1L, "Low level interfaces", 0xC0FFEE);
+
+			Book b2 = new Book(0L, "Debugging byte streams", 1337);
+			Book b3 = new Book(976243875L, "The Serialization Odysse", 42);
+
+			runTests(b111, b122, b123, b2, b3);
+		}
+
+		{
+			BookAuthor b1 = new BookAuthor(976243875L, new ArrayList<String>(), "Arno Nym");
+
+			ArrayList<String> list = new ArrayList<String>();
+			list.add("A");
+			list.add("B");
+			list.add("C");
+			list.add("D");
+			list.add("E");
+
+			BookAuthor b2 = new BookAuthor(976243875L, list, "The Saurus");
+
+			runTests(b1, b2);
+		}
+	}
+
+	// ------------------------------------------------------------------------
+
+	private final <T> void runTests(T... sortedTestData) {
+		ComparatorTestInstance<T> testBase = new ComparatorTestInstance<T>(sortedTestData);
+		testBase.testAll();
+	}
+
+	private static final <T> TypeSerializer<T> createSerializer(Class<T> type) {
+		return new AvroSerializer<T>(type);
+	}
+
+	// ------------------------------------------------------------------------
+	// test instance
+	// ------------------------------------------------------------------------
+
+	private class ComparatorTestInstance<T> extends ComparatorTestBase<T> {
+
+		private final T[] testData;
+
+		private final Class<T> type;
+
+		public ComparatorTestInstance(T[] testData) {
+			if (testData == null || testData.length == 0) {
+				throw new IllegalArgumentException();
+			}
+
+			this.testData = testData;
+			this.type = (Class<T>) testData[0].getClass();
+		}
+
+		@Override
+		protected TypeComparator<T> createComparator(boolean ascending) {
+			return new GenericTypeComparator(ascending, GenericTypeComparatorTest.createSerializer(this.type), this.type);
+		}
+
+		@Override
+		protected TypeSerializer<T> createSerializer() {
+			return GenericTypeComparatorTest.createSerializer(this.type);
+		}
+
+		@Override
+		protected T[] getSortedTestData() {
+			return this.testData;
+		}
+
+		public void testAll() {
+			testDuplicate();
+			testEquality();
+			testEqualityWithReference();
+			testInequality();
+			testInequalityWithReference();
+			testNormalizedKeysEqualsFullLength();
+			testNormalizedKeysEqualsHalfLength();
+			testNormalizedKeysGreatSmallFullLength();
+			testNormalizedKeysGreatSmallAscDescHalfLength();
+			testNormalizedKeyReadWriter();
+		}
+	}
+
+	// ------------------------------------------------------------------------
+	//  test objects
+	// ------------------------------------------------------------------------
+
+	public static final class SimpleTypes implements Comparable<SimpleTypes> {
+
+		private final int iVal;
+		private final long lVal;
+		private final byte bVal;
+		private final String sVal;
+		private final short rVal;
+		private final double dVal;
+
+		public SimpleTypes() {
+			this(0, 0, (byte) 0, "", (short) 0, 0);
+		}
+
+		public SimpleTypes(int iVal, long lVal, byte bVal, String sVal, short rVal, double dVal) {
+			this.iVal = iVal;
+			this.lVal = lVal;
+			this.bVal = bVal;
+			this.sVal = sVal;
+			this.rVal = rVal;
+			this.dVal = dVal;
+		}
+
+		@Override
+		public String toString() {
+			return String.format("(%d, %d, %d, %s, %d, %f)", iVal, lVal, bVal, sVal, rVal, dVal);
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (obj.getClass() == SimpleTypes.class) {
+				SimpleTypes other = (SimpleTypes) obj;
+
+				return other.iVal == this.iVal &&
+						other.lVal == this.lVal &&
+						other.bVal == this.bVal &&
+						other.sVal.equals(this.sVal) &&
+						other.rVal == this.rVal &&
+						other.dVal == this.dVal;
+			} else {
+				return false;
+			}
+		}
+
+		@Override
+		public int compareTo(SimpleTypes o) {
+			int cmp = (this.iVal < o.iVal ? -1 : (this.iVal == o.iVal ? 0 : 1));
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			cmp = (this.lVal < o.lVal ? -1 : (this.lVal == o.lVal ? 0 : 1));
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			cmp = (this.bVal < o.bVal ? -1 : (this.bVal == o.bVal ? 0 : 1));
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			cmp = this.sVal.compareTo(o.sVal);
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			cmp = (this.rVal < o.rVal ? -1 : (this.rVal == o.rVal ? 0 : 1));
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			return (this.dVal < o.dVal ? -1 : (this.dVal == o.dVal ? 0 : 1));
+		}
+	}
+
+	public static class ComplexNestedObject1 implements Comparable<ComplexNestedObject1> {
+
+		private double doubleValue;
+
+		private List<String> stringList;
+
+		public ComplexNestedObject1() {
+		}
+
+		public ComplexNestedObject1(double value, String... listElements) {
+			this.doubleValue = value;
+
+			this.stringList = new ArrayList<String>();
+			for (String str : listElements) {
+				this.stringList.add(str);
+			}
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (obj.getClass() == ComplexNestedObject1.class) {
+				ComplexNestedObject1 other = (ComplexNestedObject1) obj;
+				return other.doubleValue == this.doubleValue && this.stringList.equals(other.stringList);
+			} else {
+				return false;
+			}
+		}
+
+		@Override
+		public int compareTo(ComplexNestedObject1 o) {
+			int cmp = (this.doubleValue < o.doubleValue ? -1 : (this.doubleValue == o.doubleValue ? 0 : 1));
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			int size = this.stringList.size();
+			int otherSize = o.stringList.size();
+
+			cmp = (size < otherSize ? -1 : (size == otherSize ? 0 : 1));
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			for (int i = 0; i < size; i++) {
+				cmp = this.stringList.get(i).compareTo(o.stringList.get(i));
+				if (cmp != 0) {
+					return cmp;
+				}
+			}
+
+			return 0;
+		}
+	}
+
+	public static class Book implements Comparable<Book> {
+
+		private long bookId;
+		private String title;
+		private long authorId;
+
+		public Book() {
+		}
+
+		public Book(long bookId, String title, long authorId) {
+			this.bookId = bookId;
+			this.title = title;
+			this.authorId = authorId;
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (obj.getClass() == Book.class) {
+				Book other = (Book) obj;
+				return other.bookId == this.bookId && other.authorId == this.authorId && this.title.equals(other.title);
+			} else {
+				return false;
+			}
+		}
+
+		@Override
+		public int compareTo(Book o) {
+			int cmp = (this.bookId < o.bookId ? -1 : (this.bookId == o.bookId ? 0 : 1));
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			cmp = title.compareTo(o.title);
+			if (cmp != 0) {
+				return cmp;
+			}
+
+			return (this.authorId < o.authorId ? -1 : (this.authorId == o.authorId ? 0 : 1));
+		}
+	}
+
+	public static class BookAuthor implements Comparable<BookAuthor> {
+
+		private long authorId;
+		private List<String> bookTitles;
+		private String authorName;
+
+		public BookAuthor() {
+		}
+
+		public BookAuthor(long authorId, List<String> bookTitles, String authorName) {
+			this.authorId = authorId;
+			this.bookTitles = bookTitles;
+			this.authorName = authorName;
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (obj.getClass() == BookAuthor.class) {
+				BookAuthor other = (BookAuthor) obj;
+				return other.authorName.equals(this.authorName) && other.authorId == this.authorId &&
+						other.bookTitles.equals(this.bookTitles);
+			} else {
+				return false;
+			}
+		}
+
+		@Override
+		public int compareTo(BookAuthor o) {
+			int cmp = (this.authorId < o.authorId ? -1 : (this.authorId == o.authorId ? 0 : 1));
+			if (cmp != 0) return cmp;
+
+			int size = this.bookTitles.size();
+			int oSize = o.bookTitles.size();
+			cmp = (size < oSize ? -1 : (size == oSize ? 0 : 1));
+			if (cmp != 0) return cmp;
+
+			for (int i = 0; i < size; i++) {
+				cmp = this.bookTitles.get(i).compareTo(o.bookTitles.get(i));
+				if (cmp != 0) return cmp;
+			}
+
+			return this.authorName.compareTo(o.authorName);
+		}
+	}
+}

stratosphere-java/src/test/java/eu/stratosphere/api/java/typeutils/runtime/GenericSerializerTest.java → stratosphere-java/src/test/java/eu/stratosphere/api/java/typeutils/runtime/GenericTypeSerializerTest.java

@@ -14,34 +14,33 @@
  **********************************************************************************************************************/
 package eu.stratosphere.api.java.typeutils.runtime;
 
+import eu.stratosphere.api.common.typeutils.SerializerTestInstance;
+import eu.stratosphere.util.StringUtils;
+import org.junit.Test;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
 
-import org.junit.Test;
-
-import eu.stratosphere.api.common.typeutils.SerializerTestInstance;
-import eu.stratosphere.util.StringUtils;
-
 /**
  * A test for the {@link AvroSerializer}.
  */
-public class GenericSerializerTest {
+public class GenericTypeSerializerTest {
 
 	private final Random rnd = new Random(349712539451944123L);
-	
-	
+
+
 	@Test
 	public void testString() {
 		runTests("abc", "",
 				StringUtils.getRandomString(new Random(289347567856686223L), 10, 100),
 				StringUtils.getRandomString(new Random(289347567856686223L), 1000, 5000),
 				StringUtils.getRandomString(new Random(289347567856686223L), 30000, 35000),
-				StringUtils.getRandomString(new Random(289347567856686223L), 100*1024, 105*1024));
+				StringUtils.getRandomString(new Random(289347567856686223L), 100 * 1024, 105 * 1024));
 	}
-	
+
 	@Test
 	public void testSimpleTypesObjects() {
 		SimpleTypes a = new SimpleTypes();
@@ -57,10 +56,10 @@ public class GenericSerializerTest {
 				StringUtils.getRandomString(rnd, 10, 100), (short) rnd.nextInt(), rnd.nextDouble());
 		SimpleTypes g = new SimpleTypes(rnd.nextInt(), rnd.nextLong(), (byte) rnd.nextInt(),
 				StringUtils.getRandomString(rnd, 10, 100), (short) rnd.nextInt(), rnd.nextDouble());
-		
+
 		runTests(a, b, c, d, e, f, g);
 	}
-	
+
 	@Test
 	public void testCompositeObject() {
 		ComplexNestedObject1 o1 = new ComplexNestedObject1(5626435);
@@ -68,30 +67,30 @@ public class GenericSerializerTest {
 		ComplexNestedObject1 o3 = new ComplexNestedObject1(-1100);
 		ComplexNestedObject1 o4 = new ComplexNestedObject1(0);
 		ComplexNestedObject1 o5 = new ComplexNestedObject1(44);
-		
+
 		runTests(o1, o2, o3, o4, o5);
 	}
-	
+
 	@Test
 	public void testNestedObjects() {
 		ComplexNestedObject2 o1 = new ComplexNestedObject2(rnd);
 		ComplexNestedObject2 o2 = new ComplexNestedObject2();
 		ComplexNestedObject2 o3 = new ComplexNestedObject2(rnd);
 		ComplexNestedObject2 o4 = new ComplexNestedObject2(rnd);
-		
+
 		runTests(o1, o2, o3, o4);
 	}
-	
+
 	@Test
 	public void testBeanStyleObjects() {
 		{
 			Book b1 = new Book(976243875L, "The Serialization Odysse", 42);
 			Book b2 = new Book(0L, "Debugging byte streams", 1337);
 			Book b3 = new Book(-1L, "Low level interfaces", 0xC0FFEE);
-			
+
 			runTests(b1, b2, b3);
 		}
-		
+
 		// object with collection
 		{
 			ArrayList<String> list = new ArrayList<String>();
@@ -100,52 +99,52 @@ public class GenericSerializerTest {
 			list.add("C");
 			list.add("D");
 			list.add("E");
-			
+
 			BookAuthor b1 = new BookAuthor(976243875L, list, "Arno Nym");
-			
+
 			ArrayList<String> list2 = new ArrayList<String>();
 			BookAuthor b2 = new BookAuthor(987654321L, list2, "The Saurus");
-			
+
 			runTests(b1, b2);
 		}
 	}
-	
-		private final <T> void runTests(T... instances) {
+
+	private final <T> void runTests(T... instances) {
 		if (instances == null || instances.length == 0) {
 			throw new IllegalArgumentException();
 		}
-		
+
 		@SuppressWarnings("unchecked")
 		Class<T> clazz = (Class<T>) instances[0].getClass();
-		
+
 		AvroSerializer<T> serializer = createSerializer(clazz);
 		SerializerTestInstance<T> test = new SerializerTestInstance<T>(serializer, clazz, -1, instances);
 		test.testAll();
 	}
-	
+
 	private final <T> AvroSerializer<T> createSerializer(Class<T> type) {
 		return new AvroSerializer<T>(type);
 	}
-	
+
 	// --------------------------------------------------------------------------------------------
 	//  Test Objects
 	// --------------------------------------------------------------------------------------------
 
 
 	public static final class SimpleTypes {
-		
+
 		private final int iVal;
 		private final long lVal;
 		private final byte bVal;
 		private final String sVal;
 		private final short rVal;
 		private final double dVal;
-		
-		
+
+
 		public SimpleTypes() {
 			this(0, 0, (byte) 0, "", (short) 0, 0);
 		}
-		
+
 		public SimpleTypes(int iVal, long lVal, byte bVal, String sVal, short rVal, double dVal) {
 			this.iVal = iVal;
 			this.lVal = lVal;
@@ -154,41 +153,42 @@ public class GenericSerializerTest {
 			this.rVal = rVal;
 			this.dVal = dVal;
 		}
-		
+
 		@Override
 		public boolean equals(Object obj) {
 			if (obj.getClass() == SimpleTypes.class) {
 				SimpleTypes other = (SimpleTypes) obj;
-				
+
 				return other.iVal == this.iVal &&
 						other.lVal == this.lVal &&
 						other.bVal == this.bVal &&
 						other.sVal.equals(this.sVal) &&
 						other.rVal == this.rVal &&
 						other.dVal == this.dVal;
-				
+
 			} else {
 				return false;
 			}
 		}
-		
+
 		@Override
 		public String toString() {
 			return String.format("(%d, %d, %d, %s, %d, %f)", iVal, lVal, bVal, sVal, rVal, dVal);
 		}
 	}
-	
+
 	public static class ComplexNestedObject1 {
-		
+
 		private double doubleValue;
-		
+
 		private List<String> stringList;
-		
-		public ComplexNestedObject1() {}
-		
+
+		public ComplexNestedObject1() {
+		}
+
 		public ComplexNestedObject1(int offInit) {
 			this.doubleValue = 6293485.6723 + offInit;
-				
+
 			this.stringList = new ArrayList<String>();
 			this.stringList.add("A" + offInit);
 			this.stringList.add("somewhat" + offInit);
@@ -197,7 +197,7 @@ public class GenericSerializerTest {
 			this.stringList.add("of" + offInit);
 			this.stringList.add("strings" + offInit);
 		}
-		
+
 		@Override
 		public boolean equals(Object obj) {
 			if (obj.getClass() == ComplexNestedObject1.class) {
@@ -208,18 +208,19 @@ public class GenericSerializerTest {
 			}
 		}
 	}
-	
+
 	public static class ComplexNestedObject2 {
-		
+
 		private long longValue;
-		
+
 		private Map<String, ComplexNestedObject1> theMap = new HashMap<String, ComplexNestedObject1>();
-		
-		public ComplexNestedObject2() {}
-		
+
+		public ComplexNestedObject2() {
+		}
+
 		public ComplexNestedObject2(Random rnd) {
 			this.longValue = rnd.nextLong();
-			
+
 			this.theMap.put(String.valueOf(rnd.nextLong()), new ComplexNestedObject1(rnd.nextInt()));
 			this.theMap.put(String.valueOf(rnd.nextLong()), new ComplexNestedObject1(rnd.nextInt()));
 			this.theMap.put(String.valueOf(rnd.nextLong()), new ComplexNestedObject1(rnd.nextInt()));
@@ -227,7 +228,7 @@ public class GenericSerializerTest {
 			this.theMap.put(String.valueOf(rnd.nextLong()), new ComplexNestedObject1(rnd.nextInt()));
 			this.theMap.put(String.valueOf(rnd.nextLong()), new ComplexNestedObject1(rnd.nextInt()));
 		}
-		
+
 		@Override
 		public boolean equals(Object obj) {
 			if (obj.getClass() == ComplexNestedObject2.class) {
@@ -238,21 +239,22 @@ public class GenericSerializerTest {
 			}
 		}
 	}
-	
+
 	public static class Book {
 
 		private long bookId;
 		private String title;
 		private long authorId;
 
-		public Book() {}
+		public Book() {
+		}
 
 		public Book(long bookId, String title, long authorId) {
 			this.bookId = bookId;
 			this.title = title;
 			this.authorId = authorId;
 		}
-		
+
 		@Override
 		public boolean equals(Object obj) {
 			if (obj.getClass() == Book.class) {
@@ -270,14 +272,15 @@ public class GenericSerializerTest {
 		private List<String> bookTitles;
 		private String authorName;
 
-		public BookAuthor() {}
+		public BookAuthor() {
+		}
 
 		public BookAuthor(long authorId, List<String> bookTitles, String authorName) {
 			this.authorId = authorId;
 			this.bookTitles = bookTitles;
 			this.authorName = authorName;
 		}
-		
+
 		@Override
 		public boolean equals(Object obj) {
 			if (obj.getClass() == BookAuthor.class) {

stratosphere-java/src/test/java/eu/stratosphere/api/java/typeutils/runtime/TupleSerializerTest.java

@@ -27,11 +27,11 @@ import eu.stratosphere.api.java.tuple.Tuple2;
 import eu.stratosphere.api.java.tuple.Tuple5;
 import eu.stratosphere.api.java.typeutils.TupleTypeInfo;
 import eu.stratosphere.api.java.typeutils.TypeExtractor;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.Book;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.BookAuthor;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.ComplexNestedObject1;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.ComplexNestedObject2;
-import eu.stratosphere.api.java.typeutils.runtime.GenericSerializerTest.SimpleTypes;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.Book;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.BookAuthor;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.ComplexNestedObject1;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.ComplexNestedObject2;
+import eu.stratosphere.api.java.typeutils.runtime.GenericTypeSerializerTest.SimpleTypes;
 import eu.stratosphere.util.StringUtils;
 
 public class TupleSerializerTest {