Skip to content

F
flink

doujutun3207 / flink
与 Fork 源项目一致

从无法访问的项目Fork

通知 24
Star 0
Fork 0
F
flink

体验新版 GitCode，发现更多精彩内容 >>

提交 21a71586 编写于 作者: S Stephan Ewen
浏览文件
操作

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

上级 7081836e
展开全部 隐藏空白更改
内联 并排
Showing with 503 addition and 612 deletion
flink-core/pom.xml
浏览文件 @ 21a71586
......@@ -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
浏览文件 @ 21a71586
......@@ -16,7 +16,7 @@
* 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;
......@@ -24,22 +24,12 @@ 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 abstract class Keys<T> {
}
}
@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
浏览文件 @ 21a71586
......@@ -22,6 +22,7 @@ import org.apache.flink.api.common.ExecutionConfig;
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
浏览文件 @ 21a71586
......@@ -28,7 +28,9 @@ import java.util.regex.Matcher;
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;
......@@ -36,7 +38,6 @@ 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
浏览文件 @ 21a71586
......@@ -24,7 +24,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
* 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
浏览文件 @ 21a71586
......@@ -24,9 +24,10 @@ import java.util.regex.Matcher;
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
浏览文件 @ 21a71586
......@@ -30,7 +30,9 @@ import java.util.ArrayList;
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.common.typeutils.CompositeType;
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
浏览文件 @ 21a71586
......@@ -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
浏览文件 @ 21a71586
......@@ -23,13 +23,14 @@ import com.esotericsoftware.kryo.Serializer;
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 class Serializers {
}
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 @@ public class Serializers {
}
}
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);
}
}
// --------------------------------------------------------------------------------------------
......
flink-java/src/test/java/org/apache/flink/api/java/operators/ExpressionKeysTest.java → flink-core/src/test/java/org/apache/flink/api/common/operators/ExpressionKeysTest.java
浏览文件 @ 21a71586
......@@ -15,29 +15,33 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.flink.api.java.operators;
package org.apache.flink.api.common.operators;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.operators.Keys.ExpressionKeys;
import org.apache.flink.api.common.operators.Keys.ExpressionKeys;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.api.java.type.extractor.PojoTypeExtractionTest.ComplexNestedClass;
import org.apache.flink.api.java.typeutils.PojoTypeExtractionTest.ComplexNestedClass;
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 org.apache.flink.api.java.operators.SelectorFunctionKeysTest.KeySelector1;
import org.apache.flink.api.java.operators.SelectorFunctionKeysTest.KeySelector3;
import org.apache.flink.api.common.operators.SelectorFunctionKeysTest.KeySelector1;
import org.apache.flink.api.common.operators.SelectorFunctionKeysTest.KeySelector3;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.reflect.Whitebox;
......@@ -474,6 +478,4 @@ public class ExpressionKeysTest {
return 0;
}
}
}
flink-java/src/test/java/org/apache/flink/api/java/operators/SelectorFunctionKeysTest.java → flink-core/src/test/java/org/apache/flink/api/common/operators/SelectorFunctionKeysTest.java
浏览文件 @ 21a71586
......@@ -16,7 +16,7 @@
* limitations under the License.
*/
package org.apache.flink.api.java.operators;
package org.apache.flink.api.common.operators;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
......@@ -25,9 +25,10 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.operators.ExpressionKeysTest.PojoWithMultiplePojos;
import org.apache.flink.api.java.operators.ExpressionKeysTest.Pojo1;
import org.apache.flink.api.java.operators.ExpressionKeysTest.Pojo2;
import org.apache.flink.api.common.operators.ExpressionKeysTest.PojoWithMultiplePojos;
import org.apache.flink.api.common.operators.ExpressionKeysTest.Pojo1;
import org.apache.flink.api.common.operators.ExpressionKeysTest.Pojo2;
import org.junit.Assert;
import org.junit.Test;
......
flink-java/src/test/java/org/apache/flink/api/java/tuple/Tuple2Test.java → flink-core/src/test/java/org/apache/flink/api/java/tuple/Tuple2Test.java
浏览文件 @ 21a71586
......@@ -16,7 +16,6 @@
* limitations under the License.
*/
package org.apache.flink.api.java.tuple;
import org.apache.flink.types.NullFieldException;
......
flink-java/src/test/java/org/apache/flink/api/java/typeutils/EitherTypeInfoTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/EitherTypeInfoTest.java
浏览文件 @ 21a71586
......@@ -23,7 +23,8 @@ import static org.junit.Assert.assertNotEquals;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.Either.Right;
import org.apache.flink.types.Either;
import org.apache.flink.types.Either.Right;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
......
flink-java/src/test/java/org/apache/flink/api/java/type/extractor/PojoTypeExtractionTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java
浏览文件 @ 21a71586
......@@ -15,7 +15,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.flink.api.java.type.extractor;
package org.apache.flink.api.java.typeutils;
import java.util.ArrayList;
import java.util.Arrays;
......@@ -28,18 +29,11 @@ import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor;
import org.apache.flink.api.java.ExecutionEnvironment;
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.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.api.java.typeutils.PojoField;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.api.java.typeutils.TypeInfoParserTest.MyWritable;
import org.apache.flink.api.java.typeutils.WritableTypeInfo;
import org.junit.Assert;
import org.junit.Test;
......@@ -559,67 +553,7 @@ public class PojoTypeExtractionTest {
}
}
}
public static class Vertex<K, V> {
private K key1;
private K key2;
private V value;
public Vertex() {}
public Vertex(K key, V value) {
this.key1 = key;
this.key2 = key;
this.value = value;
}
public Vertex(K key1, K key2, V value) {
this.key1 = key1;
this.key2 = key2;
this.value = value;
}
public void setKey1(K key1) {
this.key1 = key1;
}
public void setKey2(K key2) {
this.key2 = key2;
}
public K getKey1() {
return key1;
}
public K getKey2() {
return key2;
}
public void setValue(V value) {
this.value = value;
}
public V getValue() {
return value;
}
}
public static class VertexTyped extends Vertex<Long, Double>{
public VertexTyped(Long l, Double d) {
super(l, d);
}
public VertexTyped() {
}
}
@Test
public void testGetterSetterWithVertex() {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.fromElements(new VertexTyped(0L, 3.0), new VertexTyped(1L, 1.0));
}
public static class MyMapper<T> implements MapFunction<PojoWithGenerics<Long, T>, PojoWithGenerics<T,T>> {
private static final long serialVersionUID = 1L;
......@@ -636,7 +570,8 @@ public class PojoTypeExtractionTest {
MapFunction<?, ?> function = new MyMapper<String>();
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation)
TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.PojoTypeExtractionTest$PojoWithGenerics<key=int,field1=Long,field2=String>"));
TypeInfoParser.parse("org.apache.flink.api.java.typeutils.PojoTypeExtractionTest$PojoWithGenerics<key=int,field1=Long,field2=String>"));
Assert.assertTrue(ti instanceof PojoTypeInfo<?>);
PojoTypeInfo<?> pti = (PojoTypeInfo<?>) ti;
for(int i = 0; i < pti.getArity(); i++) {
......@@ -710,7 +645,8 @@ public class PojoTypeExtractionTest {
MapFunction<?, ?> function = new MyMapper3<Boolean, Character>();
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation)
TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.PojoTypeExtractionTest$PojoTuple<extraField=char,f0=boolean,f1=boolean,f2=long>"));
TypeInfoParser.parse("org.apache.flink.api.java.typeutils.PojoTypeExtractionTest$PojoTuple<extraField=char,f0=boolean,f1=boolean,f2=long>"));
Assert.assertTrue(ti instanceof TupleTypeInfo<?>);
TupleTypeInfo<?> tti = (TupleTypeInfo<?>) ti;
Assert.assertEquals(BasicTypeInfo.CHAR_TYPE_INFO, tti.getTypeAt(0));
......@@ -735,7 +671,7 @@ public class PojoTypeExtractionTest {
MapFunction<?, ?> function = new MyMapper4<Byte>();
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation)
TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.PojoTypeExtractionTest$PojoWithParameterizedFields1<field=Tuple2<byte,byte>>"));
TypeInfoParser.parse("org.apache.flink.api.java.typeutils.PojoTypeExtractionTest$PojoWithParameterizedFields1<field=Tuple2<byte,byte>>"));
Assert.assertEquals(BasicTypeInfo.BYTE_TYPE_INFO, ti);
}
......@@ -757,8 +693,8 @@ public class PojoTypeExtractionTest {
MapFunction<?, ?> function = new MyMapper5<Byte>();
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation)
TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.PojoTypeExtractionTest$PojoWithParameterizedFields2<"
+ "field=org.apache.flink.api.java.type.extractor.PojoTypeExtractionTest$PojoWithGenerics<key=int,field1=byte,field2=byte>"
TypeInfoParser.parse("org.apache.flink.api.java.typeutils.PojoTypeExtractionTest$PojoWithParameterizedFields2<"
+ "field=org.apache.flink.api.java.typeutils.PojoTypeExtractionTest$PojoWithGenerics<key=int,field1=byte,field2=byte>"
+ ">"));
Assert.assertEquals(BasicTypeInfo.BYTE_TYPE_INFO, ti);
}
......@@ -781,7 +717,7 @@ public class PojoTypeExtractionTest {
MapFunction<?, ?> function = new MyMapper6<Integer>();
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation)
TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.PojoTypeExtractionTest$PojoWithParameterizedFields3<"
TypeInfoParser.parse("org.apache.flink.api.java.typeutils.PojoTypeExtractionTest$PojoWithParameterizedFields3<"
+ "field=int[]"
+ ">"));
Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, ti);
......@@ -805,7 +741,7 @@ public class PojoTypeExtractionTest {
MapFunction<?, ?> function = new MyMapper7<Integer>();
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation)
TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.PojoTypeExtractionTest$PojoWithParameterizedFields4<"
TypeInfoParser.parse("org.apache.flink.api.java.typeutils.PojoTypeExtractionTest$PojoWithParameterizedFields4<"
+ "field=Tuple1<int>[]"
+ ">"));
Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, ti);
......
flink-java/src/test/java/org/apache/flink/api/java/type/extractor/PojoTypeInformationTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeInformationTest.java
浏览文件 @ 21a71586
......@@ -17,7 +17,7 @@
*/
package org.apache.flink.api.java.type.extractor;
package org.apache.flink.api.java.typeutils;
import static org.junit.Assert.assertTrue;
......
flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorInputFormatsTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorInputFormatsTest.java
浏览文件 @ 21a71586
......@@ -16,10 +16,7 @@
* limitations under the License.
*/
package org.apache.flink.api.java.type.extractor;
import static org.junit.Assert.*;
package org.apache.flink.api.java.typeutils;
import java.io.IOException;
......@@ -30,13 +27,13 @@ import org.apache.flink.api.common.io.statistics.BaseStatistics;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputSplit;
import org.junit.Test;
import static org.junit.Assert.*;
@SuppressWarnings("serial")
public class TypeExtractorInputFormatsTest {
......
flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java
浏览文件 @ 21a71586
......@@ -16,7 +16,7 @@
* limitations under the License.
*/
package org.apache.flink.api.java.type.extractor;
package org.apache.flink.api.java.typeutils;
import java.io.DataInput;
import java.io.DataOutput;
......@@ -38,7 +38,6 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple0;
......@@ -46,30 +45,20 @@ 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.api.java.tuple.Tuple9;
import org.apache.flink.api.java.typeutils.Either;
import org.apache.flink.api.java.typeutils.EitherTypeInfo;
import org.apache.flink.api.java.typeutils.EnumTypeInfo;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.api.java.typeutils.MissingTypeInfo;
import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.api.java.typeutils.ValueTypeInfo;
import org.apache.flink.api.java.typeutils.WritableTypeInfo;
import org.apache.flink.types.Either;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.DoubleValue;
import org.apache.flink.types.IntValue;
import org.apache.flink.types.StringValue;
import org.apache.flink.types.Value;
import org.apache.flink.util.Collector;
import org.apache.hadoop.io.Writable;
import org.junit.Assert;
import org.junit.Test;
@SuppressWarnings("serial")
public class TypeExtractorTest {
......@@ -97,7 +86,7 @@ public class TypeExtractorTest {
Assert.assertEquals(ti, TypeExtractor.getForClass(Boolean.class));
// use getForObject()
Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, TypeExtractor.getForObject(Boolean.valueOf(true)));
Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, TypeExtractor.getForObject(true));
}
public static class MyWritable implements Writable {
......@@ -356,7 +345,9 @@ public class TypeExtractorTest {
}
};
TypeInformation<?> ti = TypeExtractor.getCrossReturnTypes(function, (TypeInformation) TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.TypeExtractorTest$CustomType"), (TypeInformation) TypeInfoParser.parse("Integer"));
TypeInformation<?> ti = TypeExtractor.getCrossReturnTypes(function,
(TypeInformation) TypeInfoParser.parse("org.apache.flink.api.java.typeutils.TypeExtractorTest$CustomType"),
(TypeInformation) TypeInfoParser.parse("Integer"));
Assert.assertFalse(ti.isBasicType());
Assert.assertFalse(ti.isTupleType());
......@@ -408,7 +399,8 @@ public class TypeExtractorTest {
};
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation) TypeInfoParser.parse("Tuple2<Long,org.apache.flink.api.java.type.extractor.TypeExtractorTest$CustomType>"));
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function,
(TypeInformation) TypeInfoParser.parse("Tuple2<Long,org.apache.flink.api.java.typeutils.TypeExtractorTest$CustomType>"));
Assert.assertTrue(ti.isTupleType());
Assert.assertEquals(2, ti.getArity());
......@@ -849,25 +841,6 @@ public class TypeExtractorTest {
}
}
@SuppressWarnings({"unchecked", "rawtypes"})
@Test
public void testFunctionWithMissingGenericsAndReturns() {
RichMapFunction function = new RichMapFunction() {
private static final long serialVersionUID = 1L;
@Override
public Object map(Object value) throws Exception {
return null;
}
};
TypeInformation info = ExecutionEnvironment.getExecutionEnvironment()
.fromElements("arbitrary", "data")
.map(function).returns("String").getResultType();
Assert.assertEquals(TypeInfoParser.parse("String"), info);
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testFunctionDependingOnInputAsSuperclass() {
......@@ -1256,7 +1229,8 @@ public class TypeExtractorTest {
}
};
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation) TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.TypeExtractorTest$CustomArrayObject[]"));
TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function,
(TypeInformation) TypeInfoParser.parse("org.apache.flink.api.java.typeutils.TypeExtractorTest$CustomArrayObject[]"));
Assert.assertTrue(ti instanceof ObjectArrayTypeInfo<?, ?>);
Assert.assertEquals(CustomArrayObject.class, ((ObjectArrayTypeInfo<?, ?>) ti).getComponentInfo().getTypeClass());
......@@ -1816,11 +1790,13 @@ public class TypeExtractorTest {
return null;
}
};
ti = TypeExtractor.getMapReturnTypes((MapFunction)function, TypeInfoParser.parse("org.apache.flink.api.java.type.extractor.TypeExtractorTest$CustomType<"
+ "myField1=String,myField2=int"
+ ">[][][]"));
ti = TypeExtractor.getMapReturnTypes((MapFunction)function,
TypeInfoParser.parse("org.apache.flink.api.java.typeutils.TypeExtractorTest$CustomType<"
+ "myField1=String,myField2=int"
+ ">[][][]"));
Assert.assertEquals("ObjectArrayTypeInfo<ObjectArrayTypeInfo<ObjectArrayTypeInfo<"
+ "PojoType<org.apache.flink.api.java.type.extractor.TypeExtractorTest$CustomType, fields = [myField1: String, myField2: Integer]>"
+ "PojoType<org.apache.flink.api.java.typeutils.TypeExtractorTest$CustomType, fields = [myField1: String, myField2: Integer]>"
+ ">>>", ti.toString());
// generic array
......
flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerTest.java
浏览文件 @ 21a71586
......@@ -20,15 +20,15 @@ package org.apache.flink.api.java.typeutils.runtime;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
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 org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeutils.SerializerTestInstance;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.Either;
import org.apache.flink.types.Either;
import org.apache.flink.api.java.typeutils.EitherTypeInfo;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.junit.Test;
......
flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoComparatorTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoComparatorTest.java
浏览文件 @ 21a71586
......@@ -26,7 +26,7 @@ import org.apache.flink.api.common.typeutils.ComparatorTestBase;
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.operators.Keys.ExpressionKeys;
import org.apache.flink.api.common.operators.Keys.ExpressionKeys;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.junit.Assert;
......
flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java
浏览文件 @ 21a71586
......@@ -29,8 +29,8 @@ import org.apache.flink.api.common.typeutils.SerializerTestBase;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor;
import org.apache.flink.api.java.operators.Keys.ExpressionKeys;
import org.apache.flink.api.java.operators.Keys.IncompatibleKeysException;
import org.apache.flink.api.common.operators.Keys.ExpressionKeys;
import org.apache.flink.api.common.operators.Keys.IncompatibleKeysException;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
......
flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassComparatorTest.java → flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassComparatorTest.java
浏览文件 @ 21a71586
......@@ -24,7 +24,7 @@ import org.apache.flink.api.common.typeutils.ComparatorTestBase;
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.operators.Keys.ExpressionKeys;
import org.apache.flink.api.common.operators.Keys.ExpressionKeys;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.junit.Assert;
......
flink-java/pom.xml
浏览文件 @ 21a71586
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或