org.apache.flink.table.functions.UserDefinedFunction

org.apache.flink.table.functions.ProcessTableFunction<T>

Type Parameters:: T - The type of the output row. Either an explicit composite type or an atomic type implicitly wrapped into a row consisting of one field.

All Implemented Interfaces:: Serializable, FunctionDefinition

@PublicEvolving public abstract class ProcessTableFunction<T> extends UserDefinedFunction

Base class for a user-defined process table function (PTF).

PTFs are the most powerful function kind for Flink SQL and Table API. They enable implementing user-defined operators that can be as feature-rich as built-in operations. PTFs can take (partitioned) tables to produce a new table. They have access to Flink's managed state, event-time and timer services, and underlying table changelogs.

A process table function (PTF) maps zero, one, or multiple tables to zero, one, or multiple rows (or structured types). Scalar arguments are also supported. If the output record consists of only one field, the wrapper can be omitted, and a scalar value can be emitted that will be implicitly wrapped into a row by the runtime.

Table Semantics and Virtual Processors

PTFs can produce a new table by consuming tables as arguments. For scalability, input tables are distributed across so-called "virtual processors". A virtual processor, as defined by the SQL standard, executes a PTF instance and has access only to a portion of the entire table. The argument declaration decides about the size of the portion and co-location of data. Conceptually, tables can be processed either "per row" (i.e. with row semantics) or "per set" (i.e. with set semantics).

Table Argument with Row Semantics

A PTF that takes a table with row semantics assumes that there is no correlation between rows and each row can be processed independently. The framework is free in how to distribute rows across virtual processors and each virtual processor has access only to the currently processed row.

Table Argument with Set Semantics

A PTF that takes a table with set semantics assumes that there is a correlation between rows. When calling the function, the PARTITION BY clause defines the columns for correlation. The framework ensures that all rows belonging to same set are co-located. A PTF instance is able to access all rows belonging to the same set. In other words: The virtual processor is scoped by a key context.

It is also possible not to provide a key (ArgumentTrait.OPTIONAL_PARTITION_BY), in which case only one virtual processor handles the entire table, thereby losing scalability benefits.

Implementation

The behavior of a ProcessTableFunction can be defined by implementing a custom evaluation method. The evaluation method must be declared publicly, not static, and named eval. Overloading is not supported.

For storing a user-defined function in a catalog, the class must have a default constructor and must be instantiable during runtime. Anonymous functions in Table API can only be persisted if the function object is not stateful (i.e. containing only transient and static fields).

Data Types

By default, input and output data types are automatically extracted using reflection. This includes the generic argument T of the class for determining an output data type. Input arguments are derived from the eval() method. If the reflective information is not sufficient, it can be supported and enriched with FunctionHint, ArgumentHint, and DataTypeHint annotations.

The following examples show how to specify data types:


 // Function that accepts two scalar INT arguments and emits them as an implicit ROW < INT >
 class AdditionFunction extends ProcessTableFunction<Integer> {
   public void eval(Integer a, Integer b) {
     collect(a + b);
   }
 }

 // Function that produces an explicit ROW < i INT, s STRING > from scalar arguments, the function hint helps in
 // declaring the row's fields
 @DataTypeHint("ROW< i INT, s STRING >")
 class DuplicatorFunction extends ProcessTableFunction<Row> {
   public void eval(Integer i, String s) {
     collect(Row.of(i, s));
     collect(Row.of(i, s));
   }
 }

 // Function that accepts a scalar DECIMAL(10, 4) and emits it as an explicit ROW < d DECIMAL(10, 4) >
 @FunctionHint(output = @DataTypeHint("ROW< d DECIMAL(10, 4) >"))
 class DuplicatorFunction extends ProcessTableFunction<Row> {
   public void eval(@DataTypeHint("DECIMAL(10, 4)") BigDecimal d) {
     collect(Row.of(d));
     collect(Row.of(d));
   }
 }

Arguments

