FLIP-539: Support WITH Clause in CREATE FUNCTION Statement in Flink SQL

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

The existing CREATE FUNCTION  statement in Flink SQL allows users to register user-defined functions (UDFs) by specifying the class name and the artifact (JAR) containing the implementation. While this design covers common use cases, it lacks a declarative mechanism for associating arbitrary properties or metadata with the function at creation time.

Other Flink SQL objects—such as tables—support a WITH  clause for specifying options in a key-value fashion, improving consistency, discoverability, and extensibility.

For example:

CREATE TABLE t_example ( id INT ) WITH ( 'connector' = 'kafka', 'topic' = 'demo' );

However, there is currently no option to embed properties or metadata in the CREATE FUNCTION statement, which means function features that depend on out-of-band configuration (such as resource hints, versioning, implementation characteristics, or custom runtime behaviors) either require system-level configuration or cannot be easily set per function.

This proposal aims to enable a WITH  clause for CREATE FUNCTION—bringing feature parity with other DDLs and empowering functions to be richer, more flexible, and more configurable.

Use Cases

• Custom Metadata: Users can store descriptive, operational, or semantic metadata directly with the function DDL.

• Resource Hints: Indicating resource requirements, such as optimal parallelism or memory.

• Versioning: Track function implementation details.

• Vendor/Platform-Specific Options: Allow cloud providers or Flink connectors to use custom keys (as is standard in WITH clauses elsewhere).

• Compatibility with Table DDLs: Align the function DDL syntax with familiar patterns, reducing the learning curve and risk of user error.

Public Interfaces

New Supported SQL Syntax

Add support for:

CREATE FUNCTION <function_name> AS '<class_name>' [USING JAR '<artifact_uri>'] WITH ( 'property_key1' = 'property_value1', 'property_key2' = 'property_value2', ... );

• The WITH clause will accept a list of key-value pairs similar to the syntax currently used in CREATE TABLE, CREATE VIEW, and CTAS.

• The clause is optional and backward compatible.

Example usage:

CREATE FUNCTION my_func AS 'com.example.MyFunction' 
USING JAR 'file:///xxx-udf-1.0.1-20180502.011548-12.jar' 
WITH ( 'description' = 'A demo UDF', 'version' = '1.0.0', 'custom.runtime.memory' = '2g' );

Pass options to existing interfaces

CatalogFunction:

/** Interface for a function in a catalog. */
@PublicEvolving
public interface CatalogFunction {
    /**
     * Returns a map of string-based options.
     */
    Map<String, String> getOptions();
}

TableEnvironment:

@PublicEvolving
public interface TableEnvironment {
 void createFunction(String tablePath, FunctionDescriptor functionDescriptor);
 void createFunction(String tablePath, FunctionDescriptor functionDescriptor, boolean ignoreIfExists);
 void createTemporaryFunction(String tablePath, FunctionDescriptor functionDescriptor);
 void createTemporaryFunction(String tablePath, FunctionDescriptor functionDescriptor, boolean ignoreIfExists);
 void createTemporarySystemFunction(String tablePath, FunctionDescriptor functionDescriptor);
 void createTemporarySystemFunction(String tablePath, FunctionDescriptor functionDescriptor, boolean ignoreIfExists);  }

(New class) FunctionDescriptor:

@PublicEvolving
public final class FunctionDescriptor {
    private final String className;
    private final FunctionLanguage language;
    private final List<ResourceUri> resourceUris;
    private final Map<String, String> options;

    private FunctionDescriptor(
        String className,
        FunctionLanguage language,
        List<ResourceUri> resourceUris,
        Map<String, String> options
    ) {
        this.className = className;
        this.language = language;
        this.resourceUris = resourceUris;
        this.options = options;
    }

    public static Builder forClassName(String className);

    public static Builder forFunctionClass(Class<? extends UserDefinedFunction> functionClass);

    public static final class Builder {
        private Builder(String className) {}
        public Builder language(FunctionLanguage language) ;
        public Builder jarUri(String jarUri);
        public Builder resourceUri(ResourceUri uri);
        public Builder option(String key, String value);
        public Builder options(Map<String, String> options);
    }
}

Future work

The options would not be stored in the compiled plan. Therefore the options would not be available when restoring a function definition from a compile plan using simply a class of the function definitions. If a need for accessing these options during runtime arises we can

1. Serialize the options into a compiled plan
2. Add a configure(Map<String, String>  method to FunctionDefinition  which would be called after restoring from a compiled plan with provided options.

In the meantime we will prohibit serializing CatalogFunctions into a compiled plan if they have non empty options.

Compatibility, Deprecation, and Migration Plan

• The change is fully backward compatible as the clause is optional; existing statements remain valid.