User-defined process table functions (UDPTFs)

Table arguments with row and set semantics

A PTF with the row-semantics table argument ArgumentTrait.ROW_SEMANTIC_TABLE assumes that rows are unrelated, so each row can be processed independently. The framework freely decides which PTF instance handles the current row.

A PTF with the set-semantics table argument ArgumentTrait.SET_SEMANTIC_TABLE assumes that rows are related. The PARTITION BY clause defines the columns used for the relation. The framework shuffles data by partition key to ensure that all rows with the same partition key are processed by the same parallel instance. You can also declare the argument ArgumentTrait.OPTIONAL_PARTITION_BY to omit the partition key, in which case a single PTF instance processes all data.

For example:

// A UDPTF with row semantics
// Used as SELECT * FROM f(r => TABLE t)
public void eval(
  @ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row input) {
  ...
}

// A UDPTF with set semantics
// Used as SELECT * FROM f(r => TABLE t PARTITION BY name)
//        or SELECT * FROM f(r => TABLE t)
public void eval(
  @ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row input) {
  ...
}

For more information, see table semantics and virtual processors.

State access

A PTF with set semantics can be stateful. The framework stores and restores the state across failures and restarts. State is partitioned by the partition key defined in the PARTITION BY clause. If no partition key is defined, all data forms a single partition (that is, keyed state in Flink).

State is declared in a PTF with StateHint, where you can explicitly specify parameters such as the state data type and time to live (TTL). For example:

class StateFunction extends ProcessTableFunction<String> {
  /** POJO for state. All fields nullable. */
  public static class Score {
    public String s;
    public Integer i;
  }

  public void eval(
    Context ctx,
    // Explicitly declare the data type of this state
    @StateHint(type = @DataTypeHint("ROW<emitted BOOLEAN>")) Row s0,
    // Set the TTL of this state to 5 days
    @StateHint(ttl = "5 days") Score s1,
    // Set the TTL of this state to 0 ms (never expires). A value without a unit is in ms
    @StateHint(ttl = "0") Score s2,
    // The TTL of this state follows table.exec.state.ttl
    @StateHint Score s3,
    @ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row r) {
    ...
    // Clean up state with ctx.clearState("s1") or ctx.clearAllState()
    // to prevent state bloat
  }
}

Currently, the following three state types are available in a PTF:

• Value State: a single value. This is the default state type in a PTF.

• List State: a list of values that supports appends, removals, and iteration.

• Map State: a mapping of key-value pairs for efficient lookup, modification, and removal of entries.

By default, state in a PTF is Value State: every access reads and writes the entire data from and to the state backend. To optimize state access and avoid unnecessary serialization and deserialization, the state object can be declared as:

• org.apache.flink.table.api.dataview.ListView (that is, List State)

• org.apache.flink.table.api.dataview.MapView (that is, Map State)

For more information, see state.

Time and timers

PTFs natively support event time. You can use event time and timers in a PTF to implement business logic such as delayed triggering and late data handling.

Time

Each PTF call has an optional on_time argument that declares the time attribute column used as event time. You can also declare the argument ArgumentTrait.REQUIRE_ON_TIME to make the on_time column mandatory in every call.

SELECT * FROM f(..., on_time => DESCRIPTOR(`my_timestamp`));

You can access time for custom operations by calling Context#timeContext(Class) to obtain a TimeContext. The scope of the time context is always limited to the event being processed, which can be the current input row or the timer being fired. Time and timer timestamps can be represented as java.time.Instant, java.time.LocalDateTime, or Long. These timestamps are based on milliseconds since the epoch and ignore the local session time zone. You can pass a time class to timeContext().

/** Sends a ping for each key one minute after the last observed event of the key. */
public static class PingLaterFunction
  extends ProcessTableFunction<String> {
    public void eval(
      Context ctx,
      @ArgumentHint({ArgumentTrait.SET_SEMANTIC_TABLE, 
                     ArgumentTrait.REQUIRE_ON_TIME}) 
      Row input
    ) {
      TimeContext<Instant> timeCtx = ctx.timeContext(Instant.class);
      // Replace the existing timer, which may restart the one-minute countdown
      timeCtx.registerOnTime("ping", timeCtx.time().plus(Duration.ofMinutes(1)));
    }

    public void onTimer(OnTimerContext onTimerCtx) {
      collect("ping");
    }
  }

TimeContext provides functions to get the event time of the record being processed, the timestamp that fired a timer, the watermark of the current input, and the watermark of the current PTF instance. For more information, see time.

Timers

PTFs with set semantics support timers. A timer lets you process data at a later point in time to implement waiting, synchronization, or timeout logic. When the watermark passes the registered time, the timer fires.

A PTF that uses timers must declare an onTimer() method to handle timer events. The signature of onTimer() is as follows:

onTimer( <on timer context>? , <state entry>* )

The framework stores and restores timers across failures and restarts, so timers are a special kind of state. Likewise, a timer is scoped to the state partition defined by the PARTITION BY clause and can be registered and removed only within the current partition key. If the number of timers affects performance, clean up unneeded timers with Context#clearAllTimers() or TimeContext#clearTimer(String). For more information, see timers.