The ArgumentHint annotation enables declaring the name, data type, and kind of each argument (i.e. ArgumentTrait.SCALAR, ArgumentTrait.ROW_SEMANTIC_TABLE, or ArgumentTrait.SET_SEMANTIC_TABLE). It allows specifying other traits for table arguments as well:


 // Function that has two arguments:
 // "input_table" (a table with set semantics) and "threshold" (a scalar value)
 class ThresholdFunction extends ProcessTableFunction<Integer> {
   public void eval(
       // For table arguments, a data type for Row is optional (leading to polymorphic behavior)
       @ArgumentHint(value = ArgumentTrait.SET_SEMANTIC_TABLE, name = "input_table") Row t,
       // Scalar arguments require a data type either explicit or via reflection
       @ArgumentHint(value = ArgumentTrait.SCALAR, name = "threshold") Integer threshold) {
     int amount = t.getFieldAs("amount");
     if (amount >= threshold) {
       collect(amount);
     }
   }
 }

Table arguments can declare a concrete data type (of either row or structured type) or accept any type of row in a polymorphic fashion:


 // Function with explicit table argument type of row
 class MyPTF extends ProcessTableFunction<String> {
   public void eval(Context ctx, @ArgumentHint(value = ArgumentTrait.SET_SEMANTIC_TABLE, type = @DataTypeHint("ROW < s STRING >")) Row t) {
     TableSemantics semantics = ctx.tableSemanticsFor("t");
     // Always returns "ROW < s STRING >"
     semantics.dataType();
     ...
   }
 }

 // Function with explicit table argument type of structured type "Customer"
 class MyPTF extends ProcessTableFunction<String> {
   public void eval(Context ctx, @ArgumentHint(value = ArgumentTrait.SET_SEMANTIC_TABLE) Customer c) {
     TableSemantics semantics = ctx.tableSemanticsFor("c");
     // Always returns structured type of "Customer"
     semantics.dataType();
     ...
   }
 }

 // Function with polymorphic table argument
 class MyPTF extends ProcessTableFunction<String> {
   public void eval(Context ctx, @ArgumentHint(value = ArgumentTrait.SET_SEMANTIC_TABLE) Row t) {
     TableSemantics semantics = ctx.tableSemanticsFor("t");
     // Always returns "ROW" but content depends on the table that is passed into the call
     semantics.dataType();
     ...
   }
 }

Context

A ProcessTableFunction.Context can be added as a first argument to the eval() method for additional information about the input tables and other services provided by the framework:


 // Function that accesses the Context for reading the PARTITION BY columns and
 // excluding them when building a result string
 class ConcatNonKeysFunction extends ProcessTableFunction<String> {
   public void eval(Context ctx, @ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row inputTable) {
     TableSemantics semantics = ctx.tableSemanticsFor("inputTable");
     List<Integer> keys = Arrays.asList(semantics.partitionByColumns());
     return IntStream.range(0, inputTable.getArity())
       .filter(pos -> !keys.contains(pos))
       .mapToObj(inputTable::getField)
       .map(Object::toString)
       .collect(Collectors.joining(", "));
   }
 }

State

A PTF that takes set semantic tables can be stateful. Intermediate results can be buffered, cached, aggregated, or simply stored for repeated access. A function can have one or more state entries which are managed by the framework. Flink takes care of storing and restoring those during failures or restarts (i.e. Flink managed state).

A state entry is partitioned by a key and cannot be accessed globally. The partitioning (or a single partition in case of no partitioning) is defined by the corresponding function call. In other words: Similar to how a virtual processor has access only to a portion of the entire table, a PTF has access only to a portion of the entire state defined by the PARTITION BY clause. In Flink, this concept is also known as keyed state.

State entries can be added as a mutable parameter to the eval() method. In order to distinguish them from call arguments, they must be declared before any other argument, but after an optional ProcessTableFunction.Context parameter. Furthermore, they must be annotated either via StateHint or declared as part of FunctionHint.state().

