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 f91730dccb65438cba4da88a6607bd0347b1561c Author: Jeyhun Karimov <je.kari...@gmail.com> AuthorDate: Thu May 30 13:58:13 2024 +0200 [FLINK-34977][API] Introduce TypeDescriptor and TypeUtils --- .../flink/api/common/typeinfo/TypeDescriptor.java | 39 +++++++ .../flink/api/common/typeinfo/utils/TypeUtils.java | 125 +++++++++++++++++++++ .../datastream/impl/common/TypeUtilsTest.java | 70 ++++++++++++ 3 files changed, 234 insertions(+) diff --git a/flink-core-api/src/main/java/org/apache/flink/api/common/typeinfo/TypeDescriptor.java b/flink-core-api/src/main/java/org/apache/flink/api/common/typeinfo/TypeDescriptor.java new file mode 100644 index 00000000000..9a616ea61b3 --- /dev/null +++ b/flink-core-api/src/main/java/org/apache/flink/api/common/typeinfo/TypeDescriptor.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.api.common.typeinfo; + +import org.apache.flink.annotation.Experimental; + +import java.io.Serializable; + +/** + * Descriptor interface to create TypeInformation instances. + * + * @param <T> The type represented by this type descriptor. + */ +@Experimental +public interface TypeDescriptor<T> extends Serializable { + + /** + * Gets the class of the type represented by this type descriptor. + * + * @return The class of the type represented by this type descriptor. + */ + Class<T> getTypeClass(); +} diff --git a/flink-core-api/src/main/java/org/apache/flink/api/common/typeinfo/utils/TypeUtils.java b/flink-core-api/src/main/java/org/apache/flink/api/common/typeinfo/utils/TypeUtils.java new file mode 100644 index 00000000000..04eea60a475 --- /dev/null +++ b/flink-core-api/src/main/java/org/apache/flink/api/common/typeinfo/utils/TypeUtils.java @@ -0,0 +1,125 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.api.common.typeinfo.utils; + +import org.apache.flink.annotation.Experimental; + +import java.io.Serializable; +import java.lang.reflect.Constructor; + +/** Utility class to create objects via reflection. */ +@Experimental +public class TypeUtils implements Serializable { + + public static Object getInstance(String classPath, Object... args) + throws ReflectiveOperationException { + Class<?> descriptorClass = Class.forName(classPath); + Constructor<?>[] constructors = descriptorClass.getConstructors(); + + for (Constructor<?> constructor : constructors) { + Class<?>[] parameterTypes = constructor.getParameterTypes(); + + if (areParameterTypesCompatible(parameterTypes, args)) { + return constructor.newInstance(convertArgs(parameterTypes, args)); + } + } + + throw new NoSuchMethodException("No suitable constructor found for the given arguments."); + } + + private static boolean areParameterTypesCompatible(Class<?>[] parameterTypes, Object[] args) { + if (parameterTypes.length != args.length) { + return false; + } + + for (int i = 0; i < parameterTypes.length; i++) { + if (args[i] == null) { + if (parameterTypes[i].isPrimitive()) { + return false; // primitive types cannot be null + } + } else if (!isCompatible(parameterTypes[i], args[i])) { + return false; + } + } + return true; + } + + private static boolean isCompatible(Class<?> parameterType, Object arg) { + if (parameterType.isPrimitive()) { + Class<?> wrapperType = getWrapperClass(parameterType); + return wrapperType.isInstance(arg); + } else { + return parameterType.isInstance(arg); + } + } + + private static Class<?> getWrapperClass(Class<?> primitiveType) { + if (primitiveType == boolean.class) { + return Boolean.class; + } else if (primitiveType == byte.class) { + return Byte.class; + } else if (primitiveType == char.class) { + return Character.class; + } else if (primitiveType == short.class) { + return Short.class; + } else if (primitiveType == int.class) { + return Integer.class; + } else if (primitiveType == long.class) { + return Long.class; + } else if (primitiveType == float.class) { + return Float.class; + } else if (primitiveType == double.class) { + return Double.class; + } + throw new IllegalArgumentException("Unknown primitive type: " + primitiveType.getName()); + } + + private static Object[] convertArgs(Class<?>[] parameterTypes, Object[] args) { + Object[] convertedArgs = new Object[args.length]; + for (int i = 0; i < args.length; i++) { + if (parameterTypes[i].isPrimitive()) { + convertedArgs[i] = convertToPrimitiveWrapper(parameterTypes[i], args[i]); + } else { + convertedArgs[i] = args[i]; + } + } + return convertedArgs; + } + + private static Object convertToPrimitiveWrapper(Class<?> parameterType, Object arg) { + if (parameterType == boolean.class) { + return (Boolean) arg; + } else if (parameterType == byte.class) { + return (Byte) arg; + } else if (parameterType == char.class) { + return (Character) arg; + } else if (parameterType == short.class) { + return (Short) arg; + } else if (parameterType == int.class) { + return (Integer) arg; + } else if (parameterType == long.class) { + return (Long) arg; + } else if (parameterType == float.class) { + return (Float) arg; + } else if (parameterType == double.class) { + return (Double) arg; + } + throw new IllegalArgumentException("Unknown primitive type: " + parameterType.getName()); + } +} diff --git a/flink-datastream/src/test/java/org/apache/flink/datastream/impl/common/TypeUtilsTest.java b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/common/TypeUtilsTest.java new file mode 100644 index 00000000000..9a5b2150eb1 --- /dev/null +++ b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/common/TypeUtilsTest.java @@ -0,0 +1,70 @@ +/* + * 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.datastream.impl.common; + +import org.apache.flink.api.common.typeinfo.utils.TypeUtils; +import org.apache.flink.api.common.typeutils.base.BooleanSerializer; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link org.apache.flink.api.common.typeinfo.utils.TypeUtils}. */ +class TypeUtilsTest { + + @Test + void testNewInstanceWithPrimitiveArg() throws ReflectiveOperationException { + assertThat( + TypeUtils.getInstance( + "org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus", + 0) + .toString()) + .isEqualTo("WatermarkStatus(ACTIVE)"); + } + + @Test + void testNewInstanceWithObjectArg() throws ReflectiveOperationException { + assertThat( + TypeUtils.getInstance( + "org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer", + new BooleanSerializer()) + .toString()) + .contains("StreamElementSerializer"); + } + + @Test + void testNewInstanceWithIncorrectConstructorArgType() { + assertThatThrownBy( + () -> + TypeUtils.getInstance( + "org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer", + 123)) + .isInstanceOf(NoSuchMethodException.class); + } + + @Test + void testNewInstanceWithIncorrectConstructorArgCount() { + assertThatThrownBy( + () -> + TypeUtils.getInstance( + "org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer")) + .isInstanceOf(NoSuchMethodException.class); + } +}