[FLINK-3303] [core] Move all type utilities to flink-core

flink-core/pom.xml

@@ -40,12 +40,6 @@ under the License.
 			<artifactId>flink-annotations</artifactId>
 			<version>${project.version}</version>
 		</dependency>
-
-		<dependency>
-			<groupId>org.apache.flink</groupId>
-			<artifactId>${shading-artifact.name}</artifactId>
-			<version>${project.version}</version>
-		</dependency>
 
 		<dependency>
 			<groupId>commons-collections</groupId>
@@ -56,20 +50,61 @@ under the License.
 		<dependency>
 			<groupId>com.esotericsoftware.kryo</groupId>
 			<artifactId>kryo</artifactId>
+			<!-- managed version -->
 		</dependency>
 
+		<!-- Avro is needed for the interoperability with Avro types for serialization -->
+		<dependency>
+			<groupId>org.apache.avro</groupId>
+			<artifactId>avro</artifactId>
+			<!-- managed version -->
+			<exclusions>
+				<exclusion>
+					<groupId>org.xerial.snappy</groupId>
+					<artifactId>snappy-java</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>org.apache.commons</groupId>
+					<artifactId>commons-compress</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+
+		<!-- Hadoop is only needed here for serialization interoperability with the Writable type -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>${shading-artifact.name}</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
 		<dependency>
 			<groupId>com.google.guava</groupId>
 			<artifactId>guava</artifactId>
 			<version>${guava.version}</version>
 		</dependency>
 
+		<!-- test depedencies -->
+
 		<dependency>
 			<groupId>commons-io</groupId>
 			<artifactId>commons-io</artifactId>
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>joda-time</groupId>
+			<artifactId>joda-time</artifactId>
+			<version>2.5</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.joda</groupId>
+			<artifactId>joda-convert</artifactId>
+			<version>1.7</version>
+			<scope>test</scope>
+		</dependency>
+
 	</dependencies>
 
 	<build>

flink-java/src/main/java/org/apache/flink/api/java/operators/Keys.java → flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java

@@ -16,30 +16,20 @@
  * limitations under the License.
  */
 
-package org.apache.flink.api.java.operators;
+package org.apache.flink.api.common.operators;
 
 import java.util.ArrayList;
 import java.util.List;
 
 import com.google.common.base.Joiner;
 
 import org.apache.flink.api.common.InvalidProgramException;
-import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.functions.Partitioner;
-import org.apache.flink.api.common.operators.Operator;
-import org.apache.flink.api.common.operators.UnaryOperatorInformation;
-import org.apache.flink.api.common.operators.base.MapOperatorBase;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor;
 import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.api.java.operators.translation.KeyExtractingMapper;
-import org.apache.flink.api.java.operators.translation.KeyRemovingMapper;
-import org.apache.flink.api.java.operators.translation.TwoKeyExtractingMapper;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.api.java.typeutils.GenericTypeInfo;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 
 import com.google.common.base.Preconditions;
@@ -177,87 +167,6 @@ public <E> void validateCustomPartitioner(Partitioner<E> partitioner, TypeInform
 			}
 		}
 
