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

# Process Table Functions in Confluent Manager for Apache Flink

Starting with version 2.5.0, Confluent Manager for Apache Flink® (CMF) supports process table functions
(PTFs) on compute pools that use Flink 2.3. A PTF is a user-defined function
that takes a table as input and returns a table. It can transform each row on its
own, or process the rows of each key together and keep state for every key, so
you can write custom stream processing logic, such as per-key counters, in Java
and call it from Flink SQL.

You package a PTF 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 `CREATE FUNCTION`,
the same way you register a [user-defined function](user-defined-functions.md#cmf-sql-udf).

## Prerequisites

- A compute pool that runs Flink 2.3. PTFs are an Flink 2.1 feature, so compute
  pools that run Flink 1.19 cannot run them. For how to create a compute pool
  for Flink 2.3, see [Flink Versions for Statements in Confluent Manager for Apache Flink](flink-versions.md#cmf-sql-flink-versions).
- The [environment catalog](../../configure/environment-catalog.md#cmf-environment-catalog-enable) and
  [artifact management](../../configure/artifacts.md#cmf-artifacts-enable) must both be enabled.
- At least one compile worker. When the environment catalog and artifact
  management are both enabled, the Helm chart adds one compile worker by
  default. Do not set `cmf.statementService.workers.count` to 0. For details,
  see [Compile workers](../../installation/sql-compilers.md#cmf-compile-workers).

Set the current catalog to the environment catalog, `_env_<environment>`, and
the current database to `default`. Then submit the statements on this page to
the Flink 2.3 compute pool.

The examples on this page use the `clicks` table in the examples catalog. To
use the catalog, install CMF with `cmf.sql.examples-catalog.enabled=true`.

## Row and set semantics

A PTF declares how Flink passes it each table argument:

Row semantics
: The function processes each row on its own. Flink can send any row to any
  parallel instance of the function, so the function cannot keep state for a
  key. Use row semantics to filter, reshape, or expand rows.

Set semantics
: The function processes the rows that share a key together. You set the key
  with `PARTITION BY` when you call the function, and the function can keep
  state for each key. Use set semantics for per-key logic, such as counting or
  deduplicating events.

## Write a PTF

A PTF is a Java class that extends `ProcessTableFunction` and implements an
`eval` method. Annotate the table argument with `@ArgumentHint` to choose
row or set semantics, and give every argument a name, so that callers can pass
arguments by name. The function emits output rows by calling `collect`.

The following function has row semantics. For each click that has a view time
of at least `min_view_time`, it emits the click’s URL and view time:

```java
package com.example;

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.ArgumentTrait;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.types.Row;

@DataTypeHint("ROW<url STRING, view_time INT>")
public class LongViews extends ProcessTableFunction<Row> {
    public void eval(
            @ArgumentHint(value = ArgumentTrait.ROW_SEMANTIC_TABLE, name = "input") Row input,
            @ArgumentHint(name = "min_view_time") int minViewTime) {
        Integer viewTime = input.getFieldAs("view_time");
        if (viewTime != null && viewTime >= minViewTime) {
            collect(Row.of(input.getFieldAs("url"), viewTime));
        }
    }
}
```

The following function has set semantics. It keeps a click count for each key
in state that it declares with `@StateHint`, and emits each click’s URL with
the count so far. A state class needs a public no-argument constructor and
public fields:

```java
package com.example;

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.ArgumentTrait;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.StateHint;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.types.Row;

@DataTypeHint("ROW<url STRING, click_count BIGINT>")
public class CountClicks extends ProcessTableFunction<Row> {

    public static class ClickState {
        public long count = 0L;
    }

    public void eval(
            @StateHint(name = "clicks") ClickState state,
            @ArgumentHint(value = ArgumentTrait.SET_SEMANTIC_TABLE, name = "input") Row input) {
        state.count++;
        collect(Row.of(input.getFieldAs("url"), state.count));
    }
}
```

Compile the classes against `org.apache.flink:flink-table-common` version
2.3.0 as a `provided` dependency, target Java 17, and package them as a JAR:

```xml
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-common</artifactId>
  <version>2.3.0</version>
  <scope>provided</scope>
</dependency>
```

For the complete PTF API, including timers, multiple table arguments, and
updating input tables, see
[Process Table Functions](https://nightlies.apache.org/flink/flink-docs-release-2.3/docs/dev/table/functions/ptfs/)
in the Flink documentation.

## Upload the JAR

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

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "ptfs.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/ptfs.jar'
```

For artifact names, versions, and updates, see [Manage Artifacts in Confluent Manager for Apache Flink](../../configure/artifacts.md#cmf-artifacts).

## Register a PTF

Register each PTF with `CREATE FUNCTION`, using the `USING JAR` clause to
reference the uploaded artifact. The syntax and the `cmf://` URI rules are the
same as for [user-defined functions](user-defined-functions.md#cmf-sql-udf).

```bash
confluent flink statement create create-long-views \
  --url http://cmf:8080 \
  --environment prod \
  --compute-pool pool-2-3 \
  --catalog _env_prod \
  --database default \
  --sql "CREATE FUNCTION long_views AS 'com.example.LongViews' USING JAR 'cmf://prod/ptfs.jar';"
```

Register the set-semantic function the same way, and then list the registered
functions. Submit each of the following statements with the same options as the
previous command:

```sql
CREATE FUNCTION count_clicks AS 'com.example.CountClicks' USING JAR 'cmf://prod/ptfs.jar';

SHOW USER FUNCTIONS;
```

## Call a PTF

Call a PTF in the `FROM` clause of a query. Pass a table argument with the
`TABLE` keyword, and pass each argument by name with `=>`.

A function with row semantics takes the table as it is:

```sql
SELECT * FROM long_views(input => TABLE examples.marketplace.clicks, min_view_time => 100);
```

The result has the columns that the function emits, `url` and `view_time`.

A function with set semantics requires a `PARTITION BY` clause on the table
argument:

```sql
SELECT * FROM count_clicks(input => TABLE examples.marketplace.clicks PARTITION BY user_id);
```

The partition key columns come first in the result, followed by the columns
that the function emits, so the result has the columns `user_id`, `url`,
and `click_count`. Without `PARTITION BY`, the statement fails validation.

## Drop a PTF

To drop a PTF, use `DROP FUNCTION`:

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

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

## Limitations

- PTFs run only on compute pools that use Flink 2.3. Register and call them from
  statements on those compute pools.
- The limitations of [user-defined functions](user-defined-functions.md#cmf-sql-udf) also apply to
  PTFs: they must be implemented in Java, `ALTER FUNCTION` and temporary
  functions are not supported, and function code can load only classes that
  match `cmf.sql.classAllowlist`.

## Security considerations

#### IMPORTANT
To compile a statement, a compile worker loads and runs the function class,
and the Flink cluster runs it during the job. Treat a PTF JAR as trusted code.
Review and scan function code before you upload it, restrict who can upload
artifacts and submit statements with RBAC, and run PTF workloads on
dedicated compute pools rather than shared ones. Secure the compile workers as
described in [Secure the compile workers](../../installation/sql-compilers.md#cmf-compile-workers-secure). The class allowlist is a
best-effort control, not a security boundary.

## Related content

- [User-Defined Functions in Confluent Manager for Apache Flink](user-defined-functions.md#cmf-sql-udf)
- [Flink Versions for Statements in Confluent Manager for Apache Flink](flink-versions.md#cmf-sql-flink-versions)
- [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)
- [Features & Support for Statements in Confluent Manager for Apache Flink](features-support.md#flink-sql-features-support)