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 2025/01/14 02:20:45 UTC
(flink) branch master updated (4a6f84be0b2 -> 7588a4dcda5)
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 4a6f84be0b2 [FLINK-37101] Remove legacy internal methods for changing serializer behavior in SerializerConfig
new 24b661f782e [FLINK-34123][core] Introduce built-in serialization support for Map, List, and Collection
new aaa88d4e6b3 [FLINK-34123][docs][type] Add doc for built-in serialization support for Map, List, and Collection
new f46bb434404 [FLINK-35068][core] Introduce built-in serialization support for java.util.Set
new 7588a4dcda5 [FLINK-35555][core] Introduce Nullable{Map/List/Set}TypeInfo for collection type serialization with null value support
The 4 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 | 26 ++-
.../apache/flink/api/common/typeinfo/Types.java | 19 +++
.../api/common/typeutils/base/SetSerializer.java | 181 +++++++++++++++++++++
.../typeutils/base/SetSerializerSnapshot.java | 59 +++++++
.../flink/api/java/typeutils/ListTypeInfo.java | 2 +-
.../api/java/typeutils/NullableListTypeInfo.java | 60 +++++++
.../api/java/typeutils/NullableMapTypeInfo.java | 63 +++++++
.../api/java/typeutils/NullableSetTypeInfo.java | 58 +++++++
.../{ListTypeInfo.java => SetTypeInfo.java} | 40 ++---
.../flink/api/java/typeutils/TypeExtractor.java | 34 ++++
.../typeutils/base/NullableListSerializerTest.java | 78 +++++++++
.../typeutils/base/NullableMapSerializerTest.java | 80 +++++++++
.../typeutils/base/NullableSetSerializerTest.java | 78 +++++++++
.../common/typeutils/base/SetSerializerTest.java | 86 ++++++++++
.../java/typeutils/NullableListTypeInfoTest.java | 35 ++++
.../java/typeutils/NullableMapTypeInfoTest.java | 35 ++++
.../java/typeutils/NullableSetTypeInfoTest.java | 35 ++++
.../api/java/typeutils/PojoTypeExtractionTest.java | 3 +-
.../flink/api/java/typeutils/SetTypeInfoTest.java | 35 ++++
.../api/java/typeutils/TypeExtractorTest.java | 97 ++++++++++-
.../org/apache/flink/types/PojoTestUtilsTest.java | 2 +-
.../TypeSerializerTestCoverageTest.java | 4 +-
22 files changed, 1076 insertions(+), 34 deletions(-)
create mode 100644 flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SetSerializer.java
create mode 100644 flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SetSerializerSnapshot.java
create mode 100644 flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableListTypeInfo.java
create mode 100644 flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableMapTypeInfo.java
create mode 100644 flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableSetTypeInfo.java
copy flink-core/src/main/java/org/apache/flink/api/java/typeutils/{ListTypeInfo.java => SetTypeInfo.java} (71%)
create mode 100644 flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableListSerializerTest.java
create mode 100644 flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableMapSerializerTest.java
create mode 100644 flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableSetSerializerTest.java
create mode 100644 flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/SetSerializerTest.java
create mode 100644 flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableListTypeInfoTest.java
create mode 100644 flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableMapTypeInfoTest.java
create mode 100644 flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableSetTypeInfoTest.java
create mode 100644 flink-core/src/test/java/org/apache/flink/api/java/typeutils/SetTypeInfoTest.java
(flink) 02/04: [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 aaa88d4e6b3d9a493e713c48d9ddcef9b2ce3a34
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 | 26 +++++++++++++++++-----
1 file changed, 21 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 0d4441f4d3f..766024ccd75 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,21 @@ 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 by avoiding analysis and serialization of the type metadata.
+Currently, only `Map`, `List`, `Set` 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) 04/04: [FLINK-35555][core] Introduce Nullable{Map/List/Set}TypeInfo for collection type serialization with null value support
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 7588a4dcda5f76d3606ba5d3982b48ebae8b653a
Author: Zhanghao Chen <m1...@outlook.com>
AuthorDate: Mon Jan 6 21:04:09 2025 +0800
[FLINK-35555][core] Introduce Nullable{Map/List/Set}TypeInfo for collection type serialization with null value support
This closes #25797
---
.../flink/api/java/typeutils/ListTypeInfo.java | 2 +-
.../api/java/typeutils/NullableListTypeInfo.java | 60 ++++++++++++++++
.../api/java/typeutils/NullableMapTypeInfo.java | 63 +++++++++++++++++
.../api/java/typeutils/NullableSetTypeInfo.java | 58 ++++++++++++++++
.../flink/api/java/typeutils/SetTypeInfo.java | 2 +-
.../flink/api/java/typeutils/TypeExtractor.java | 8 +--
.../typeutils/base/NullableListSerializerTest.java | 78 +++++++++++++++++++++
.../typeutils/base/NullableMapSerializerTest.java | 80 ++++++++++++++++++++++
.../typeutils/base/NullableSetSerializerTest.java | 78 +++++++++++++++++++++
.../java/typeutils/NullableListTypeInfoTest.java | 35 ++++++++++
.../java/typeutils/NullableMapTypeInfoTest.java | 35 ++++++++++
.../java/typeutils/NullableSetTypeInfoTest.java | 35 ++++++++++
.../api/java/typeutils/PojoTypeExtractionTest.java | 3 +-
13 files changed, 530 insertions(+), 7 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ListTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ListTypeInfo.java
index e58544f337a..3385e3ca173 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ListTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ListTypeInfo.java
@@ -34,7 +34,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* @param <T> The type of the elements in the list.
*/
@PublicEvolving
-public final class ListTypeInfo<T> extends TypeInformation<List<T>> {
+public class ListTypeInfo<T> extends TypeInformation<List<T>> {
private static final long serialVersionUID = 1L;
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableListTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableListTypeInfo.java
new file mode 100644
index 00000000000..63448242a35
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableListTypeInfo.java
@@ -0,0 +1,60 @@
+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.SerializerConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer;
+
+import java.util.List;
+
+/**
+ * A {@link TypeInformation} for the list types of the Java API, accepting null collection and null
+ * elements.
+ *
+ * @param <T> The type of the elements in the list.
+ */
+@PublicEvolving
+public class NullableListTypeInfo<T> extends ListTypeInfo<T> {
+
+ public NullableListTypeInfo(Class<T> elementTypeClass) {
+ super(elementTypeClass);
+ }
+
+ public NullableListTypeInfo(TypeInformation<T> elementTypeInfo) {
+ super(elementTypeInfo);
+ }
+
+ @Override
+ public TypeSerializer<List<T>> createSerializer(SerializerConfig config) {
+ // ListSerializer does not support null elements
+ TypeSerializer<T> elementTypeSerializer =
+ NullableSerializer.wrap(getElementTypeInfo().createSerializer(config), false);
+ ListSerializer<T> listSerializer = new ListSerializer<>(elementTypeSerializer);
+ return NullableSerializer.wrap(listSerializer, false);
+ }
+
+ @Override
+ public String toString() {
+ return "NullableList<" + getElementTypeInfo() + '>';
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableMapTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableMapTypeInfo.java
new file mode 100644
index 00000000000..408f34426c7
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableMapTypeInfo.java
@@ -0,0 +1,63 @@
+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.SerializerConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.MapSerializer;
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer;
+
+import java.util.Map;
+
+/**
+ * A {@link TypeInformation} for the map types of the Java API, accepting null collection and null
+ * key/values.
+ *
+ * @param <K> The type of the keys in the map.
+ * @param <V> The type of the values in the map.
+ */
+@PublicEvolving
+public class NullableMapTypeInfo<K, V> extends MapTypeInfo<K, V> {
+
+ public NullableMapTypeInfo(TypeInformation<K> keyTypeInfo, TypeInformation<V> valueTypeInfo) {
+ super(keyTypeInfo, valueTypeInfo);
+ }
+
+ public NullableMapTypeInfo(Class<K> keyClass, Class<V> valueClass) {
+ super(keyClass, valueClass);
+ }
+
+ @Override
+ public TypeSerializer<Map<K, V>> createSerializer(SerializerConfig config) {
+ // MapSerializer does not support null key
+ TypeSerializer<K> keyTypeSerializer =
+ NullableSerializer.wrap(getKeyTypeInfo().createSerializer(config), false);
+ TypeSerializer<V> valueTypeSerializer = getValueTypeInfo().createSerializer(config);
+ MapSerializer<K, V> mapSerializer =
+ new MapSerializer<>(keyTypeSerializer, valueTypeSerializer);
+ return NullableSerializer.wrap(mapSerializer, false);
+ }
+
+ @Override
+ public String toString() {
+ return "NullableMap<" + getKeyTypeInfo() + ", " + getValueTypeInfo() + ">";
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableSetTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableSetTypeInfo.java
new file mode 100644
index 00000000000..296fd1d7712
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/NullableSetTypeInfo.java
@@ -0,0 +1,58 @@
+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.SerializerConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.SetSerializer;
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer;
+
+import java.util.Set;
+
+/**
+ * A {@link TypeInformation} for the set types of the Java API, accepting null collection and null
+ * elements.
+ *
+ * @param <T> The type of the elements in the set.
+ */
+@PublicEvolving
+public class NullableSetTypeInfo<T> extends SetTypeInfo<T> {
+
+ public NullableSetTypeInfo(Class<T> elementTypeClass) {
+ super(elementTypeClass);
+ }
+
+ public NullableSetTypeInfo(TypeInformation<T> elementTypeInfo) {
+ super(elementTypeInfo);
+ }
+
+ @Override
+ public TypeSerializer<Set<T>> createSerializer(SerializerConfig config) {
+ TypeSerializer<T> elementTypeSerializer = getElementTypeInfo().createSerializer(config);
+ SetSerializer<T> setSerializer = new SetSerializer<>(elementTypeSerializer);
+ return NullableSerializer.wrap(setSerializer, false);
+ }
+
+ @Override
+ public String toString() {
+ return "NullableSet<" + getElementTypeInfo() + '>';
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/SetTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/SetTypeInfo.java
index fba52102ed6..3dd2b52d978 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/SetTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/SetTypeInfo.java
@@ -34,7 +34,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* @param <T> The type of the elements in the set.
*/
@PublicEvolving
-public final class SetTypeInfo<T> extends TypeInformation<Set<T>> {
+public class SetTypeInfo<T> extends TypeInformation<Set<T>> {
private static final long serialVersionUID = 1L;
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 c1acc304829..9846ee40e85 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,7 +42,6 @@ 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;
@@ -1989,15 +1988,16 @@ public class TypeExtractor {
Class<?> valueClass = (Class<?>) actualTypeArguments[1];
TypeInformation<?> keyTypeInfo = createTypeInfo(keyClass);
TypeInformation<?> valueTypeInfo = createTypeInfo(valueClass);
- return (TypeInformation<OUT>) Types.MAP(keyTypeInfo, valueTypeInfo);
+ return (TypeInformation<OUT>)
+ new NullableMapTypeInfo<>(keyTypeInfo, valueTypeInfo);
} else if (clazz.isAssignableFrom(List.class)) {
Class<?> elementClass = (Class<?>) actualTypeArguments[0];
TypeInformation<?> elementTypeInfo = createTypeInfo(elementClass);
- return (TypeInformation<OUT>) Types.LIST(elementTypeInfo);
+ return (TypeInformation<OUT>) new NullableListTypeInfo<>(elementTypeInfo);
} else if (clazz.isAssignableFrom(Set.class)) {
Class<?> elementClass = (Class<?>) actualTypeArguments[0];
TypeInformation<?> elementTypeInfo = createTypeInfo(elementClass);
- return (TypeInformation<OUT>) Types.SET(elementTypeInfo);
+ return (TypeInformation<OUT>) new NullableSetTypeInfo<>(elementTypeInfo);
}
}
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableListSerializerTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableListSerializerTest.java
new file mode 100644
index 00000000000..9f3e59e00fd
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableListSerializerTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.typeutils.base;
+
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.NullableListTypeInfo;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+/** A test for the serializer of {@link NullableListTypeInfo}. */
+class NullableListSerializerTest extends SerializerTestBase<List<Long>> {
+
+ @Override
+ protected TypeSerializer<List<Long>> createSerializer() {
+ return new NullableListTypeInfo<>(BasicTypeInfo.LONG_TYPE_INFO)
+ .createSerializer(new SerializerConfigImpl());
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected Class<List<Long>> getTypeClass() {
+ return (Class<List<Long>>) (Class<?>) List.class;
+ }
+
+ @SuppressWarnings({"unchecked"})
+ @Override
+ protected List<Long>[] getTestData() {
+ final Random rnd = new Random(123654789);
+
+ // null list
+ final List<Long> list1 = null;
+
+ // empty lists
+ final List<Long> list2 = Collections.emptyList();
+ final List<Long> list3 = new ArrayList<>();
+
+ // single element lists
+ final List<Long> list4 = Collections.singletonList(55L);
+ final List<Long> list5 = new ArrayList<>();
+ list5.add(777888L);
+
+ // longer list with null value
+ final List<Long> list6 = new ArrayList<>();
+ for (int i = 0; i < rnd.nextInt(200); i++) {
+ list6.add(rnd.nextLong());
+ }
+ list6.add(null);
+
+ return (List<Long>[]) new List[] {list1, list2, list3, list4, list5, list6};
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableMapSerializerTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableMapSerializerTest.java
new file mode 100644
index 00000000000..069e858f31e
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableMapSerializerTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.typeutils.base;
+
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.NullableMapTypeInfo;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+/** A test for the serializer of {@link NullableMapTypeInfo}. */
+class NullableMapSerializerTest extends SerializerTestBase<Map<Long, String>> {
+
+ @Override
+ protected TypeSerializer<Map<Long, String>> createSerializer() {
+ return new NullableMapTypeInfo<>(
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
+ .createSerializer(new SerializerConfigImpl());
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected Class<Map<Long, String>> getTypeClass() {
+ return (Class<Map<Long, String>>) (Class<?>) Map.class;
+ }
+
+ @SuppressWarnings({"unchecked"})
+ @Override
+ protected Map<Long, String>[] getTestData() {
+ final Random rnd = new Random(123654789);
+
+ // null map
+ final Map<Long, String> map1 = null;
+
+ // empty maps
+ final Map<Long, String> map2 = Collections.emptyMap();
+ final Map<Long, String> map3 = new HashMap<>();
+
+ // single element maps
+ final Map<Long, String> map4 = Collections.singletonMap(0L, "hello");
+ final Map<Long, String> map5 = new HashMap<>();
+ map5.put(12345L, "12345L");
+
+ // longer maps with null key and null value
+ final Map<Long, String> map6 = new HashMap<>();
+ for (int i = 0; i < rnd.nextInt(200); i++) {
+ map6.put(rnd.nextLong(), Long.toString(rnd.nextLong()));
+ }
+ map6.put(rnd.nextLong(), null);
+ map6.put(null, Long.toString(rnd.nextLong()));
+
+ return (Map<Long, String>[]) new Map[] {map1, map2, map3, map4, map5, map6};
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableSetSerializerTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableSetSerializerTest.java
new file mode 100644
index 00000000000..d74cafd4460
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/NullableSetSerializerTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.typeutils.base;
+
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.NullableSetTypeInfo;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+
+/** A test for the serializer of {@link NullableSetTypeInfo}. */
+class NullableSetSerializerTest extends SerializerTestBase<Set<Long>> {
+
+ @Override
+ protected TypeSerializer<Set<Long>> createSerializer() {
+ return new NullableSetTypeInfo<>(BasicTypeInfo.LONG_TYPE_INFO)
+ .createSerializer(new SerializerConfigImpl());
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected Class<Set<Long>> getTypeClass() {
+ return (Class<Set<Long>>) (Class<?>) Set.class;
+ }
+
+ @SuppressWarnings({"unchecked"})
+ @Override
+ protected Set<Long>[] getTestData() {
+ final Random rnd = new Random(123654789);
+
+ // null set
+ final Set<Long> set1 = null;
+
+ // empty sets
+ final Set<Long> set2 = Collections.emptySet();
+ final Set<Long> set3 = new HashSet<>();
+
+ // single element sets
+ final Set<Long> set4 = Collections.singleton(55L);
+ final Set<Long> set5 = new HashSet<>();
+ set5.add(12345L);
+
+ // longer sets with null value
+ final Set<Long> set6 = new HashSet<>();
+ for (int i = 0; i < rnd.nextInt(200); i++) {
+ set6.add(rnd.nextLong());
+ }
+ set6.add(null);
+
+ return (Set<Long>[]) new Set[] {set1, set2, set3, set4, set5, set6};
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableListTypeInfoTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableListTypeInfoTest.java
new file mode 100644
index 00000000000..7c3172af2a8
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableListTypeInfoTest.java
@@ -0,0 +1,35 @@
+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.TypeInformationTestBase;
+
+/** Test for {@link NullableListTypeInfoTest}. */
+class NullableListTypeInfoTest extends TypeInformationTestBase<NullableListTypeInfo<?>> {
+
+ @Override
+ protected NullableListTypeInfo<?>[] getTestData() {
+ return new NullableListTypeInfo<?>[] {
+ new NullableListTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO),
+ new NullableListTypeInfo<>(BasicTypeInfo.BOOLEAN_TYPE_INFO),
+ new NullableListTypeInfo<>(Object.class)
+ };
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableMapTypeInfoTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableMapTypeInfoTest.java
new file mode 100644
index 00000000000..ae6b806e446
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableMapTypeInfoTest.java
@@ -0,0 +1,35 @@
+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.TypeInformationTestBase;
+
+/** Test for {@link NullableMapTypeInfo}. */
+class NullableMapTypeInfoTest extends TypeInformationTestBase<NullableMapTypeInfo<?, ?>> {
+
+ @Override
+ protected NullableMapTypeInfo<?, ?>[] getTestData() {
+ return new NullableMapTypeInfo<?, ?>[] {
+ new NullableMapTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO),
+ new NullableMapTypeInfo<>(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO),
+ new NullableMapTypeInfo<>(String.class, Boolean.class)
+ };
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableSetTypeInfoTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableSetTypeInfoTest.java
new file mode 100644
index 00000000000..f0ade0253a1
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/NullableSetTypeInfoTest.java
@@ -0,0 +1,35 @@
+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.TypeInformationTestBase;
+
+/** Test for {@link NullableSetTypeInfo}. */
+class NullableSetTypeInfoTest extends TypeInformationTestBase<NullableSetTypeInfo<?>> {
+
+ @Override
+ protected NullableSetTypeInfo<?>[] getTestData() {
+ return new NullableSetTypeInfo<?>[] {
+ new NullableSetTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO),
+ new NullableSetTypeInfo<>(BasicTypeInfo.BOOLEAN_TYPE_INFO),
+ new NullableSetTypeInfo<>(Object.class),
+ };
+ }
+}
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 eef6ba5d868..28977a32488 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,8 @@ public class PojoTypeExtractionTest {
fail("already seen");
}
collectionSeen = true;
- assertThat(field.getTypeInformation()).isEqualTo(new ListTypeInfo<>(String.class));
+ assertThat(field.getTypeInformation())
+ .isEqualTo(new NullableListTypeInfo<>(String.class));
} else {
fail("Unexpected field " + field);
(flink) 03/04: [FLINK-35068][core] Introduce built-in serialization support for java.util.Set
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 f46bb434404f93772b38d55b94f599d92eafe82a
Author: Zhanghao Chen <m1...@outlook.com>
AuthorDate: Tue Apr 9 12:52:14 2024 +0800
[FLINK-35068][core] Introduce built-in serialization support for java.util.Set
---
.../apache/flink/api/common/typeinfo/Types.java | 19 +++
.../api/common/typeutils/base/SetSerializer.java | 181 +++++++++++++++++++++
.../typeutils/base/SetSerializerSnapshot.java | 59 +++++++
.../flink/api/java/typeutils/SetTypeInfo.java | 131 +++++++++++++++
.../flink/api/java/typeutils/TypeExtractor.java | 5 +
.../common/typeutils/base/SetSerializerTest.java | 86 ++++++++++
.../flink/api/java/typeutils/SetTypeInfoTest.java | 35 ++++
.../api/java/typeutils/TypeExtractorTest.java | 5 +
.../TypeSerializerTestCoverageTest.java | 4 +-
9 files changed, 524 insertions(+), 1 deletion(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java
index a66e3be2b13..511635215ca 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java
@@ -30,6 +30,7 @@ import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
import org.apache.flink.api.java.typeutils.PojoField;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.SetTypeInfo;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.ValueTypeInfo;
@@ -49,8 +50,10 @@ import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* This class gives access to the type information of the most common types for which Flink has
@@ -443,6 +446,22 @@ public class Types {
return new ListTypeInfo<>(elementType);
}
+ /**
+ * Returns type information for a Java {@link java.util.Set}. A set must not be null. Null
+ * values in elements are not supported.
+ *
+ * <p>By default, sets are untyped and treated as a generic type in Flink; therefore, it is
+ * useful to pass type information whenever a set is used.
+ *
+ * <p><strong>Note:</strong> Flink does not preserve the concrete {@link Set} type. It converts
+ * a list into {@link HashSet} when copying or deserializing.
+ *
+ * @param elementType type information for the set's elements
+ */
+ public static <E> TypeInformation<Set<E>> SET(TypeInformation<E> elementType) {
+ return new SetTypeInfo<>(elementType);
+ }
+
/**
* Returns type information for Java enumerations. Null values are not supported.
*
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SetSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SetSerializer.java
new file mode 100644
index 00000000000..0f86346e267
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SetSerializer.java
@@ -0,0 +1,181 @@
+/*
+ * 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.typeutils.base;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A serializer for {@link java.util.Set}. The serializer relies on an element serializer for the
+ * serialization of the set's elements.
+ *
+ * <p>The serialization format for the set is as follows: four bytes for the length of the set,
+ * followed by the serialized representation of each element. To allow null values, each value is
+ * prefixed by a null marker.
+ *
+ * @param <T> The type of element in the set.
+ */
+@Internal
+public final class SetSerializer<T> extends TypeSerializer<Set<T>> {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The serializer for the elements of the set. */
+ private final TypeSerializer<T> elementSerializer;
+
+ /**
+ * Creates a set serializer that uses the given serializer to serialize the set's elements.
+ *
+ * @param elementSerializer The serializer for the elements of the set
+ */
+ public SetSerializer(TypeSerializer<T> elementSerializer) {
+ this.elementSerializer = checkNotNull(elementSerializer);
+ }
+
+ // ------------------------------------------------------------------------
+ // SetSerializer specific properties
+ // ------------------------------------------------------------------------
+
+ /**
+ * Gets the serializer for the elements of the set.
+ *
+ * @return The serializer for the elements of the set
+ */
+ public TypeSerializer<T> getElementSerializer() {
+ return elementSerializer;
+ }
+
+ // ------------------------------------------------------------------------
+ // Type Serializer implementation
+ // ------------------------------------------------------------------------
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<Set<T>> duplicate() {
+ TypeSerializer<T> duplicateElement = elementSerializer.duplicate();
+ return duplicateElement == elementSerializer ? this : new SetSerializer<>(duplicateElement);
+ }
+
+ @Override
+ public Set<T> createInstance() {
+ return new HashSet<>(0);
+ }
+
+ @Override
+ public Set<T> copy(Set<T> from) {
+ Set<T> newSet = new HashSet<>(from.size());
+ for (T element : from) {
+ T newElement = element == null ? null : elementSerializer.copy(element);
+ newSet.add(newElement);
+ }
+ return newSet;
+ }
+
+ @Override
+ public Set<T> copy(Set<T> from, Set<T> reuse) {
+ return copy(from);
+ }
+
+ @Override
+ public int getLength() {
+ return -1; // var length
+ }
+
+ @Override
+ public void serialize(Set<T> set, DataOutputView target) throws IOException {
+ final int size = set.size();
+ target.writeInt(size);
+ for (T element : set) {
+ if (element == null) {
+ target.writeBoolean(true);
+ } else {
+ target.writeBoolean(false);
+ elementSerializer.serialize(element, target);
+ }
+ }
+ }
+
+ @Override
+ public Set<T> deserialize(DataInputView source) throws IOException {
+ final int size = source.readInt();
+ final Set<T> set = new HashSet<>(size);
+ for (int i = 0; i < size; i++) {
+ boolean isNull = source.readBoolean();
+ T element = isNull ? null : elementSerializer.deserialize(source);
+ set.add(element);
+ }
+ return set;
+ }
+
+ @Override
+ public Set<T> deserialize(Set<T> reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ // copy number of elements
+ final int num = source.readInt();
+ target.writeInt(num);
+ for (int i = 0; i < num; i++) {
+ boolean isNull = source.readBoolean();
+ target.writeBoolean(isNull);
+ if (!isNull) {
+ elementSerializer.copy(source, target);
+ }
+ }
+ }
+
+ // --------------------------------------------------------------------
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj == this
+ || (obj != null
+ && obj.getClass() == getClass()
+ && elementSerializer.equals(((SetSerializer<?>) obj).elementSerializer));
+ }
+
+ @Override
+ public int hashCode() {
+ return elementSerializer.hashCode();
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Serializer configuration snapshot & compatibility
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public TypeSerializerSnapshot<Set<T>> snapshotConfiguration() {
+ return new SetSerializerSnapshot<>(this);
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SetSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SetSerializerSnapshot.java
new file mode 100644
index 00000000000..cc4ca52052b
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SetSerializerSnapshot.java
@@ -0,0 +1,59 @@
+/*
+ * 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.typeutils.base;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+import java.util.Set;
+
+/** Snapshot class for the {@link SetSerializer}. */
+@Internal
+public class SetSerializerSnapshot<T>
+ extends CompositeTypeSerializerSnapshot<Set<T>, SetSerializer<T>> {
+
+ private static final int CURRENT_VERSION = 1;
+
+ /** Constructor for read instantiation. */
+ public SetSerializerSnapshot() {}
+
+ /** Constructor to create the snapshot for writing. */
+ public SetSerializerSnapshot(SetSerializer<T> setSerializer) {
+ super(setSerializer);
+ }
+
+ @Override
+ public int getCurrentOuterSnapshotVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ protected SetSerializer<T> createOuterSerializerWithNestedSerializers(
+ TypeSerializer<?>[] nestedSerializers) {
+ @SuppressWarnings("unchecked")
+ TypeSerializer<T> elementSerializer = (TypeSerializer<T>) nestedSerializers[0];
+ return new SetSerializer<>(elementSerializer);
+ }
+
+ @Override
+ protected TypeSerializer<?>[] getNestedSerializers(SetSerializer<T> outerSerializer) {
+ return new TypeSerializer<?>[] {outerSerializer.getElementSerializer()};
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/SetTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/SetTypeInfo.java
new file mode 100644
index 00000000000..fba52102ed6
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/SetTypeInfo.java
@@ -0,0 +1,131 @@
+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.SerializerConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.SetSerializer;
+
+import java.util.Set;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A {@link TypeInformation} for the set types of the Java API.
+ *
+ * @param <T> The type of the elements in the set.
+ */
+@PublicEvolving
+public final class SetTypeInfo<T> extends TypeInformation<Set<T>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final TypeInformation<T> elementTypeInfo;
+
+ public SetTypeInfo(Class<T> elementTypeClass) {
+ this.elementTypeInfo =
+ of(checkNotNull(elementTypeClass, "The element type class cannot be null."));
+ }
+
+ public SetTypeInfo(TypeInformation<T> elementTypeInfo) {
+ this.elementTypeInfo =
+ checkNotNull(elementTypeInfo, "The element type information cannot be null.");
+ }
+
+ // ------------------------------------------------------------------------
+ // SetTypeInfo specific properties
+ // ------------------------------------------------------------------------
+
+ /** Gets the type information for the elements contained in the set. */
+ public TypeInformation<T> getElementTypeInfo() {
+ return elementTypeInfo;
+ }
+
+ // ------------------------------------------------------------------------
+ // TypeInformation implementation
+ // ------------------------------------------------------------------------
+
+ @Override
+ public boolean isBasicType() {
+ return false;
+ }
+
+ @Override
+ public boolean isTupleType() {
+ return false;
+ }
+
+ @Override
+ public int getArity() {
+ return 0;
+ }
+
+ @Override
+ public int getTotalFields() {
+ return 1;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Class<Set<T>> getTypeClass() {
+ return (Class<Set<T>>) (Class<?>) Set.class;
+ }
+
+ @Override
+ public boolean isKeyType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<Set<T>> createSerializer(SerializerConfig config) {
+ TypeSerializer<T> elementTypeSerializer = elementTypeInfo.createSerializer(config);
+ return new SetSerializer<>(elementTypeSerializer);
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public String toString() {
+ return "Set<" + elementTypeInfo + '>';
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ } else if (obj instanceof SetTypeInfo) {
+ final SetTypeInfo<?> other = (SetTypeInfo<?>) obj;
+ return other.canEqual(this) && elementTypeInfo.equals(other.elementTypeInfo);
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public int hashCode() {
+ return 31 * elementTypeInfo.hashCode() + 1;
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj != null && obj.getClass() == getClass();
+ }
+}
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 b761186dd52..c1acc304829 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
@@ -73,6 +73,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.checkAndExtractLambda;
import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.getAllDeclaredMethods;
@@ -1993,6 +1994,10 @@ public class TypeExtractor {
Class<?> elementClass = (Class<?>) actualTypeArguments[0];
TypeInformation<?> elementTypeInfo = createTypeInfo(elementClass);
return (TypeInformation<OUT>) Types.LIST(elementTypeInfo);
+ } else if (clazz.isAssignableFrom(Set.class)) {
+ Class<?> elementClass = (Class<?>) actualTypeArguments[0];
+ TypeInformation<?> elementTypeInfo = createTypeInfo(elementClass);
+ return (TypeInformation<OUT>) Types.SET(elementTypeInfo);
}
}
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/SetSerializerTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/SetSerializerTest.java
new file mode 100644
index 00000000000..3ffa481956b
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/SetSerializerTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.typeutils.base;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+import java.util.TreeSet;
+
+/** A test for the {@link SetSerializer}. */
+class SetSerializerTest extends SerializerTestBase<Set<Long>> {
+
+ @Override
+ protected TypeSerializer<Set<Long>> createSerializer() {
+ return new SetSerializer<>(LongSerializer.INSTANCE);
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected Class<Set<Long>> getTypeClass() {
+ return (Class<Set<Long>>) (Class<?>) Set.class;
+ }
+
+ @SuppressWarnings({"unchecked"})
+ @Override
+ protected Set<Long>[] getTestData() {
+ final Random rnd = new Random(123654789);
+
+ // empty sets
+ final Set<Long> set1 = Collections.emptySet();
+ final Set<Long> set2 = new HashSet<>();
+ final Set<Long> set3 = new TreeSet<>();
+
+ // single element sets
+ final Set<Long> set4 = Collections.singleton(55L);
+ final Set<Long> set5 = new HashSet<>();
+ set5.add(12345L);
+ final Set<Long> set6 = new TreeSet<>();
+ set6.add(777888L);
+
+ // longer sets
+ final Set<Long> set7 = new HashSet<>();
+ for (int i = 0; i < rnd.nextInt(200); i++) {
+ set7.add(rnd.nextLong());
+ }
+
+ int set8Len = rnd.nextInt(200);
+ final Set<Long> set8 = new HashSet<>(set8Len);
+ for (int i = 0; i < set8Len; i++) {
+ set8.add(rnd.nextLong());
+ }
+
+ // null-value sets
+ final Set<Long> set9 = Collections.singleton(null);
+ final Set<Long> set10 = new HashSet<>();
+ set10.add(null);
+
+ return (Set<Long>[])
+ new Set[] {set1, set2, set3, set4, set5, set6, set7, set8, set9, set10};
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/SetTypeInfoTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/SetTypeInfoTest.java
new file mode 100644
index 00000000000..f301fe81d2a
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/SetTypeInfoTest.java
@@ -0,0 +1,35 @@
+/*
+ * 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.java.typeutils;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.TypeInformationTestBase;
+
+/** Test for {@link SetTypeInfo}. */
+class SetTypeInfoTest extends TypeInformationTestBase<SetTypeInfo<?>> {
+
+ @Override
+ protected SetTypeInfo<?>[] getTestData() {
+ return new SetTypeInfo<?>[] {
+ new SetTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO),
+ new SetTypeInfo<>(BasicTypeInfo.BOOLEAN_TYPE_INFO),
+ new SetTypeInfo<>(Object.class),
+ };
+ }
+}
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 72a1df065f7..012183f270c 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
@@ -65,10 +65,12 @@ import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -2513,6 +2515,7 @@ public class TypeExtractorTest {
public Map<String, Integer> mapVal = new HashMap<>();
public List<String> listVal = new ArrayList<>();
public Collection<String> collectionVal = new ArrayList<>();
+ public Set<String> setVal = new HashSet<>();
// Collection fields with unsupported collection types, treated as generic types
public LinkedList<String> linkedListVal = new LinkedList<>();
@@ -2567,6 +2570,8 @@ public class TypeExtractorTest {
pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("collectionVal"))
.getTypeInformation())
.isInstanceOf(ListTypeInfo.class);
+ assertThat(pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("setVal")).getTypeInformation())
+ .isInstanceOf(SetTypeInfo.class);
assertThat(
pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("linkedListVal"))
.getTypeInformation())
diff --git a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
index a5d288b37d9..59f0d256f13 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
@@ -30,6 +30,7 @@ import org.apache.flink.api.common.typeutils.base.LocalDateSerializer;
import org.apache.flink.api.common.typeutils.base.LocalDateTimeSerializer;
import org.apache.flink.api.common.typeutils.base.LocalTimeSerializer;
import org.apache.flink.api.common.typeutils.base.NullValueSerializer;
+import org.apache.flink.api.common.typeutils.base.SetSerializer;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.api.common.typeutils.base.array.BooleanPrimitiveArraySerializer;
import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
@@ -243,7 +244,8 @@ public class TypeSerializerTestCoverageTest extends TestLogger {
AvroSerializer.class.getName(),
// KeyAndValueSerializer shouldn't be used to serialize data to state and
// doesn't need to ensure upgrade compatibility.
- "org.apache.flink.streaming.api.operators.sortpartition.KeyAndValueSerializer");
+ "org.apache.flink.streaming.api.operators.sortpartition.KeyAndValueSerializer",
+ SetSerializer.class.getName());
// check if a test exists for each type serializer
for (Class<? extends TypeSerializer> typeSerializer : typeSerializers) {
(flink) 01/04: [FLINK-34123][core] 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 24b661f782e494b2287b4f8ed48fa5a7197eada7
Author: Zhanghao Chen <m1...@outlook.com>
AuthorDate: Mon Apr 8 21:21:21 2024 +0800
[FLINK-34123][core] 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 93902c4d84c..b761186dd52 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;
@@ -1968,6 +1970,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 (AvroUtils.isAvroSpecificRecord(clazz)) {
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 bc5506bb836..eef6ba5d868 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 84c2a9255f0..72a1df065f7 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
@@ -63,7 +63,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;
@@ -2316,15 +2319,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;
}
@@ -2332,7 +2335,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);
}
@@ -2344,7 +2347,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);
})
@@ -2500,4 +2505,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 506124816b1..2cc1550741f 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
@@ -76,7 +76,7 @@ class PojoTestUtilsTest {
}
public static class PojoRequiringKryo {
- public List<Integer> x;
+ public List<?> x;
}
@TypeInfo(FooFactory.class)