Dinky是一个基于 Apache Flink 二次开发且易扩展的一站式开发运维 FlinkSQL 及 SQL 的实时计算平台

/* * 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.table.types.extraction; import com.dlink.pool.ClassPool; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.shaded.asm7.org.objectweb.asm.*; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.StructuredType; import javax.annotation.Nullable; import java.io.IOException; import java.io.InputStream; import java.lang.annotation.Annotation; import java.lang.reflect.Type; import java.lang.reflect.*; import java.util.*; import java.util.function.Function; import java.util.regex.Pattern; import java.util.stream.Collectors; import java.util.stream.Stream; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getConstructorDescriptor; import static org.apache.flink.shaded.asm7.org.objectweb.asm.Type.getMethodDescriptor; /** * Utilities for performing reflection tasks. */ @Internal public final class ExtractionUtils { // -------------------------------------------------------------------------------------------- // Methods shared across packages // -------------------------------------------------------------------------------------------- /** * Collects methods of the given name. */ public static List<Method> collectMethods(Class<?> function, String methodName) { return Arrays.stream(function.getMethods()) .filter(method -> method.getName().equals(methodName)) .sorted(Comparator.comparing(Method::toString)) // for deterministic order .collect(Collectors.toList()); } /** * Checks whether a method/constructor can be called with the given argument classes. This * includes type widening and vararg. {@code null} is a wildcard. * * <p>E.g., {@code (int.class, int.class)} matches {@code f(Object...), f(int, int), f(Integer, * Object)} and so forth. */ public static boolean isInvokable(Executable executable, Class<?>... classes) { final int m = executable.getModifiers(); if (!Modifier.isPublic(m)) { return false; } final int paramCount = executable.getParameterCount(); final int classCount = classes.length; // check for enough classes for each parameter if ((!executable.isVarArgs() && classCount != paramCount) || (executable.isVarArgs() && classCount < paramCount - 1)) { return false; } int currentClass = 0; for (int currentParam = 0; currentParam < paramCount; currentParam++) { final Class<?> param = executable.getParameterTypes()[currentParam]; // last parameter is a vararg that needs to consume remaining classes if (currentParam == paramCount - 1 && executable.isVarArgs()) { final Class<?> paramComponent = executable.getParameterTypes()[currentParam].getComponentType(); // we have more than 1 classes left so the vararg needs to consume them all if (classCount - currentClass > 1) { while (currentClass < classCount && ExtractionUtils.isAssignable( classes[currentClass], paramComponent, true)) { currentClass++; } } else if (currentClass < classCount && (parameterMatches(classes[currentClass], param) || parameterMatches(classes[currentClass], paramComponent))) { currentClass++; } } // entire parameter matches else if (parameterMatches(classes[currentClass], param)) { currentClass++; } } // check if all classes have been consumed return currentClass == classCount; } private static boolean parameterMatches(Class<?> clz, Class<?> param) { return clz == null || ExtractionUtils.isAssignable(clz, param, true); } /** * Creates a method signature string like {@code int eval(Integer, String)}. */ public static String createMethodSignatureString( String methodName, Class<?>[] parameters, @Nullable Class<?> returnType) { final StringBuilder builder = new StringBuilder(); if (returnType != null) { builder.append(returnType.getCanonicalName()).append(" "); } builder.append(methodName) .append( Stream.of(parameters) .map( parameter -> { // in case we don't know the parameter at this location // (i.e. for accumulators) if (parameter == null) { return "_"; } else { return parameter.getCanonicalName(); } }) .collect(Collectors.joining(", ", "(", ")"))); return builder.toString(); } /** * Validates the characteristics of a class for a {@link StructuredType} such as accessibility. */ public static void validateStructuredClass(Class<?> clazz) { final int m = clazz.getModifiers(); if (Modifier.isAbstract(m)) { throw extractionError("Class '%s' must not be abstract.", clazz.getName()); } if (!Modifier.isPublic(m)) { throw extractionError("Class '%s' is not public.", clazz.getName()); } if (clazz.getEnclosingClass() != null && (clazz.getDeclaringClass() == null || !Modifier.isStatic(m))) { throw extractionError( "Class '%s' is a not a static, globally accessible class.", clazz.getName()); } } /** * Returns the field of a structured type. The logic is as broad as possible to support both * Java and Scala in different flavors. */ public static Field getStructuredField(Class<?> clazz, String fieldName) { final String normalizedFieldName = fieldName.toUpperCase(); final List<Field> fields = collectStructuredFields(clazz); for (Field field : fields) { if (field.getName().toUpperCase().equals(normalizedFieldName)) { return field; } } throw extractionError( "Could not find a field named '%s' in class '%s' for structured type.", fieldName, clazz.getName()); } /** * Checks for a field getter of a structured type. The logic is as broad as possible to support * both Java and Scala in different flavors. */ public static Optional<Method> getS