<a id="cmf-sql-udf"></a>

# User-Defined Functions in Confluent Manager for Apache Flink

Starting with version 2.4.0, Confluent Manager for Apache Flink® (CMF) lets you extend Flink SQL with
your own user-defined functions (UDFs). You implement a UDF as a Java class,
package it as a JAR, upload the JAR to
[artifact management](../../configure/artifacts.md#cmf-artifacts), and register the function in an
[environment catalog](../../configure/environment-catalog.md#cmf-environment-catalog) with a `CREATE FUNCTION`
statement.

After registration, the function is available to statements that run in that
environment, the same way a built-in function is.

## Prerequisites

- The [environment catalog](../../configure/environment-catalog.md#cmf-environment-catalog-enable) must be enabled.
- [Artifact management](../../configure/artifacts.md#cmf-artifacts-enable) must be enabled, so that you
  can upload the function JAR and reference it from your DDL.
- Submit function statements with the current catalog set to the environment
  catalog (`_env_<environment>`) and the current database set to `default`.
  For details, see [Manage Environment Catalogs in Confluent Manager for Apache Flink](../../configure/environment-catalog.md#cmf-environment-catalog).

## Write and package a function

Implement the function as a Java class that extends the appropriate Flink base
class, for example `ScalarFunction` for a scalar function:

```java
package com.example;

import org.apache.flink.table.functions.ScalarFunction;

public class ToUpperCase extends ScalarFunction {
    public String eval(String input) {
        return input == null ? null : input.toUpperCase();
    }
}
```

Compile the class against `org.apache.flink:flink-table-common` as a
`provided` dependency, target Java 11, and package it as a JAR. If the function
depends on libraries that are not on the CMF or Flink classpath, shade them into
the JAR rather than widening the class allowlist (see [Limitations]()).

## Deliver the function JAR

Upload the JAR to [artifact management](../../configure/artifacts.md#cmf-artifacts-upload), then
reference it from `CREATE FUNCTION` with a `cmf://` URI. CMF downloads the
JAR when it compiles the statement, and the Flink cluster fetches it when it runs
the job, so you do not need to rebuild any images and you can version the JAR
independently.

Put the artifact metadata in a file, `artifact.json`:

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "udfs.jar"
  },
  "spec": {}
}
```

Upload the JAR, passing `artifact.json` in the `artifact` part:

```bash
curl -X POST http://cmf:8080/cmf/api/v1/environments/prod/artifacts \
  -F 'artifact=@artifact.json;type=application/json' \
  -F 'file=@target/udfs.jar'
```

## Register a function

Register a function with `CREATE FUNCTION`, using the `USING JAR` clause to
reference the uploaded artifact.

### Syntax

```sql
CREATE FUNCTION [IF NOT EXISTS] function_name
  AS 'fully.qualified.ClassName'
  USING JAR 'cmf://{environment}/{artifact-name}[?version={N}]'
    [, JAR 'cmf://{environment}/{artifact-name}[?version={N}]']*;
```

- `IF NOT EXISTS`: Prevents an error if a function with the same name already
  exists.
- `AS`: The fully qualified name of the Java class that implements the function.
- `USING JAR`: The `cmf://` URI of the artifact that contains the class. If
  the function’s classes span more than one artifact, add additional
  `, JAR '...'` clauses. `{environment}` must be the environment in which
  you register the function. CMF rejects a reference to an artifact in another
  environment. `{N}` must be a positive integer or `latest`. For the URI
  form and version behavior, see [Reference artifacts in SQL statements](../../configure/artifacts.md#cmf-artifacts-reference-sql).

### Example

Register a function from a managed artifact and pin it to a version:

```bash
confluent flink statement create create-to-upper \
  --url http://cmf:8080 \
  --environment prod \
  --compute-pool pool \
  --catalog _env_prod \
  --database default \
  --sql "CREATE FUNCTION to_upper AS 'com.example.ToUpperCase' USING JAR 'cmf://prod/udfs.jar?version=1';"
```

## Use a function

Call the function in a query the same way you call a built-in function, and list
the registered functions with `SHOW USER FUNCTIONS`:

```sql
SHOW USER FUNCTIONS;

SELECT to_upper(name) AS name_upper FROM users;
```

## Drop a function

```sql
DROP FUNCTION [IF EXISTS] function_name;
```

Dropping a function does not affect jobs that are already running; they keep
using the version they were compiled with.

## Limitations

- User-defined functions must be implemented in Java. The `LANGUAGE` clause for
  other languages, such as Python or Scala, is not supported.
- Process table functions (PTFs) are not supported.
- `ALTER FUNCTION` is not supported. To change a function’s class or JAR, drop
  the function and create it again.
- `CREATE TEMPORARY FUNCTION` and `CREATE TEMPORARY SYSTEM FUNCTION` are not
  supported.
- When CMF compiles a statement, function code can load only classes whose
  fully qualified names match the `cmf.sql.classAllowlist` configuration. The
  default allowlist has seven entries. Three cover the JDK: `java`,
  `javax`, and `jdk.internal`. The others cover Flink, Confluent Flink
  packages, SLF4J, and Kafka clients. If you set a list, it replaces the whole
  default list. Keep all of the default entries if you add to it. Shade any
  other dependencies into the function JAR instead of widening the allowlist.

## Security considerations

Review the following guidance before you enable user-defined functions in
production.

#### IMPORTANT
CMF does not sandbox user-defined function code. To compile a statement,
CMF loads and runs the function class in the CMF process, and the Flink
cluster runs it during the job. Treat a UDF JAR as trusted code that runs with
CMF privileges. Review and scan function code before you upload it, restrict
who can upload artifacts and submit statements with RBAC, and run UDF
workloads on dedicated compute pools rather than shared ones. The class
allowlist is a best-effort control, not a security boundary.

## Related content

- [Manage Environment Catalogs in Confluent Manager for Apache Flink](../../configure/environment-catalog.md#cmf-environment-catalog)
- [Manage Artifacts in Confluent Manager for Apache Flink](../../configure/artifacts.md#cmf-artifacts)
- [Custom Connectors and Formats in Confluent Manager for Apache Flink](custom-connectors.md#cmf-sql-custom-connectors)
- [Redact a Column with a User-Defined Function](examples/udf.md#cmf-sql-example-udf)
- [Features & Support for Statements in Confluent Manager for Apache Flink](features-support.md#flink-sql-features-support)