You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2024/05/27 02:02:21 UTC
(flink) branch master updated (c3221a649b9 -> f860631c523)
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from c3221a649b9 [hotfix][sqlgateway] Do not introduce new method in SqlGatewayEndpointFactory#Context
new 73676da5958 [FLINK-34123][core][type] Introduce built-in serialization support for Map, List, and Collection
new f860631c523 [FLINK-34123][docs][type] Add doc for built-in serialization support for Map, List, and Collection
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../serialization/types_serialization.md | 25 ++++--
.../flink/api/java/typeutils/TypeExtractor.java | 29 +++++++
.../api/java/typeutils/PojoTypeExtractionTest.java | 2 +-
.../api/java/typeutils/TypeExtractorTest.java | 92 ++++++++++++++++++++--
.../org/apache/flink/types/PojoTestUtilsTest.java | 2 +-
5 files changed, 138 insertions(+), 12 deletions(-)
(flink) 02/02: [FLINK-34123][docs][type] Add doc for built-in serialization support for Map, List, and Collection
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f860631c523c1d446c0d01046f0fbe6055174dc6
Author: Zhanghao Chen <m1...@outlook.com>
AuthorDate: Mon Apr 8 22:07:01 2024 +0800
[FLINK-34123][docs][type] Add doc for built-in serialization support for Map, List, and Collection
---
.../serialization/types_serialization.md | 25 +++++++++++++++++-----
1 file changed, 20 insertions(+), 5 deletions(-)
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md b/docs/content/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md
index 4b3dc6f76c2..cf97beef619 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md
@@ -37,15 +37,16 @@ Flink places some restrictions on the type of elements that can be in a DataStre
The reason for this is that the system analyzes the types to determine
efficient execution strategies.
-There are seven different categories of data types:
+There are eight different categories of data types:
1. **Java Tuples** and **Scala Case Classes**
2. **Java POJOs**
3. **Primitive Types**
-4. **Regular Classes**
-5. **Values**
-6. **Hadoop Writables**
-7. **Special Types**
+4. **Common Collection Types**
+5. **Regular Classes**
+6. **Values**
+7. **Hadoop Writables**
+8. **Special Types**
#### Tuples and Case Classes
@@ -167,6 +168,20 @@ input.keyBy(_.word)
Flink supports all Java and Scala primitive types such as `Integer`, `String`, and `Double`.
+#### Common Collection Types
+
+Flink comes with dedicated serialization support for common Java collection types, which is more efficient than going
+through a general purpose serialization framework. Currently, only `Map`, `List` and its super interface `Collection`
+are supported. To utilize it, you need to declare the collection type with:
+
+1. Concrete type arguments: e.g. `List<String>` but not `List`, `List<T>`, or `List<?>`, as Flink needs them to dispatch
+ serialization of the element types.
+2. Interface types: e.g. `List<String>` but not `LinkedList<String>`, as Flink does not preserve the underlying
+ implementation types across serialization.
+
+Other nonqualified collection types will be handled by Flink as general class types. If the implementation types are
+also required to be preserved, you also need to register it with a custom serializer.
+
#### General Class Types
Flink supports most Java and Scala classes (API and custom).
(flink) 01/02: [FLINK-34123][core][type] Introduce built-in serialization support for Map, List, and Collection
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 73676da59584f360c02d139a916033e8967583b3
Author: Zhanghao Chen <m1...@outlook.com>
AuthorDate: Mon Apr 8 21:21:21 2024 +0800
[FLINK-34123][core][type] Introduce built-in serialization support for Map, List, and Collection
---
.../flink/api/java/typeutils/TypeExtractor.java | 29 +++++++
.../api/java/typeutils/PojoTypeExtractionTest.java | 2 +-
.../api/java/typeutils/TypeExtractorTest.java | 92 ++++++++++++++++++++--
.../org/apache/flink/types/PojoTestUtilsTest.java | 2 +-
4 files changed, 118 insertions(+), 7 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
index fd7978d3235..ca77ab1719a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
@@ -42,6 +42,7 @@ import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.common.typeutils.CompositeType;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
@@ -66,6 +67,7 @@ import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.lang.reflect.TypeVariable;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -1970,6 +1972,33 @@ public class TypeExtractor {
return new EnumTypeInfo(clazz);
}
+ // check for parameterized Collections, requirement:
+ // 1. Interface types: the underlying implementation types are not preserved across
+ // serialization
+ // 2. Concrete type arguments: Flink needs them to dispatch serialization of element types
+ // Example:
+ // - OK: List<String>, Collection<String>
+ // - not OK: LinkedList<String> (implementation type), List (raw type), List<T> (generic
+ // type argument), or List<?> (wildcard type argument)
+ if (parameterizedType != null) {
+ Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
+ boolean allTypeArgumentsConcrete =
+ Arrays.stream(actualTypeArguments).allMatch(arg -> arg instanceof Class<?>);
+ if (allTypeArgumentsConcrete) {
+ if (clazz.isAssignableFrom(Map.class)) {
+ Class<?> keyClass = (Class<?>) actualTypeArguments[0];
+ Class<?> valueClass = (Class<?>) actualTypeArguments[1];
+ TypeInformation<?> keyTypeInfo = createTypeInfo(keyClass);
+ TypeInformation<?> valueTypeInfo = createTypeInfo(valueClass);
+ return (TypeInformation<OUT>) Types.MAP(keyTypeInfo, valueTypeInfo);
+ } else if (clazz.isAssignableFrom(List.class)) {
+ Class<?> elementClass = (Class<?>) actualTypeArguments[0];
+ TypeInformation<?> elementTypeInfo = createTypeInfo(elementClass);
+ return (TypeInformation<OUT>) Types.LIST(elementTypeInfo);
+ }
+ }
+ }
+
// special case for POJOs generated by Avro.
if (hasSuperclass(clazz, AVRO_SPECIFIC_RECORD_BASE_CLASS)) {
return AvroUtils.getAvroUtils().createAvroTypeInfo(clazz);
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java
index 22d62bf622e..203d623e336 100644
--- a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java
@@ -421,7 +421,7 @@ public class PojoTypeExtractionTest {
fail("already seen");
}
collectionSeen = true;
- assertThat(field.getTypeInformation()).isEqualTo(new GenericTypeInfo(List.class));
+ assertThat(field.getTypeInformation()).isEqualTo(new ListTypeInfo<>(String.class));
} else {
fail("Unexpected field " + field);
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java
index 458eb5fcef2..471e50b0f2f 100644
--- a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java
@@ -64,7 +64,10 @@ import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -2317,15 +2320,15 @@ public class TypeExtractorTest {
@SuppressWarnings({"unchecked", "rawtypes"})
@Test
void testGenericTypeWithSubclassInput() {
- Map<String, Object> inputMap = new HashMap<>();
+ HashMap<String, Object> inputMap = new LinkedHashMap<>();
inputMap.put("a", "b");
TypeInformation<?> inputType = TypeExtractor.createTypeInfo(inputMap.getClass());
MapFunction<?, ?> function =
- new MapFunction<Map<String, Object>, Map<String, Object>>() {
+ new MapFunction<HashMap<String, Object>, HashMap<String, Object>>() {
@Override
- public Map<String, Object> map(Map<String, Object> stringObjectMap)
+ public HashMap<String, Object> map(HashMap<String, Object> stringObjectMap)
throws Exception {
return stringObjectMap;
}
@@ -2333,7 +2336,7 @@ public class TypeExtractorTest {
TypeInformation<?> ti =
TypeExtractor.getMapReturnTypes(function, (TypeInformation) inputType);
- TypeInformation<?> expected = TypeExtractor.createTypeInfo(Map.class);
+ TypeInformation<?> expected = TypeExtractor.createTypeInfo(HashMap.class);
assertThat(ti).isEqualTo(expected);
}
@@ -2345,7 +2348,9 @@ public class TypeExtractorTest {
TypeInformation<?> inputType = TypeExtractor.createTypeInfo(Map.class);
MapFunction<?, ?> function =
- (MapFunction<HashMap<String, Object>, Map<String, Object>>)
+ (MapFunction<
+ LinkedHashMap<String, Object>,
+ Map<String, Object>>)
stringObjectMap -> stringObjectMap;
TypeExtractor.getMapReturnTypes(function, (TypeInformation) inputType);
})
@@ -2506,4 +2511,81 @@ public class TypeExtractorTest {
assertThat(TypeExtractor.getForObject(Timestamp.valueOf("1998-12-12 12:37:45")))
.isEqualTo(SqlTimeTypeInfo.TIMESTAMP);
}
+
+ @SuppressWarnings({"rawtypes"})
+ public static class PojoWithCollections<T> {
+ // Supported collection types with concrete type arguments, expected built-in serialization
+ // support
+ public Map<String, Integer> mapVal = new HashMap<>();
+ public List<String> listVal = new ArrayList<>();
+ public Collection<String> collectionVal = new ArrayList<>();
+
+ // Collection fields with unsupported collection types, treated as generic types
+ public LinkedList<String> linkedListVal = new LinkedList<>();
+
+ // Collection fields with raw type, treated as generic types
+ public List rawListVal = new ArrayList<>();
+
+ // Collection fields with generic type arguments, treated as generic types
+ public List<T> genericListVal = new ArrayList<>();
+
+ // Collection fields with wildcard type arguments, treated as generic types
+ public List<?> wildcardListVal = new ArrayList<>();
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ @Test
+ public <T> void testCollectionTypes() {
+ MapFunction<?, ?> function =
+ new MapFunction<PojoWithCollections<T>, PojoWithCollections<T>>() {
+ @Override
+ public PojoWithCollections map(PojoWithCollections<T> value) {
+ return null;
+ }
+ };
+ TypeInformation<?> ti =
+ TypeExtractor.getMapReturnTypes(
+ function,
+ (TypeInformation)
+ TypeInformation.of(new TypeHint<PojoWithCollections<T>>() {}));
+ assertThat(ti).isInstanceOf(PojoTypeInfo.class);
+ testCollectionTypesInternal(ti);
+
+ // use getForClass()
+ TypeInformation<?> ti2 = TypeExtractor.getForClass(PojoWithCollections.class);
+ assertThat(ti2).isInstanceOf(PojoTypeInfo.class);
+ testCollectionTypesInternal(ti2);
+
+ // use getForObject()
+ PojoWithCollections<T> t = new PojoWithCollections<>();
+ TypeInformation<?> ti3 = TypeExtractor.getForObject(t);
+ assertThat(ti3).isInstanceOf(PojoTypeInfo.class);
+ testCollectionTypesInternal(ti3);
+ }
+
+ private void testCollectionTypesInternal(TypeInformation<?> ti) {
+ PojoTypeInfo<?> pojoTi = (PojoTypeInfo<?>) ti;
+ assertThat(pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("mapVal")).getTypeInformation())
+ .isInstanceOf(MapTypeInfo.class);
+ assertThat(pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("listVal")).getTypeInformation())
+ .isInstanceOf(ListTypeInfo.class);
+ assertThat(
+ pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("collectionVal"))
+ .getTypeInformation())
+ .isInstanceOf(ListTypeInfo.class);
+ assertThat(
+ pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("linkedListVal"))
+ .getTypeInformation())
+ .isInstanceOf(GenericTypeInfo.class);
+ assertThat(pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("rawListVal")).getTypeInformation())
+ .isInstanceOf(GenericTypeInfo.class);
+ assertThat(
+ pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("genericListVal"))
+ .getTypeInformation())
+ .isInstanceOf(GenericTypeInfo.class);
+ assertThat(
+ pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("wildcardListVal"))
+ .getTypeInformation())
+ .isInstanceOf(GenericTypeInfo.class);
+ }
}
diff --git a/flink-test-utils-parent/flink-test-utils/src/test/java/org/apache/flink/types/PojoTestUtilsTest.java b/flink-test-utils-parent/flink-test-utils/src/test/java/org/apache/flink/types/PojoTestUtilsTest.java
index 6d9f399ea78..e0217bc04af 100644
--- a/flink-test-utils-parent/flink-test-utils/src/test/java/org/apache/flink/types/PojoTestUtilsTest.java
+++ b/flink-test-utils-parent/flink-test-utils/src/test/java/org/apache/flink/types/PojoTestUtilsTest.java
@@ -62,6 +62,6 @@ class PojoTestUtilsTest {
}
public static class PojoRequiringKryo {
- public List<Integer> x;
+ public List<?> x;
}
}