For read and write access, only row or structured types (i.e. POJOs with default constructor) qualify as a data type. If no state is present, all fields are set to null (in case of a row type) or fields are set to their default value (in case of a structured type). For state efficiency, it is recommended to keep all fields nullable.


 // Function that counts and stores its intermediate result in the CountState object
 // which will be persisted by Flink
 class CountingFunction extends ProcessTableFunction<String> {
   public static class CountState {
     public long count = 0L;
   }

   public void eval(@StateHint CountState memory, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
     memory.count++;
     collect("Seen rows: " + memory.count);
   }
 }

 // Function that waits for a second event coming in
 class CountingFunction extends ProcessTableFunction<String> {
   public static class SeenState {
     public String first;
   }

   public void eval(@StateHint SeenState memory, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
     if (memory.first == null) {
       memory.first = input.toString();
     } else {
       collect("Event 1: " + memory.first + " and Event 2: " + input.toString());
     }
   }
 }

 // Function that uses Row for state
 class CountingFunction extends ProcessTableFunction<String> {
   public void eval(@StateHint(type = @DataTypeHint("ROW < count BIGINT >")) Row memory, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
     Long newCount = 1L;
     if (memory.getField("count") != null) {
       newCount += memory.getFieldAs("count");
     }
     memory.setField("count", newCount);
     collect("Seen rows: " + newCount);
   }
 }

Efficiency and Design Principles

A stateful function also means that data layout and data retention should be well thought through. An ever-growing state can happen by an unlimited number of partitions (i.e. an open keyspace) or even within a partition. Consider setting a StateHint.ttl() or call ProcessTableFunction.Context.clearAllState() eventually:


 // Function that waits for a second event coming in BUT with better state efficiency
 class CountingFunction extends ProcessTableFunction<String> {
   public static class SeenState {
     public String first;
   }

   public void eval(Context ctx, @StateHint(ttl = "1 day") SeenState memory, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
     if (memory.first == null) {
       memory.first = input.toString();
     } else {
       collect("Event 1: " + memory.first + " and Event 2: " + input.toString());
       ctx.clearAllState();
     }
   }
 }

Large State

Flink's state backends provide different types of state to efficiently handle large state.

Currently, PTFs support three types of state:

Value state: Represents a single value.
List state: Represents a list of values, supporting operations like appending, removing, and iterating.
Map state: Represents a map (key-value pair) for efficient lookups, modifications, and removal of individual entries.

By default, state entries in a PTF are represented as value state. This means that every state entry is fully read from the state backend when the evaluation method is called, and the value is written back to the state backend once the evaluation method finishes.

To optimize state access and avoid unnecessary (de)serialization, state entries can be declared as ListView or MapView. These provide direct views to the underlying Flink state backend.

For example, when using a MapView, accessing a value via MapView.get(Object) will only deserialize the value associated with the specified key. This allows for efficient access to individual entries without needing to load the entire map. This approach is particularly useful when the map does not fit entirely into memory.

State TTL is applied individually to each entry in a list or map, allowing for fine-grained expiration control over state elements.


 // Function that uses a map view for storing a large map for an event history per user
 class HistoryFunction extends ProcessTableFunction<String> {
   public void eval(@StateHint MapView<String, Integer> largeMemory, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
     String eventId = input.getFieldAs("eventId");
     Integer count = largeMemory.get(eventId);
     if (count == null) {
       largeMemory.put(eventId, 1);
     } else {
       if (count > 1000) {
         collect("Anomaly detected: " + eventId);
       }
       largeMemory.put(eventId, count + 1);
     }
   }
 }

Time and Timers

A PTF supports event time natively. Time-based services are available via ProcessTableFunction.Context.timeContext(Class).

Time

Every PTF takes an optional on_time argument. The on_time argument in the function call declares the time attribute column for which a watermark has been declared. When processing a table's row, this timestamp can be accessed via ProcessTableFunction.TimeContext.time() and the watermark via ProcessTableFunction.TimeContext.currentWatermark()/ProcessTableFunction.TimeContext.tableWatermark() respectively.

Specifying an on_time argument in the function call instructs the framework to return a rowtime column in the function's output for subsequent time-based operations.

The ArgumentTrait.REQUIRE_ON_TIME makes the on_time argument mandatory if necessary.

Timers

A PTF that takes set semantic tables can support timers. Timers allow for continuing the processing at a later point in time. This makes waiting, synchronization, or timeouts possible. A timer fires for the registered time when the watermark progresses the logical clock.

Timers can be named (ProcessTableFunction.TimeContext.registerOnTime(String, Object)) or unnamed (ProcessTableFunction.TimeContext.registerOnTime(Object)). The name of a timer can be useful for replacing or deleting an existing timer, or for identifying multiple timers via ProcessTableFunction.OnTimerContext.currentTimer() when they fire.

