twalthr commented on a change in pull request #10942: [FLINK-15487][table]
Allow registering FLIP-65 functions in TableEnvironment
URL: https://github.com/apache/flink/pull/10942#discussion_r372289711
##########
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
##########
@@ -147,39 +157,156 @@
}
/**
- * Prepares a {@link UserDefinedFunction} for usage in the API.
+ * Instantiates a {@link UserDefinedFunction} assuming a default
constructor.
*/
- public static void prepareFunction(TableConfig config,
UserDefinedFunction function) {
- if (function instanceof TableFunction) {
-
UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
+ public static UserDefinedFunction instantiateFunction(Class<? extends
UserDefinedFunction> functionClass) {
+ validateClass(functionClass, true);
+ try {
+ return functionClass.newInstance();
+ } catch (Exception e) {
+ throw new ValidationException(
+ String.format("Cannot instantiate user-defined
function class '%s'.", functionClass.getName()),
+ e);
}
-
UserDefinedFunctionHelper.validateInstantiation(function.getClass());
- UserDefinedFunctionHelper.cleanFunction(config, function);
}
/**
- * Checks if a user-defined function can be easily instantiated.
+ * Prepares a {@link UserDefinedFunction} instance for usage in the API.
*/
- private static void validateInstantiation(Class<?> clazz) {
- if (!InstantiationUtil.isPublic(clazz)) {
- throw new ValidationException(String.format("Function
class %s is not public.", clazz.getCanonicalName()));
- } else if (!InstantiationUtil.isProperClass(clazz)) {
- throw new ValidationException(String.format(
- "Function class %s is no proper class," +
- " it is either abstract, an interface,
or a primitive type.", clazz.getCanonicalName()));
+ public static void prepareInstance(TableConfig config,
UserDefinedFunction function) {
+ validateClass(function.getClass(), false);
+ cleanFunction(config, function);
+ }
+
+ /**
+ * Validates a {@link UserDefinedFunction} class for usage in the API.
+ *
+ * <p>Note: This is an initial validation to indicate common errors
early. The concrete signature
+ * validation happens in the code generation when the actual {@link
DataType}s for arguments and result
+ * are known.
+ */
+ public static void validateClass(Class<? extends UserDefinedFunction>
functionClass) {
+ validateClass(functionClass, true);
+ }
+
+ /**
+ * Validates a {@link UserDefinedFunction} class for usage in the API.
+ */
+ private static void validateClass(
+ Class<? extends UserDefinedFunction> functionClass,
+ boolean requiresDefaultConstructor) {
+ if (TableFunction.class.isAssignableFrom(functionClass)) {
+ validateNotSingleton(functionClass);
}
+ validateInstantiation(functionClass,
requiresDefaultConstructor);
+ validateImplementationMethods(functionClass);
}
/**
* Check whether this is a Scala object. Using Scala objects can lead
to concurrency issues,
* e.g., due to a shared collector.
*/
private static void validateNotSingleton(Class<?> clazz) {
- // TODO it is not a good way to check singleton. Maybe improve
it further.
if (Arrays.stream(clazz.getFields()).anyMatch(f ->
f.getName().equals("MODULE$"))) {
throw new ValidationException(String.format(
"Function implemented by class %s is a Scala
object. This is forbidden because of concurrency" +
- " problems when using them.",
clazz.getCanonicalName()));
+ " problems when using them.",
clazz.getName()));
+ }
+ }
+
+ /**
+ * Validates the implementation methods such as {@code eval()} or
{@code accumulate()} depending
+ * on the {@link UserDefinedFunction} subclass.
+ *
+ * <p>This method must be kept in sync with the code generation
requirements and the individual
+ * docs of each function.
+ */
+ private static void validateImplementationMethods(Class<? extends
UserDefinedFunction> functionClass) {
+ if (ScalarFunction.class.isAssignableFrom(functionClass)) {
+ validateImplementationMethod(functionClass, false,
false, "eval");
Review comment:
I think this helper class is a good location. Will update the PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