-		@SuppressWarnings("unchecked")
-		public static <T, K> Operator<Tuple2<K, T>> appendKeyExtractor(
-			Operator<T> input,
-			SelectorFunctionKeys<T, K> key)
-		{
-
-			TypeInformation<T> inputType = key.getInputType();
-			TypeInformation<Tuple2<K, T>> typeInfoWithKey = createTypeWithKey(key);
-			KeyExtractingMapper<T, K> extractor = new KeyExtractingMapper(key.getKeyExtractor());
-
-			MapOperatorBase<T, Tuple2<K, T>, MapFunction<T, Tuple2<K, T>>> mapper =
-				new MapOperatorBase<T, Tuple2<K, T>, MapFunction<T, Tuple2<K, T>>>(
-					extractor,
-					new UnaryOperatorInformation(inputType, typeInfoWithKey),
-					"Key Extractor"
-				);
-
-			mapper.setInput(input);
-			mapper.setParallelism(input.getParallelism());
-
-			return mapper;
-		}
-
-		@SuppressWarnings("unchecked")
-		public static <T, K1, K2> Operator<Tuple3<K1, K2, T>> appendKeyExtractor(
-			Operator<T> input,
-			SelectorFunctionKeys<T, K1> key1,
-			SelectorFunctionKeys<T, K2> key2)
-		{
-
-			TypeInformation<T> inputType = key1.getInputType();
-			TypeInformation<Tuple3<K1, K2, T>> typeInfoWithKey = createTypeWithKey(key1, key2);
-			TwoKeyExtractingMapper<T, K1, K2> extractor =
-				new TwoKeyExtractingMapper<>(key1.getKeyExtractor(), key2.getKeyExtractor());
-
-			MapOperatorBase<T, Tuple3<K1, K2, T>, MapFunction<T, Tuple3<K1, K2, T>>> mapper =
-				new MapOperatorBase<T, Tuple3<K1, K2, T>, MapFunction<T, Tuple3<K1, K2, T>>>(
-					extractor,
-					new UnaryOperatorInformation<>(inputType, typeInfoWithKey),
-					"Key Extractor"
-				);
-
-			mapper.setInput(input);
-			mapper.setParallelism(input.getParallelism());
-
-			return mapper;
-		}
-
-		public static <T, K> org.apache.flink.api.common.operators.SingleInputOperator<?, T, ?> appendKeyRemover(
-			Operator<Tuple2<K, T>> inputWithKey,
-			SelectorFunctionKeys<T, K> key)
-		{
-
-			TypeInformation<T> inputType = key.getInputType();
-			TypeInformation<Tuple2<K, T>> typeInfoWithKey = createTypeWithKey(key);
-
-			MapOperatorBase<Tuple2<K, T>, T, MapFunction<Tuple2<K, T>, T>> mapper =
-				new MapOperatorBase<Tuple2<K, T>, T, MapFunction<Tuple2<K, T>, T>>(
-					new KeyRemovingMapper<T, K>(),
-					new UnaryOperatorInformation<>(typeInfoWithKey, inputType),
-					"Key Remover"
-				);
-			mapper.setInput(inputWithKey);
-			mapper.setParallelism(inputWithKey.getParallelism());
-
-			return mapper;
-		}
-
-		public static <T, K> TypeInformation<Tuple2<K, T>> createTypeWithKey(
-			SelectorFunctionKeys<T, K> key)
-		{
-			return new TupleTypeInfo<>(key.getKeyType(), key.getInputType());
-		}
-
-		public static <T, K1, K2> TypeInformation<Tuple3<K1, K2, T>> createTypeWithKey(
-			SelectorFunctionKeys<T, K1> key1,
-			SelectorFunctionKeys<T, K2> key2)
-		{
-			return new TupleTypeInfo<>(key1.getKeyType(), key2.getKeyType(), key1.getInputType());
-		}
-
 		@Override
 		public String toString() {
 			return "Key function (Type: " + keyType + ")";

flink-java/src/main/java/org/apache/flink/api/java/typeutils/EitherTypeInfo.java → flink-core/src/main/java/org/apache/flink/api/java/typeutils/EitherTypeInfo.java

@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.typeutils.runtime.EitherSerializer;
+import org.apache.flink.types.Either;
 
 /**
  * A {@link TypeInformation} for the {@link Either} type of the Java API.

flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java → flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java

@@ -28,15 +28,16 @@
 import java.util.regex.Pattern;
 
 import com.google.common.base.Preconditions;
+
 import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.operators.Keys.ExpressionKeys;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.common.typeutils.TypeComparator;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.typeutils.runtime.AvroSerializer;
 import org.apache.flink.api.java.typeutils.runtime.PojoComparator;
 import org.apache.flink.api.java.typeutils.runtime.PojoSerializer;
-import org.apache.flink.api.java.operators.Keys.ExpressionKeys;
 
 import com.google.common.base.Joiner;
 import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;

flink-java/src/main/java/org/apache/flink/api/java/typeutils/ResultTypeQueryable.java → flink-core/src/main/java/org/apache/flink/api/java/typeutils/ResultTypeQueryable.java

@@ -24,7 +24,7 @@
  * This interface can be implemented by functions and input formats to tell the framework
  * about their produced data type. This method acts as an alternative to the reflection analysis
  * that is otherwise performed and is useful in situations where the produced data type may vary
- * depending on parameterization.
+ * depending on parametrization.
  */
 public interface ResultTypeQueryable<T> {
 

flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java → flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java

@@ -24,9 +24,10 @@
 import java.util.regex.Pattern;
 
 import com.google.common.base.Preconditions;
+
+import org.apache.flink.api.common.operators.Keys.ExpressionKeys;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
-import org.apache.flink.api.java.operators.Keys.ExpressionKeys;
 
 public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
 
@@ -35,7 +36,7 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
 	private final static String REGEX_FIELD = "(f?)([0-9]+)";
 	private final static String REGEX_NESTED_FIELDS = "("+REGEX_FIELD+")(\\.(.+))?";
 	private final static String REGEX_NESTED_FIELDS_WILDCARD = REGEX_NESTED_FIELDS
-			+"|\\"+ExpressionKeys.SELECT_ALL_CHAR
+			+"|\\"+ ExpressionKeys.SELECT_ALL_CHAR
 			+"|\\"+ExpressionKeys.SELECT_ALL_CHAR_SCALA;
 
 	private static final Pattern PATTERN_FIELD = Pattern.compile(REGEX_FIELD);

flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java → flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java

@@ -30,7 +30,9 @@
 import java.util.List;
 
 import org.apache.avro.specific.SpecificRecordBase;
+
 import org.apache.commons.lang3.ClassUtils;
+
 import org.apache.flink.api.common.functions.CoGroupFunction;
 import org.apache.flink.api.common.functions.CrossFunction;
 import org.apache.flink.api.common.functions.FlatJoinFunction;
@@ -54,9 +56,12 @@
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple0;
+import org.apache.flink.types.Either;
 import org.apache.flink.types.Value;
 import org.apache.flink.util.Collector;
+
 import org.apache.hadoop.io.Writable;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializer.java → flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializer.java

@@ -18,13 +18,13 @@
 
 package org.apache.flink.api.java.typeutils.runtime;
 
-import static org.apache.flink.api.java.typeutils.Either.Left;
-import static org.apache.flink.api.java.typeutils.Either.Right;
+import static org.apache.flink.types.Either.Left;
+import static org.apache.flink.types.Either.Right;
 
 import java.io.IOException;
 
 import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.typeutils.Either;
+import org.apache.flink.types.Either;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 

flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java → flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java

@@ -23,13 +23,14 @@
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
 import com.esotericsoftware.kryo.serializers.CollectionSerializer;
+
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.specific.SpecificRecordBase;
+
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
-import org.apache.flink.api.java.Utils;
 import org.apache.flink.api.java.typeutils.GenericTypeInfo;
 import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
@@ -64,7 +65,7 @@ public static void recursivelyRegisterType(TypeInformation<?> typeInfo, Executio
 		}
 		else if (typeInfo instanceof CompositeType) {
 			List<GenericTypeInfo<?>> genericTypesInComposite = new ArrayList<>();
-			Utils.getContainedGenericTypes((CompositeType<?>)typeInfo, genericTypesInComposite);
+			getContainedGenericTypes((CompositeType<?>)typeInfo, genericTypesInComposite);
 			for (GenericTypeInfo<?> gt : genericTypesInComposite) {
 				Serializers.recursivelyRegisterType(gt.getTypeClass(), config, alreadySeen);
 			}
@@ -127,41 +128,38 @@ else if (fieldType instanceof Class) {
 		}
 	}
 
-	private static void checkAndAddSerializerForTypeAvro(ExecutionConfig reg, Class<?> type) {
-		if (GenericData.Record.class.isAssignableFrom(type)) {
-			registerGenericAvro(reg);
-		}
-		if (SpecificRecordBase.class.isAssignableFrom(type)) {
-			@SuppressWarnings("unchecked")
-			Class<? extends SpecificRecordBase> specRecordClass = (Class<? extends SpecificRecordBase>) type;
-			registerSpecificAvro(reg, specRecordClass);
-		}
-	}
-
 	/**
-	 * Register these serializers for using Avro's {@link GenericData.Record} and classes
-	 * implementing {@link org.apache.avro.specific.SpecificRecordBase}
+	 * Returns all GenericTypeInfos contained in a composite type.
+	 *
+	 * @param typeInfo {@link CompositeType}
 	 */
-	private static void registerGenericAvro(ExecutionConfig reg) {
-		// Avro POJOs contain java.util.List which have GenericData.Array as their runtime type
-		// because Kryo is not able to serialize them properly, we use this serializer for them
-		reg.registerTypeWithKryoSerializer(GenericData.Array.class, SpecificInstanceCollectionSerializerForArrayList.class);
-
-		// We register this serializer for users who want to use untyped Avro records (GenericData.Record).
-		// Kryo is able to serialize everything in there, except for the Schema.
-		// This serializer is very slow, but using the GenericData.Records of Kryo is in general a bad idea.
-		// we add the serializer as a default serializer because Avro is using a private sub-type at runtime.
-		reg.addDefaultKryoSerializer(Schema.class, AvroSchemaSerializer.class);
+	private static void getContainedGenericTypes(CompositeType<?> typeInfo, List<GenericTypeInfo<?>> target) {
+		for (int i = 0; i < typeInfo.getArity(); i++) {
+			TypeInformation<?> type = typeInfo.getTypeAt(i);
+			if (type instanceof CompositeType) {
+				getContainedGenericTypes((CompositeType<?>) type, target);
+			} else if (type instanceof GenericTypeInfo) {
+				if (!target.contains(type)) {
+					target.add((GenericTypeInfo<?>) type);
+				}
+			}
+		}
 	}
+
+	// ------------------------------------------------------------------------
+
+	private static void checkAndAddSerializerForTypeAvro(ExecutionConfig reg, Class<?> type) {
+		if (GenericData.Record.class.isAssignableFrom(type) || SpecificRecordBase.class.isAssignableFrom(type)) {
+			// Avro POJOs contain java.util.List which have GenericData.Array as their runtime type
+			// because Kryo is not able to serialize them properly, we use this serializer for them
+			reg.registerTypeWithKryoSerializer(GenericData.Array.class, SpecificInstanceCollectionSerializerForArrayList.class);
 
-	private static void registerSpecificAvro(ExecutionConfig reg, Class<? extends SpecificRecordBase> avroType) {
-		registerGenericAvro(reg);
-		// This rule only applies if users explicitly use the GenericTypeInformation for the avro types
-		// usually, we are able to handle Avro POJOs with the POJO serializer.
-		// (However only if the GenericData.Array type is registered!)
-
-	//	ClassTag<SpecificRecordBase> tag = scala.reflect.ClassTag$.MODULE$.apply(avroType);
-	//	reg.registerTypeWithKryoSerializer(avroType, com.twitter.chill.avro.AvroSerializer.SpecificRecordSerializer(tag));
+			// We register this serializer for users who want to use untyped Avro records (GenericData.Record).
+			// Kryo is able to serialize everything in there, except for the Schema.
+			// This serializer is very slow, but using the GenericData.Records of Kryo is in general a bad idea.
+			// we add the serializer as a default serializer because Avro is using a private sub-type at runtime.
+			reg.addDefaultKryoSerializer(Schema.class, AvroSchemaSerializer.class);
+		}
 	}
 
 	// --------------------------------------------------------------------------------------------