[FLINK-2788] [apis] Add TypeHint class to allow type-safe generic typ…

…e parsing

This closes apache#1744

flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeHint.java

@@ -0,0 +1,77 @@
+/*
+ * 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.api.common.typeinfo;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+
+/**
+ * A utility class for describing generic types. It can be used to obtain a type information via:
+ * 
+ * <pre>{@code
+ * TypeInformation<Tuple2<String, Long>> info = TypeInformation.of(new TypeHint<Tuple2<String, Long>>(){});
+ * }</pre>
+ * or
+ * <pre>{@code
+ * TypeInformation<Tuple2<String, Long>> info = new TypeHint<Tuple2<String, Long>>(){}.getTypeInfo();
+ * }</pre>
+ * 
+ * @param <T> The type information to hint.
+ */
+@Public
+public abstract class TypeHint<T> {
+
+	/** The type information described by the hint */
+	private final TypeInformation<T> typeInfo;
+
+	/**
+	 * Creates a hint for the generic type in the class signature.
+	 */
+	public TypeHint() {
+		this.typeInfo = TypeExtractor.createTypeInfo(this, TypeHint.class, getClass(), 0);
+	}
+
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Gets the type information described by this TypeHint.
+	 * @return The type information described by this TypeHint.
+	 */
+	public TypeInformation<T> getTypeInfo() {
+		return typeInfo;
+	}
+
+	// ------------------------------------------------------------------------
+
+	@Override
+	public int hashCode() {
+		return typeInfo.hashCode();
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		return obj == this || 
+			obj instanceof TypeHint && this.typeInfo.equals(((TypeHint<?>) obj).typeInfo);
+	}
+
+	@Override
+	public String toString() {
+		return "TypeHint: " + typeInfo;
+	}
+}

flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java

@@ -22,9 +22,10 @@
 import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
 
 import java.io.Serializable;
-import java.util.LinkedList;
+import java.util.Collections;
 import java.util.List;
 
 /**
@@ -128,7 +129,7 @@ public abstract class TypeInformation<T> implements Serializable {
 	@PublicEvolving
 	public List<TypeInformation<?>> getGenericParameters() {
 		// Return an empty list as the default implementation
-		return new LinkedList<>();
+		return Collections.emptyList();
 	}
 
 	/**
@@ -175,4 +176,39 @@ public boolean isSortKeyType() {
 	 * @return true if obj can be equaled with this, otherwise false
 	 */
 	public abstract boolean canEqual(Object obj);
+
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Creates a TypeInformation for the type described by the given class.
+	 * 
+	 * <p>This method only works for non-generic types. For generic types, use the
+	 * {@link #of(TypeHint)} method.
+	 *
+	 * @param typeClass The class of the type.
+	 * @param <T> The generic type.
+	 *
+	 * @return The TypeInformation object for the type described by the hint.
+	 */
+	public static <T> TypeInformation<T> of(Class<T> typeClass) {
+		return TypeExtractor.createTypeInfo(typeClass);
+	}
+
+	/**
+	 * Creates a TypeInformation for a generic type via a utility "type hint".
+	 * This method can be used as follows:
+	 * <pre>
+	 * {@code
+	 * TypeInformation<Tuple2<String, Long>> info = TypeInformation.of(new TypeHint<Tuple2<String, Long>>(){});
+	 * }
+	 * </pre>
+	 * 
+	 * @param typeHint The hint for the generic type.
+	 * @param <T> The generic type.
+	 *    
+	 * @return The TypeInformation object for the type described by the hint.
+	 */
+	public static <T> TypeInformation<T> of(TypeHint<T> typeHint) {
+		return typeHint.getTypeInfo();
+	}
 }

flink-core/src/test/java/org/apache/flink/api/common/typeinfo/TypeHintTest.java

@@ -0,0 +1,62 @@
+/*
+ * 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.api.common.typeinfo;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class TypeHintTest {
+
+	@Test
+	public void testTypeInfoDirect() {
+
+		// simple (non-generic case)
+		TypeHint<String> stringInfo1 = new TypeHint<String>(){};
+		TypeHint<String> stringInfo2 = new TypeHint<String>(){};
+
+		assertEquals(BasicTypeInfo.STRING_TYPE_INFO, stringInfo1.getTypeInfo());
+
+		assertTrue(stringInfo1.hashCode() == stringInfo2.hashCode());
+		assertTrue(stringInfo1.equals(stringInfo2));
+		assertTrue(stringInfo1.toString().equals(stringInfo2.toString()));
+
+		// generic case
+		TypeHint<Tuple3<String, Double, Boolean>> generic = new TypeHint<Tuple3<String, Double, Boolean>>(){};
+
+		TypeInformation<Tuple3<String, Double, Boolean>> tupleInfo = 
+				new TupleTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.DOUBLE_TYPE_INFO, BasicTypeInfo.BOOLEAN_TYPE_INFO);
+
+		assertEquals(tupleInfo, generic.getTypeInfo());
+	}
+
+	@Test
+	public void testTypeInfoOf() {
+		assertEquals(BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(String.class));
+		assertEquals(BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(new TypeHint<String>(){}));
+
+
+		TypeInformation<Tuple3<String, Double, Boolean>> tupleInfo =
+				new TupleTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.DOUBLE_TYPE_INFO, BasicTypeInfo.BOOLEAN_TYPE_INFO);
+
+		assertEquals(tupleInfo, TypeInformation.of(new TypeHint<Tuple3<String, Double, Boolean>>(){}));
+	}
+}

flink-java/src/main/java/org/apache/flink/api/java/DataSet.java

@@ -104,7 +104,6 @@
  *
  * @param <T> The type of the DataSet, i.e., the type of the elements of the DataSet.
  */
-
 @Public
 public abstract class DataSet<T> {