<a id="flink-ptfs"></a>

# Process Table Functions in Confluent Cloud for Apache Flink

A *process table function* (PTF) is the most flexible type of
user-defined function in Confluent Cloud for Apache Flink®. PTFs are the “escape hatch” to
Flink’s lowest-level stream-processing primitives — managed state,
event-time timers, and multi-table inputs — for cases where standard SQL
and simple user-defined functions can’t express your business logic.

Conceptually, a PTF is itself a user-defined function that provides all the
primitives to build functions ranging from a simple scalar function to advanced
windowing and aggregation functions.

PTFs are also called *polymorphic table functions* based on the SQL:2016
standard.

#### NOTE
Flink PTFs are available in all regions that support UDFs. For more
information, see [UDF regional availability](user-defined-functions.md#flink-sql-udfs-availability). For information about
limitations and unsupported features, see [Process table function limitations](#flink-ptfs-limitations).

## What are process table functions?

Unlike traditional user-defined functions that process one row at a time (1-to-1)
or aggregate functions that process many rows to produce one result (N-to-1),
PTFs support N-to-M semantics. This means a PTF can:

- Consume multiple input rows from one or more tables.
- Emit multiple output rows.
- Maintain persistent, per-partition state across events.
- Schedule future actions using event-time timers.

PTFs combine the simplicity of serverless functions with the power of stateful
stream processing. They provide an escape hatch when SQL alone cannot express
your business logic.

<a id="flink-ptfs-differences-from-oss"></a>

## Differences from Apache Flink

PTFs in Confluent Cloud for Apache Flink® are based on the Flink 2.2 `ProcessTableFunction` API.
The 20-table limit on multi-table arguments, event-time timers, changelog
support (the `SUPPORT_UPDATES` and `REQUIRE_UPDATE_BEFORE` traits and the
`ChangelogFunction` interface), and the `PASS_COLUMNS_THROUGH` trait for
single-table PTFs all behave the same as Flink open source. The table below
lists the runtime differences to keep in mind when porting a PTF between the
two.

| Area                                                   | Flink open source                                                                     | Confluent Cloud for Apache Flink®                                                    |
|--------------------------------------------------------|---------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------|
| `@StateHint(ttl=…)`                                    | Supported                                                                             | Supported                                                                            |
| `PARTITION BY` on a `SET_SEMANTIC_TABLE` argument      | Mandatory by default; can be made optional with the<br/>`OPTIONAL_PARTITION_BY` trait | Required                                                                             |
| `MapView` and `ListView` state types                   | Supported                                                                             | Not supported. Use a plain `Map` or `List` field in a<br/>`@StateHint` POJO instead. |
| `TIMESTAMP_LTZ` precision in PTF input and output rows | Up to nanoseconds (precision 9)                                                       | Capped at microseconds (precision 6)                                                 |
| State POJO field visibility                            | Public fields, or private fields with accessors                                       | Public fields with default values required                                           |

## Why use process table functions?

Use PTFs when you need to implement complex, stateful logic that cannot be
achieved with standard SQL or simple user-defined functions. Common use cases
include:

- **Custom windowing logic**: Build hybrid windows that emit data based on time
  intervals or event counts.
- **Deduplication with state**: Remember when events were seen over a specific
  time period to remove duplicates.
- **Timer-based alerts**: Trigger alerts when expected events do not occur within
  a time threshold.
- **Complex state machines**: Track multi-step processes, such as user journeys
  through a checkout flow.
- **Control streams**: Use a low-cardinality side input to mutate per-key
  behavior in a main stream.
- **Real-time analytics**: Calculate custom metrics that require accessing
  historical state.

#### NOTE
Reach for built-in SQL operators first. PTFs give you direct access to
state, timers, and side outputs — but with that power comes
responsibility for managing each one correctly. If a `TUMBLE`,
`OVER`, `MATCH_RECOGNIZE`, or built-in function already does what
you need, use it instead.

<a id="flink-ptfs-common-patterns"></a>

## Common PTF patterns

The following patterns recur often enough across customer workloads that
they are worth recognizing as templates. Each is a thumbnail; full
walkthroughs live in [Create a Process Table Function](../how-to-guides/create-ptf.md#flink-ptfs-quickstart)
and the Flink OSS documentation.

| Pattern                              | Shape                                                                                                                                                                                    |
|--------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Time-bounded deduplication           | `@StateHint` boolean per ID + event-time timer that clears the<br/>state after the deduplication window.                                                                                 |
| Event-absence / timeout detection    | Store the triggering event in state, register a timer for the<br/>deadline, and `clearTimer()` if the expected follow-up event<br/>arrives. See [Event-time timers](#flink-ptfs-timers). |
| Custom micro-batching                | `@StateHint` `List` buffer + count- and time-based emit<br/>conditions inside `eval()` and `onTimer()`.                                                                                  |
| Temporal stream join with TTL        | Multi-table PTF that stores one side of the join in state and uses<br/>`@StateHint(ttl=…)` to evict orphaned rows. See<br/>[Multi-table inputs](#flink-ptfs-multi-table).                |
| Threshold breaching with early reset | Counter in state + timer for the rolling window, plus an explicit<br/>`clearAll()` on the “success” or “reset” event.                                                                    |
| Missing-heartbeat / liveness check   | Named timer that’s replaced on each heartbeat; `onTimer()` fires<br/>only when no heartbeat arrives in time.                                                                             |
| Sensor debouncing                    | Last-seen value in state; emit only when the new value differs by<br/>more than a threshold.                                                                                             |
| Early-firing sessionization          | `@StateHint` `List` buffers the session; a terminal event<br/>(checkout, logout) flushes and clears state immediately instead of<br/>waiting for the inactivity timer.                   |

## PTFs vs other function types

The following table compares PTFs with other function types available in Flink:

| Feature           | Scalar UDF             | Table UDF               | Aggregate UDF       | Process table function     |
|-------------------|------------------------|-------------------------|---------------------|----------------------------|
| Input rows        | 1                      | 1                       | N                   | N                          |
| Output rows       | 1                      | N                       | 1                   | N                          |
| Managed state     | No                     | No                      | No                  | Yes                        |
| Timers            | No                     | No                      | No                  | Yes                        |
| Multi-table input | No                     | No                      | No                  | Yes                        |
| Use case          | Transform single value | Split one row into many | Aggregate many rows | Stateful stream processing |

## How do process table functions work?

PTFs operate on input tables and produce output tables. A PTF receives one or
more table arguments, processes rows within each logical partition, and emits
output rows. You call a PTF from SQL using named arguments:

```sql
SELECT user_id, click_id, running_count
FROM EventCounter(
    input => TABLE examples.marketplace.clicks PARTITION BY user_id,
    uid => 'event-counter-v1'
);
```

The key elements of a PTF invocation are:

`TABLE ... PARTITION BY`
: Passes an input table to the PTF and defines the row grouping. The PTF
  processes each unique `user_id` independently with its own isolated state.
  In Confluent Cloud for Apache Flink®, `PARTITION BY` is required on every `SET_SEMANTIC_TABLE`
  argument.

`uid => '...'`
: Assigns a stable identifier for state persistence across query restarts.

You can use PTFs in two ways:

SQL-first approach
: Register a PTF in Java and call it directly from a SQL `SELECT` statement
  to solve complex windowing or stateful processing problems, as the previous
  example shows.

Table API approach
: Build full applications using the Table API with PTFs. Define state using Java
  annotations and deploy to Confluent Cloud. The platform handles automatic scaling,
  exactly-once processing, and serverless operations.

Because PTFs are discovered through Java reflection, you must annotate the
class, the eval method, and its parameters to tell the runtime how to call
the function and what to return. The annotation surface is small:

| Annotation                                               | Purpose                                                                                                                                                                                                     |
|----------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `@DataTypeHint("ROW<…>")` on the class                   | Declares the PTF’s output schema. Required whenever the function<br/>extends `ProcessTableFunction<Row>`; auto-inferred for typed<br/>parameterizations like `ProcessTableFunction<Long>`.                  |
| `@StateHint` on an `eval()` parameter                    | Marks a parameter as managed, per-partition state. The state class<br/>is a POJO with public fields. Add `ttl="…"` to auto-expire idle<br/>state.                                                           |
| `@ArgumentHint(SET_SEMANTIC_TABLE)` on a `Row` parameter | Declares that the parameter is a table argument. Combine with other<br/>traits like `PASS_COLUMNS_THROUGH` or `REQUIRE_ON_TIME` as<br/>needed. See [Table argument semantics](#flink-ptfs-table-semantics). |
| `@DataTypeHint("…")` on a scalar `eval()` parameter      | Declares the SQL type of a non-table argument. Useful when Java<br/>reflection can’t infer the precision you want (e.g.,<br/>`@DataTypeHint("DECIMAL(10, 2)")`).                                            |

<a id="flink-ptfs-table-semantics"></a>

## Table argument semantics

A PTF declares each table argument with one of two argument traits:

`SET_SEMANTIC_TABLE`
: The PTF processes a logical partition of rows, defined by the
  `PARTITION BY` clause on the SQL invocation. Set semantics give the PTF
  access to managed state, state TTL, and timers — all of which are scoped
  per partition. Every stateful example on this page uses set semantics. In
  Confluent Cloud for Apache Flink®, `PARTITION BY` is required on every `SET_SEMANTIC_TABLE`
  argument.

`ROW_SEMANTIC_TABLE`
: The PTF processes one row at a time, with no partition concept and no
  per-key state. Row semantics are useful for stateless enrichment, where
  the PTF acts as a 1:1 row transformer. Use `ROW_SEMANTIC_TABLE` when
  the PTF doesn’t need `PARTITION BY`, `@StateHint`, or timers.

Declare the trait on the table parameter with `@ArgumentHint`:

```java
@ArgumentHint(SET_SEMANTIC_TABLE) Row input        // partition-scoped
@ArgumentHint(ROW_SEMANTIC_TABLE) Row input        // row-by-row
```

Either semantic can be combined with other traits — for example,
`@ArgumentHint({SET_SEMANTIC_TABLE, PASS_COLUMNS_THROUGH})` for
pass-through enrichment within a partition, or
`@ArgumentHint({SET_SEMANTIC_TABLE, REQUIRE_ON_TIME})` when the PTF
schedules event-time timers.

<a id="flink-ptfs-state-ttl"></a>

## State TTL

Confluent Cloud for Apache Flink® supports automatic state expiration on PTF state entries through
the `ttl` attribute of `@StateHint`. When a partition key receives no read
or write for the configured duration, Flink garbage-collects the state entry.
Use state TTL to bound the size of long-running stateful PTFs without writing
manual cleanup logic.

Set `ttl` to a duration string on the `@StateHint` annotation:

```java
import static org.apache.flink.table.annotation.ArgumentTrait.SET_SEMANTIC_TABLE;

import org.apache.flink.table.annotation.ArgumentHint;
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;

// Output excludes the partition key — Flink auto-prepends it from PARTITION BY.
@DataTypeHint("ROW<counter BIGINT>")
public class TtlCountdown extends ProcessTableFunction<Row> {

    public static class CountState {
        public long counter = 0L;
    }

    public void eval(
        @StateHint(ttl = "30 seconds") CountState state,
        @ArgumentHint(SET_SEMANTIC_TABLE) Row input
    ) {
        state.counter++;
        collect(Row.of(state.counter));
    }
}
```

In this example, `state.counter` resets to `0` if no event arrives for the
partition key within 30 seconds. TTL applies independently to every partition
key, so an active key is unaffected by other keys’ inactivity.

The `ttl` attribute accepts standard Flink duration syntax: `"30 seconds"`,
`"5 minutes"`, `"1 hour"`, `"7 days"`. TTL is measured against
state-backend access time, not event time.

<a id="flink-ptfs-multi-table"></a>

## Multi-table inputs

A PTF can declare multiple `SET_SEMANTIC_TABLE` arguments. Each input table
is co-partitioned by its `PARTITION BY` key and delivered to the same PTF
instance, so the function can join, enrich, or correlate streams across inputs
with shared state. Use multi-table inputs for control streams that mutate
per-key behavior in a main stream, side inputs that look up reference data,
or self-joins that need two independent traversals of the same source.

```java
import static org.apache.flink.table.annotation.ArgumentTrait.SET_SEMANTIC_TABLE;

import org.apache.flink.table.annotation.ArgumentHint;
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;

// Output excludes the partition key — Flink auto-prepends tenant_id from PARTITION BY.
@DataTypeHint("ROW<payload STRING, active_mode STRING>")
public class MtControlStream extends ProcessTableFunction<Row> {

    public static class TenantConfig {
        public String activeMode = "default";
    }

    public void eval(
        @StateHint TenantConfig state,
        @ArgumentHint(SET_SEMANTIC_TABLE) Row events,
        @ArgumentHint(SET_SEMANTIC_TABLE) Row control
    ) {
        // Control rows update the per-tenant active mode; no row is emitted.
        if (control != null) {
            String mode = control.getFieldAs("mode");
            if (mode != null) {
                state.activeMode = mode;
            }
            return;
        }
        // Event rows are tagged with the current mode.
        if (events != null) {
            String payload = events.getFieldAs("payload");
            collect(Row.of(payload, state.activeMode));
        }
    }
}
```

Call the PTF with both tables in a single SQL invocation:

```sql
SELECT *
FROM MtControlStream(
    events  => TABLE events  PARTITION BY tenant_id,
    control => TABLE control PARTITION BY tenant_id
);
```

Confluent Cloud for Apache Flink® supports up to 20 table arguments per PTF, the same as Flink open
source. Every `SET_SEMANTIC_TABLE` argument must include a `PARTITION BY`
clause; Confluent Cloud for Apache Flink® does not allow unpartitioned set-semantic tables.

<a id="flink-ptfs-pass-through"></a>

## Pass-through columns

The `PASS_COLUMNS_THROUGH` trait copies every column of the input table
into the output. Use it when a PTF enriches each row with one or more
additional fields without re-emitting the input schema in Java.

```java
import static org.apache.flink.table.annotation.ArgumentTrait.PASS_COLUMNS_THROUGH;
import static org.apache.flink.table.annotation.ArgumentTrait.SET_SEMANTIC_TABLE;

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

public class PassThroughEnrich extends ProcessTableFunction<Long> {

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

    public void eval(
        @StateHint CountState state,
        @ArgumentHint({SET_SEMANTIC_TABLE, PASS_COLUMNS_THROUGH}) Row input
    ) {
        state.count++;
        collect(state.count);
    }
}
```

The output table contains every column of `input` followed by the
single `running_count` value emitted by `collect()`. The PTF does not
have to project or re-emit the input columns — Flink splices them in
automatically. `PASS_COLUMNS_THROUGH` applies to single-table PTFs only;
multi-table PTFs cannot use this trait, the same as in Flink open source.

<a id="flink-ptfs-changelog-mode"></a>

## Changelog mode

By default, a PTF accepts only an append-only (insert-only) input table: every
row is a new fact, and rows are never updated or retracted. Many streaming
sources are instead [changelog](../../_glossary.md#term-changelog) (updating) streams — for example, an
[upsert](../../_glossary.md#term-upsert)
table with a `PRIMARY KEY`, a versioned view, or a change-data-capture (CDC)
feed from a database. These streams emit inserts, updates, and deletions.

To consume a changelog input table, add the `SUPPORT_UPDATES` argument trait
to a `SET_SEMANTIC_TABLE` argument. Without it, a table argument is
insert-only and updates are rejected. Inspect each row’s *row kind* with
`Row.getKind()` to branch on the change type:

- `INSERT` — a new row.
- `UPDATE_BEFORE` — the previous image of an updated row.
- `UPDATE_AFTER` — the new image of an updated row.
- `DELETE` — a removed row.

Which row kinds actually enter your function depends on how the source encodes
its changelog. An upsert source — for example, a `PRIMARY KEY` table backed by
an upsert changelog, partitioned by its key — delivers `UPDATE_AFTER` and
`DELETE`, and does not necessarily synthesize a separate `INSERT` for the
first row of a key or a preceding `UPDATE_BEFORE`. To normalize the input so
that the first row per key arrives as `INSERT` and every update arrives as an
`UPDATE_BEFORE` / `UPDATE_AFTER` pair, also add the `REQUIRE_UPDATE_BEFORE`
trait; Flink then keeps the per-key state needed to reconstruct the previous
image.

The following PTF consumes a changelog stream of currency rates and converts it
into an append-only audit trail. It declares `REQUIRE_UPDATE_BEFORE` so the
first change for each currency arrives as an `INSERT`, and it keeps the
previous rate for each key in state so that each emitted row carries both the
old and new value.

```java
import static org.apache.flink.table.annotation.ArgumentTrait.REQUIRE_UPDATE_BEFORE;
import static org.apache.flink.table.annotation.ArgumentTrait.SET_SEMANTIC_TABLE;
import static org.apache.flink.table.annotation.ArgumentTrait.SUPPORT_UPDATES;

import java.math.BigDecimal;
import org.apache.flink.table.annotation.ArgumentHint;
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;

// Output excludes the partition key — Flink auto-prepends currency from PARTITION BY.
@DataTypeHint("ROW<change_type STRING, old_rate DECIMAL(10, 4), new_rate DECIMAL(10, 4)>")
public class ChangelogAuditor extends ProcessTableFunction<Row> {

    public static class RateState {
        @DataTypeHint("DECIMAL(10, 4)")
        public BigDecimal previousRate = null;
    }

    public void eval(
        @StateHint RateState state,  // Add ttl for production use; see the how-to example.
        @ArgumentHint({SET_SEMANTIC_TABLE, SUPPORT_UPDATES, REQUIRE_UPDATE_BEFORE}) Row input
    ) {
        BigDecimal rate = input.getFieldAs("rate");

        switch (input.getKind()) {
            case INSERT:
                collect(Row.of("INSERT", null, rate));
                state.previousRate = rate;
                break;
            case UPDATE_AFTER:
                collect(Row.of("UPDATE", state.previousRate, rate));
                state.previousRate = rate;
                break;
            case DELETE:
                collect(Row.of("DELETE", state.previousRate, null));
                state.previousRate = null;
                break;
            case UPDATE_BEFORE:
                // Ignored: UPDATE_AFTER carries the new value, and the prior
                // value is already tracked in state.
                break;
        }
    }
}
```

Keep the following in mind when you consume changelog input:

- `SUPPORT_UPDATES` and `REQUIRE_UPDATE_BEFORE` apply to
  `SET_SEMANTIC_TABLE` arguments. Partition the input by the changelog key
  with `PARTITION BY` so that each key’s changes are processed in order by a
  single PTF instance.
- Add `REQUIRE_UPDATE_BEFORE` when your logic depends on seeing an `INSERT`
  for the first row of a key, or an `UPDATE_BEFORE` for each update. Without
  it, an upsert source delivers only `UPDATE_AFTER` and `DELETE`.
- The preceding PTF output is append-only because it emits rows with `Row.of(...)`.
  To emit a *changelog* result instead, implement the `ChangelogFunction`
  interface, declare the changelog you emit from `getChangelogMode()`, and set
  the row kind on each emitted row with `Row.ofKind(RowKind.UPDATE_AFTER, ...)`.
  For upsert output, the upsert key must match the `PARTITION BY` key, and the
  sink must accept the changelog (for example, a table with a `PRIMARY KEY`).
- `SUPPORT_UPDATES` can’t be combined with the `PASS_COLUMNS_THROUGH` trait.

For complete, deployable walkthroughs, see
[Example: Changelog audit trail](../how-to-guides/create-ptf.md#flink-ptfs-changelog-audit-example)
(consuming a changelog) and
[Example: Emit a changelog](../how-to-guides/create-ptf.md#flink-ptfs-emit-changelog-example)
(producing one).

<a id="flink-ptfs-timers"></a>

## Event-time timers

PTFs support event-time timers for scheduling callbacks at specific
[watermark](timely-stream-processing.md#flink-sql-timely-stream-processing) positions. Timers enable
time-based logic such as session timeouts, delayed processing, and alerting
when expected events do not arrive.

### How timers work

A timer fires when the current watermark advances past the timer’s registered
timestamp. The virtual processor defined by `PARTITION BY` scopes the
timers — each partition maintains its own independent set of timers. Flink
persists and restores timers during failures and restarts.

To use timers, add `REQUIRE_ON_TIME` to the `@ArgumentHint` annotation on
the table argument. This makes the `on_time` descriptor argument mandatory
in the SQL call, which specifies which column provides the event-time timestamp.

### Registering and clearing timers

Register timers inside `eval()` or `onTimer()` using the `TimeContext`
API:

```java
TimeContext<Instant> timeCtx = ctx.timeContext(Instant.class);

// Named timer — can be replaced or cleared by name
timeCtx.registerOnTime("timeout", timeCtx.time().plusSeconds(60));

// Unnamed timer — one per timestamp value
timeCtx.registerOnTime(timeCtx.time().plus(Duration.ofMinutes(5)));
```

To clear timers:

```java
timeCtx.clearTimer("timeout");       // Clear a named timer
timeCtx.clearTimer(someTimestamp);    // Clear an unnamed timer by timestamp
ctx.clearAllTimers();                 // Clear all timers in the partition
ctx.clearAll();                       // Clear all state and timers
```

### The onTimer() method

When a timer fires, Flink calls the `onTimer()` method. This method has access
to the same state entries declared in `eval()`:

```java
public void onTimer(OnTimerContext onTimerCtx, MyState state) {
    String timerName = onTimerCtx.currentTimer(); // null for unnamed timers
    collect(new Alert(state.userId, "Timeout reached"));
}
```

The `onTimer()` method signature supports:

1. An optional `OnTimerContext` as the first parameter.
2. Zero or more `@StateHint` parameters matching those in `eval()`.

### SQL invocation with timers

When calling a PTF with timers, use the `on_time` descriptor argument to
specify the watermarked timestamp column:

```sql
SELECT *
FROM InactivityAlert(
    input => TABLE user_events PARTITION BY user_id,
    on_time => DESCRIPTOR(event_time),
    uid => 'inactivity-alert-v1'
);
```

The `on_time` column must have a watermark declaration on the source table.

### Common timer pitfalls

A few timer mistakes appear repeatedly in PTF code reviews. Knowing them
up-front saves debugging time:

- **Duplicate registration.** Re-registering a named timer silently
  replaces the previous one — this is usually what you want (it’s the
  “reset on activity” pattern). Re-registering an *unnamed* timer at the
  same timestamp inflates state for no benefit; prefer named timers when
  the registration is conditional.
- **Forgotten cancellation.** A timer that’s no longer needed but still
  registered will still fire. `onTimer()` then runs against whatever
  state it finds — which may be empty, stale, or wrong for the original
  condition. Call `clearTimer()` as soon as the awaited event arrives.
- **Event-time vs. processing-time confusion.** Event-time timers fire on
  *watermark advance*, not on wall-clock progress. A stream whose
  watermark stops advancing — for example, an idle partition with no
  source rows — will never fire any event-time timer, even after hours of
  real time. Inspect watermarks first when an alert “never fires”.

### Timer best practices

- Clean up timers that you no longer need. Use `clearTimer()` after the
  condition the timer was waiting for is satisfied.
- Avoid registering an excessive number of timers per partition. Each timer
  consumes state, and large numbers of timers can affect performance and
  recovery time.
- Use named timers when you need to replace or cancel a specific timer.
  Named timers use a string name as their identifier, and re-registering
  the same name replaces the previous timer.
- Use unnamed timers when you need multiple independent timers at different
  timestamps within the same partition.

<a id="flink-ptfs-limitations"></a>

## Process table function limitations

This section describes the limitations and unsupported features of
[process table functions (PTFs)](#flink-ptfs) in Confluent Cloud for Apache Flink.

### State view limitations

#### ListView not supported

`ListView` state is not available in PTFs. `ListView` is a specialized
state interface designed for extremely large state collections that support
iteration but not random seeking.

For lists that fit in memory, use `@StateHint` with a state POJO containing
a `List` field. Flink handles serialization automatically:

```java
import static org.apache.flink.table.annotation.ArgumentTrait.SET_SEMANTIC_TABLE;

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.StateHint;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.types.Row;
import java.util.ArrayList;
import java.util.List;

public class ListStatePTF extends ProcessTableFunction<Row> {

    public static class ListStateData {
        public List<String> items = new ArrayList<>();
    }

    public void eval(
        @StateHint ListStateData state,
        @ArgumentHint(SET_SEMANTIC_TABLE) Row input
    ) {
        state.items.add(input.getFieldAs(0).toString());
    }
}
```

#### MapView not supported

`MapView` state is not available in PTFs. `MapView` is a specialized state
interface designed for extremely large state collections.

For maps that fit in memory, use `@StateHint` with a state POJO containing
a `Map` field. Flink handles serialization automatically:

```java
import static org.apache.flink.table.annotation.ArgumentTrait.SET_SEMANTIC_TABLE;

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.StateHint;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.types.Row;
import java.util.HashMap;
import java.util.Map;

public class MapStatePTF extends ProcessTableFunction<Row> {

    public static class MapStateData {
        public Map<String, Integer> counts = new HashMap<>();
    }

    public void eval(
        @StateHint MapStateData state,
        @ArgumentHint(SET_SEMANTIC_TABLE) Row input
    ) {
        String key = input.getFieldAs(0).toString();
        state.counts.put(key, state.counts.getOrDefault(key, 0) + 1);
    }
}
```

#### IMPORTANT
`ListView` and `MapView` are designed for state too large to fit in
task-manager memory. For most use cases, a plain `List` or `Map` field
in a `@StateHint` POJO is sufficient and handles serialization
automatically. Pair these fields with `@StateHint(ttl=…)` to bound
growth in long-running statements.

### Language support limitations

Python support for PTFs is not available. PTFs must be written in Java.

For information about Python user-defined functions (non-PTF), see
[Create a User-Defined Function with Confluent Cloud for Apache Flink](../how-to-guides/create-udf.md#flink-sql-create-udf).

## Best practices

- Keep state POJOs small. Avoid storing large collections in `@StateHint`
  fields. For normal-sized `List` or `Map` fields, Flink handles
  serialization automatically, but extremely large state can impact
  performance. `ListView` and `MapView` support for large state
  collections is planned for a future release.
- Be aware that every read or write of a `@StateHint` POJO serializes
  the entire field. A growing `List` or `Map` field will get more
  expensive per event as it grows; bound the growth with
  `@StateHint(ttl=…)` or by partitioning more aggressively so each key
  holds less data. `ListView`/`MapView` will land in a future release
  for truly large collections.
- Use `@StateHint(ttl=…)` to bound state growth in long-running stateful
  PTFs. See [State TTL](#flink-ptfs-state-ttl).
- Choose partition keys carefully. The `PARTITION BY` clause determines
  the state distribution. Too few unique keys can create hot partitions,
  while extremely high cardinality keys increase the total amount of
  state Flink must manage.
- Declare state POJO fields as `public` with default values. Flink requires
  public fields to serialize and deserialize state automatically. Flink
  doesn’t support private fields with getters and setters.
- Test PTFs with realistic data volumes before production deployment.
- Follow [timer best practices](#flink-ptfs-timers) when using
  event-time timers. Clean up timers after you no longer need them to
  avoid excessive state growth.

## Related content

- [Create a Process Table Function](../how-to-guides/create-ptf.md#flink-ptfs-quickstart)
- [Apache Flink PTF Documentation](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/ptfs/)
- [Apache Flink State Documentation](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/fault-tolerance/state/)
- [SQL:2016 Polymorphic Table Functions](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=298781093)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).