Sorting

PTFs with set semantics can optionally specify an ORDER BY clause in a call to define the processing order of rows within each partition. The ORDER BY clause guarantees that rows are passed to the eval() method in the specified order.

The first column of the ORDER BY clause must be a time attribute column (a TIMESTAMP or TIMESTAMP_LTZ column with a declared watermark), and the first ORDER BY column must be in ascending order to guarantee event-time processing. You can add more columns as secondary sort keys to order rows that share the same timestamp. For more information, see ordering.

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

Multi-table processing

A PTF can process multiple tables at the same time. The eval() method can take multiple table arguments to support multiple inputs. All table arguments must be declared with set semantics and use consistent partitioning: the number of columns and their data types in the PARTITION BY clauses of all related table arguments must be identical.

Rows from the inputs are passed to the function one row at a time, so only one table argument is non-null at any moment. Use null checks to determine which input is being processed.

The following example implements a custom join:

/** Greets a customer and recommends the last purchased item when the information is available. */
public static class GreetingWithLastPurchase extends ProcessTableFunction<String> {

  /** Keeps the most recently purchased item in state. */
  public static class LastItemState {
    public String lastItem;
  }

  /** The eval() method takes two @ArgumentHint(SET_SEMANTIC_TABLE) arguments */
  public void eval(
    @StateHint LastItemState state,
    @ArgumentHint(SET_SEMANTIC_TABLE) Row visit,
    @ArgumentHint(SET_SEMANTIC_TABLE) Row purchase) {

    // Handles rows from the Purchases table
    if (purchase != null) {
      state.lastItem = purchase.getFieldAs("item");
    }

      // Handles rows from the Visits table
    else if (visit != null) {
      if (state.lastItem == null) {
        collect("Hello " + visit.getFieldAs("name") + ", let me know if I can help!");
      } else {
        collect("Hello " + visit.getFieldAs("name") + ", here to buy " + state.lastItem + " again?");
      }
    }
  }
}

For more information, see multiple tables.

Pass-through columns

Depending on the table semantics and whether the on_time argument is defined, the framework adds extra columns to the output of the PTF.

For PTFs with set semantics, the output is prefixed with the PARTITION BY columns. For calls with the on_time argument, the output is suffixed with <rowtime>. The default output format is therefore:

<PARTITION BY keys> | <function output> | <rowtime>

You can declare the ArgumentTrait.PASS_COLUMNS_THROUGH argument to make the PTF output all columns of the input table. For example, given an input table t with columns k and v, and a PTF f() that produces columns c1 and c2, the query SELECT * FROM f(table_arg => TABLE t PARTITION BY k) has the following output schema:

Default: 
    | k | c1 | c2 |
With ArgumentTrait.PASS_COLUMNS_THROUGH: 
    | k | v | c1 | c2 |

For more information, see pass-through columns.

Updates and changelogs

By default, a PTF can consume and produce only append-only streams. The following arguments and interfaces enable a PTF to process update streams:

The following example consumes an update stream:

@DataTypeHint("ROW<flag STRING, sum INT>")
public static class ToChangelogFunction extends ProcessTableFunction<Row> {
  public void eval(@ArgumentHint({SET_SEMANTIC_TABLE, SUPPORT_UPDATES}) Row input) {
    // Forwards the sum column and includes the row kind as a string column.
    Row changelogRow =
      Row.of(
        input.getKind().toString(),
        input.getField("sum"));

    collect(changelogRow);
  }
}

The following example produces an update stream:

@DataTypeHint("ROW<sum INT>")
public static class CustomAggregation
  extends ProcessTableFunction<Row>
  implements ChangelogFunction {

    @Override
    public ChangelogMode getChangelogMode(ChangelogContext changelogContext) {
      // Tells the framework that the PTF produces a retract stream
      return ChangelogMode.all();
    }

    public static class Accumulator {
      public Integer sum = 0;
    }

    public void eval(
      @StateHint Accumulator state,
      @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
      int score = input.getFieldAs("score");

      // A negative state value indicates that the partition
      // key was previously marked as invalid
      if (state.sum == -1) {
        return;
      }

      // A negative score invalidates the entire aggregation result.
      if (score < 0) {
        // Emits -D for the affected partition key and
        // marks the state as invalid. All subsequent
        // operations and the sink remove the aggregation result.
        collect(Row.ofKind(RowKind.DELETE, state.sum));
        state.sum = -1;
      } else {
        if (state.sum == 0) {
          // Emits +I for the first valid aggregation result.
          state.sum += score;
          collect(Row.ofKind(RowKind.INSERT, state.sum));
        } else {
          // Emits -U (old aggregation result) and +U (new aggregation result)
          // to encode an update.
          collect(Row.ofKind(RowKind.UPDATE_BEFORE, state.sum));
          state.sum += score;
          collect(Row.ofKind(RowKind.UPDATE_AFTER, state.sum));
        }
      }
    }
  }

For more information, see updates and changelogs.