An onTimer() method must be declared next to the eval() method for reacting to timer events. The signature of the onTimer() method must contain an optional ProcessTableFunction.OnTimerContext followed by all state entries (as declared in the eval() method).

Flink takes care of storing and restoring timers during failures or restarts. Thus, timers are a special kind of state. Similarly, timers are scoped to a virtual processor defined by the PARTITION BY clause. A timer can only be registered and deleted in the current virtual processor.


 // Function that waits for a second event or timeouts after 60 seconds
 class TimerFunction extends ProcessTableFunction<String> {
   public static class SeenState {
     public String seen = null;
   }

   public void eval(Context ctx, @StateHint SeenState memory, @ArgumentHint( { SET_SEMANTIC_TABLE, REQUIRE_ON_TIME } ) Row input) {
     TimeContext<Instant> timeCtx = ctx.timeContext(Instant.class);
     if (memory.seen == null) {
       memory.seen = input.getField(0).toString();
       timeCtx.registerOnTimer("timeout", timeCtx.time().plusSeconds(60));
     } else {
       collect("Second event arrived for: " + memory.seen);
       ctx.clearAll();
     }
   }

   public void onTimer(SeenState memory) {
     collect("Timeout for: " + memory.seen);
   }
 }

Handling of Late Records

A late record is a record with a time attribute value that is less than or equal to the current watermark. PTFs handle late records just like non-late records by calling the eval() method. If the on_time argument is specified, the late timestamp is preserved in the output. This behavior is the same for PTFs with row and set semantics.

Registering a timer for a time that is less than or equal to the current watermark is allowed. If registered from within eval(), the timer fires on the next watermark advance. If registered from within onTimer(), the timer fires immediately after the current timer finishes. Note that unconditionally re-registering a past-time timer from within onTimer() causes an infinite loop.

Efficiency and Design Principles

Registering too many timers might affect performance. An ever-growing timer state can happen by an unlimited number of partitions (i.e. an open keyspace) or even within a partition. Thus, reduce the number of registered timers to a minimum and consider cleaning up timers if they are not needed anymore via ProcessTableFunction.Context.clearAllTimers() or ProcessTableFunction.TimeContext.clearTimer(String).

Ordering

A PTF that takes a table with set semantics can optionally specify an ORDER BY clause in the function call to define the order in which rows are processed within each partition. The ORDER BY clause guarantees that rows are delivered to the eval() method in the specified order.

The ORDER BY clause requires that the first column is a time attribute column (i.e., a TIMESTAMP or TIMESTAMP_LTZ column with a watermark declaration). The first ORDER BY column must be specified in ascending order. This ensures that rows are processed in event-time order. Additional columns can be specified as secondary sort keys to define the ordering of rows with the same timestamp.

ORDER BY provides strict time-based ordering at the cost of holding back records until a watermark advances the logical clock. Consequently, any late data that violates this ordering is dropped. If you need to process late data or access intermediate results, it is highly recommended to bypass ORDER BY and implement custom handling within the PTF using state and timers.

Example SQL syntax:


 SELECT * FROM my_ptf(
   input_table => TABLE source_table
     PARTITION BY user_id
     ORDER BY (event_time ASC, priority DESC NULLS FIRST)
 )

Difference Between ORDER BY and on_time Argument

While both ORDER BY and the on_time argument relate to time attributes, they serve different purposes:

on_time: Declares which time attribute column powers the time context (ProcessTableFunction.TimeContext.time()) and output timestamp. It does NOT affect the processing order of rows.
ORDER BY: Physically buffers and sorts rows within each partition to guarantee ordered delivery to the eval() method. If both ORDER BY and on_time are specified for the same table argument, they must reference the same time attribute column.

Ordering Guarantees and Late Events

When ORDER BY is specified on a time attribute column, the framework maintains a sort buffer per partition and input table to reorder out-of-order events. The sort buffer is flushed when the watermark for the given input table advances, at which point all buffered rows with timestamps less than or equal to the watermark are delivered to the eval() method in sorted order. Late events (arriving after the watermark) are dropped to maintain the ordering guarantee.

See Also:

Serialized Form

Nested Class Summary

Nested Classes

Modifier and Type

Class

Description

static interface

ProcessTableFunction.Context

Context that can be added as a first argument to the eval() method for additional information about the input tables and other services provided by the framework.

static interface

ProcessTableFunction.OnTimerContext

Special ProcessTableFunction.Context that is available when the onTimer() method is called.

static interface

ProcessTableFunction.TimeContext<TimeType>

A context that gives access to Flink's concepts of time and timers.
Constructor Summary

Constructors

Constructor

Description

ProcessTableFunction()
Method Summary

Modifier and Type

Method

Description

protected final void

collect(T row)

Emits an (implicit or explicit) output row.

final FunctionKind

getKind()

Returns the kind of function this definition describes.

TypeInference

getTypeInference(DataTypeFactory typeFactory)

Returns the logic for performing type inference of a call to this function definition.

final void

setCollector(Collector<T> collector)

Internal use.

Methods inherited from class org.apache.flink.table.functions.UserDefinedFunction
close, functionIdentifier, open, toString

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.apache.flink.table.functions.FunctionDefinition
getRequirements, isDeterministic, supportsConstantFolding

Constructor Details
- ProcessTableFunction
  
  public ProcessTableFunction()
Method Details
- setCollector
  
  public final void setCollector(Collector<T> collector)
  
  Internal use. Sets the current collector.
- collect
  
  protected final void collect(T row)
  
  Emits an (implicit or explicit) output row.
  If null is emitted as an explicit row, it will be skipped by the runtime. For implicit rows, the row's field will be null.
  
  Parameters:
  
  row - the output row
- getKind
  
  public final FunctionKind getKind()
  
  Description copied from interface: FunctionDefinition
  
  Returns the kind of function this definition describes.
- getTypeInference
  
  public TypeInference getTypeInference(DataTypeFactory typeFactory)
  
  Description copied from class: UserDefinedFunction
  
  Returns the logic for performing type inference of a call to this function definition.
  The type inference process is responsible for inferring unknown types of input arguments, validating input arguments, and producing result types. The type inference process happens independent of a function body. The output of the type inference is used to search for a corresponding runtime implementation.
  Instances of type inference can be created by using TypeInference.newBuilder().
  See BuiltInFunctionDefinitions for concrete usage examples.
  The type inference for user-defined functions is automatically extracted using reflection. It does this by analyzing implementation methods such as eval() or accumulate() and the generic parameters of a function class if present. If the reflective information is not sufficient, it can be supported and enriched with DataTypeHint and FunctionHint annotations.
  Note: Overriding this method is only recommended for advanced users. If a custom type inference is specified, it is the responsibility of the implementer to make sure that the output of the type inference process matches with the implementation method:
  The implementation method must comply with each DataType.getConversionClass() returned by the type inference. For example, if DataTypes.TIMESTAMP(3).bridgedTo(java.sql.Timestamp.class) is an expected argument type, the method must accept a call eval(java.sql.Timestamp).
  Regular Java calling semantics (including type widening and autoboxing) are applied when calling an implementation method which means that the signature can be eval(java.lang.Object).
  The runtime will take care of converting the data to the data format specified by the DataType.getConversionClass() coming from the type inference logic.
  
  Specified by:
  
  getTypeInference in interface FunctionDefinition
  
  Specified by:
  
  getTypeInference in class UserDefinedFunction

Class ProcessTableFunction<T>

Table Semantics and Virtual Processors

Table Argument with Row Semantics

Table Argument with Set Semantics

Implementation

Data Types

Arguments

Context

State

Efficiency and Design Principles

Large State

Time and Timers

Time

Timers

Handling of Late Records

Efficiency and Design Principles

Ordering

Difference Between ORDER BY and on_time Argument

Ordering Guarantees and Late Events

Nested Class Summary

Constructor Summary

Method Summary

Methods inherited from class org.apache.flink.table.functions.UserDefinedFunction

Methods inherited from class java.lang.Object

Methods inherited from interface org.apache.flink.table.functions.FunctionDefinition

Constructor Details

ProcessTableFunction

Method Details

setCollector

collect

getKind

getTypeInference