public class CountWithTimeout extends ProcessTableFunction<Count> {

  // Define state as a Java object
  static class Count {
    public Long count = 0;
  }

  // Implement an eval() method with the full signature
  public void eval(Context ctx, @StateHint CountState state, @ArgumentHint(ArgumentTrait.TABLE_AS_SET) Row input) {
    // Update the state content
    state.count++;

    // Set (or replace an existing) timer
    Timers<Instant> timers = ctx.getTimersAs(Instant.class);
    timers.registerOnTime("timeout", timers.currentTimestamp().plus(Duration.ofMinutes(1)));
  }

  // Implement an onTimer() method with state signature
  public void onTimer(OnTimerContext ctx, @StateHint CountState state) {
    // Outputs a result
    collect(state);

    // Clears all state and timers
    ctx.clearAll();
  }
}

SELECT user_id, count
FROM
  CountWithTimeout(

    -- Split the input table into partitions per virtual processor
    input => TABLE(data) PARTITION BY user_id,

    -- Let the system know what "timers" and "time"  means
    on_time => DESCRIPTOR(rowtime),

    -- Support for query evolution
    uid => 'main-counting'
  )

package org.apache.flink.table.functions;

/**
 * Base class for a user-defined process table function. A user-defined process table function maps zero, one, or
 * multiple tables and scalar values to zero, one, or multiple rows (or structured types).
 *
 * <p>The behavior of a {@link TableFunction} can be defined by implementing a custom evaluation
 * method. An evaluation method must be declared publicly, not static, and named <code>eval</code>.
 * 
 * ... More explanation will follow ...
 */
public class ProcessTableFunction<T> extends UserDefinedFunction {

    /** The code generated collector used to emit rows. */
    private transient Collector<T> collector;

    /** Internal use. Sets the current collector. */
    public final void setCollector(Collector<T> collector) {
        this.collector = collector;
    }

    @Override
    public FunctionKind getKind() {
        return FunctionKind.PROCESS_TABLE;
    }

    @Override
    public TypeInference getTypeInference(DataTypeFactory typeFactory) {
        return TypeInferenceExtractor.forProcessTableFunction(typeFactory, getClass());
    }

    public void finish() throws Exception {
        // empty by default
    }

    /** Context for eval() method. */
    public interface Context {

      /** 
       * Gives access to time.
       * 
       * @param conversionClass defines the target class e.g. Instant, Long, LocalDateTime, TimestampData.
       */
      Timers<T> getTimersAs(Class<T> conversionClass);

      /** 
       * Gives access to the virtual partition of a table with set semantics.
       * 
       * @param arg name of the input argument
       */
      SetSemantics getSetSemanticsFor(String tableArgName);
      
      /**
       * Clears the state of a virtual partition. Semantically equal to setting
       * all fields of the row/structured type to null.
       */
      void clearState(String stateArgName);

      /**
       * Clears all state of a virtual partition.
       */
      void clearAllState();

      /**
       * Clears all timers of a virtual partition.
       */
      void clearAllTimers();

      /**
       * Clears the virtual partition incl. timers and state.
       */
      void clearAll();
    }

    /** Context for onTimer() methods. */
    public interface OnTimerContext extends Context {

        /** Returns the name of the currently firing timer.
         *
         * @return {@code null} if no name was used.
         */
        String currentTimer();
    }
}

/**
 * Gives access to time.
 * 
 * <p> Timers are scoped to a virtual processor (i.e. partition). The timer can only be registered
 * and deleted in the current virtual processor.
 */
interface Timers<T> {

    /**
     * Returns the current event-time defined by the {@code ON_TIME} argument descriptor.
     * If multiple tables are defined as input, it returns the timestamp of the currently
     * processed table.
     *
     * @return {@code null} if no time attribute is present.
     */
     T currentTime();

    /**
     * Returns the current event-time watermark.
     *
     * @return {@code null} if no watermark has been received yet.
     */
    T currentWatermark();
    
   /**
     * Returns the next event-time watermark in case of {@link #registerOnWatermark}.
     * Allows to flushing buffers shortly before the new watermark arrives.
     *
     * @return {@code null} if no watermark has been received yet.
     */
    T followingWatermark();

    /**
     * Registers a timer to be fired when the event time watermark passes the given time.
     * Replaces an existing timer under the same name.
     */
    void registerOnTime(String name, T time);

    /**
     * Registers a timer to be fired when the event time watermark passes the given time.
     * Has no effect if an existing timer exists under the same time.
     */
    void registerOnTime(T time);

    /**
     * Registers a periodic timer to be fired whenever the event time watermark passes the given time.
     * Replaces an existing timer under the same name.
     */
    void registerOnPeriodicTime(String name, T relativeTime, Duration period);

    /**
     * Registers a timer to be fired before a watermark passes.
     * Timer will get the name "watermark".
     */
    void registerOnWatermark();

    /**
     * Registers a periodic timer to be fired every time before a watermark passes.
     * Timer will get the name "watermark".
     */
    void registerOnPeriodicWatermark();

    /**
     * Deletes an event-time timer. This method has only an effect if such a timer was previously
     * registered and did not already expire.
     */
    void deleteTimer(String name);

    /**
     * Deletes an event-time timer without name. This method has only an effect if such a timer was
     * previously registered and did not already expire.
     */
    void deleteTimer(T time);

}

/**
 * Gives access to set semantics of a table argument.
 */
class interface SetSemantics {

  /**
   * An array of index paths that specifies the passed PARTITION BY columns.
   */
  int[][] partitionByColumns();
  
  /**
   * An array of index paths that specifies the passed ORDER BY columns.
   */
  int[][] orderByColumns();
  
  /**
   * Position of the passed "ON_TIME" column.
   */
  int timeColumn();
  
  /**
   * List of table argument names that are copartitioned with this table argument.
   */
  List<String> coPartitionArgs();
}

package org.apache.flink.table.types.inference;

public final class TypeInference {

  /**
   * Returns a list of state declarations.
   *
   * <p>For example, an accumulator for {@link AggregateFunction} or 
   * one or more intermediates for {@link ProcessTableFunction}.
   */
  public LinkedHashMap<String, TypeStrategy> getStateTypeStrategies();
  
  /**
   * Returns a list of static, non-overloaded, non-vararg arguments.
   */
  public Optional<List<StaticArgument>> getStaticArguments();
  
  /**
   * Argument that is part of static, not overloaded, not vararg signature.
   */
  static class StaticArgument {

      /**
       * Declares a scalar argument such as f(12) or f(otherField).
       *
       * @param name whether the assignment operator can be used e.g. f(myArg => 12)
       * @param type explicit type to which the argument is cast
       * @param isOptional whether the argument is optional and might be filled 
       *                   with a default value
       */
      static StaticArgument scalar(String name, DataType type, boolean isOptional);
      
      /**
       * Declares a table argument such as f(myTable) or f(TABLE myTable).
       * Uses {@link StaticArgumentTraits#TABLE_AS_ROW} by default.
       *
       * <p>If no data type is defined, it operates in a "Polymorphic" behavior
       * accepting all kind of tables. But requires a row class (Row or RowData).
       *
       * @param traits allows to specify further characteristics such
       *.       as {@link StaticArgumentTraits#TABLE_AS_ROW}
       *        or {@link StaticArgumentTraits#TABLE_AS_SET}
       */
      static StaticArgument table(
        String name,
        Class<?> conversionClass,
        boolean isOptional,
        EnumSet<ArgTrait> traits);
        
      /**
       * Declares a table argument such as f(myTable) or f(TABLE myTable).
       * Uses {@link StaticArgumentTraits#TABLE_AS_ROW} by default.
       *
       * @param type data structure how the table is represented, must be a
       *        structured or row type. If a data type is set, the function looses the
       *         "Polymorphic" behavior and expects a certain table structure.
       * @param traits allows to specify further characteristics such
       *.       as {@link StaticArgumentTraits#TABLE_AS_ROW}
       *        or {@link StaticArgumentTraits#TABLE_AS_SET}
       */
      static StaticArgument table(
        String name,
        DataType type,
        boolean isOptional,
        EnumSet<StaticArgumentTraits> traits);

      /**
       * Declares a model argument such as f(myModel).
       * Only intended for built-in functions.
       */
      @Internal
      static StaticArgument model(
        String name,
        boolean isOptional,
        EnumSet<StaticArgumentTraits> traits);

      public DataType getDataType(DataType input);

      public boolean isOptional();

      public String getName();

      public Set<StaticArgumentTraits> getTraits();
    }
    
    /**
     * List of characteristics an argument can declare.
     * Kept generic for future extensions.
     */
    enum StaticArgumentTraits {
      SCALAR(null),
      TABLE_AS_ROW(null),
      TABLE_AS_SET(null),
      MODEL(null),
      OPTIONAL_PARTITION_BY(TABLE_AS_SET),
      SUPPORTS_NESTED_PARTITION_BY(TABLE_AS_SET),
      REQUIRES_ORDER_BY(TABLE_AS_SET),
      SUPPORTS_NESTED_ORDER_BY(TABLE_AS_SET),
      OPTIONAL_COPARTITION(TABLE_AS_SET, OPTIONAL_PARTITION_BY),
      REQUIRES_TIME_ATTRIBUTE(TABLE_AS_SET);
      
      private final Set<ArgumentTrait> requires;
      
      ArgumentTrait(ArgumentTrait... requires) {
        this.requires = Set.of(requires);
      }
    }
    
    /** Builder for configuring and creating instances of {@link TypeInference}. */
    public static class Builder {
    
      public Builder staticArguments(StaticArgument... staticArgument);

      public Builder staticArguments(List<StaticArgument> staticArgument);

      public Builder staticArgument(StaticArgument staticArgument);

      public Builder stateTypeStrategy(String name, TypeStrategy strategy);

      public Builder stateTypeStrategies(
        LinkedHashMap<String, TypeStrategy> stateTypeStrategies);
    }
}

CallContext {
     /** 
       * Gives access to the virtual partition of a table with set semantics.
       * 
       * @param arg name of the input argument
       */
      SetSemantics getSetSemantics(int pos); 
}

/**
 * Definition of a function. Instances of this class provide all details necessary to validate a
 * function call and perform planning.
 */
public interface FunctionDefinition {
  
   /**
     * Returns the set of changes that the function accepts during runtime.
     *
     * Note: This method only applies to {@link FunctionKind.PROCESS_TABLE},
     * {@link FunctionKind.AGGREGATE}, and {@link FunctionKind.TABLE_AGGREGATE}.
     */
  default ChangelogMode getInputChangelogMode() {
    return ChangelogMode.insertOnly();
  }
  
   /**
     * Returns the set of changes that the function produces during runtime.
     *
     * Note: This method only applies to {@link FunctionKind.PROCESS_TABLE},
     * {@link FunctionKind.AGGREGATE}, and {@link FunctionKind.TABLE_AGGREGATE}.
     */
  default ChangelogMode getOutputChangelogMode() {
    return ChangelogMode.insertOnly();
  }
}

package org.apache.flink.table.types.logical;

public enum LogicalTypeRoot {
  DESCRIPTOR(LogicalTypeFamily.PREDEFINED),
}


package org.apache.flink.table.types.logical;

public final class DescriptorType extends LogicalType {

    private static final String FORMAT = "DESCRIPTOR";

    public DescriptorType(boolean isNullable) {
        super(isNullable, LogicalTypeRoot.DESCRIPTOR);
    }

    public DescriptorType() {
        this(true);
    }

    @Override
    public String asSummaryString() {
        return withNullability(FORMAT);
    }

    @Override
    public String asSerializableString() {
        return withNullability(FORMAT);
    }

    @Override
    public boolean supportsInputConversion(Class<?> clazz) {
        return ColumnList.class.isAssignableFrom(clazz);
    }

    @Override
    public boolean supportsOutputConversion(Class<?> clazz) {
        return ColumnList.class.isAssignableFrom(clazz);
    }

    @Override
    public Class<?> getDefaultConversion() {
        return ColumnList.class;
    }
}



package org.apache.flink.table.api;

public final class DataTypes {

  /**
   * Data type for passing a list of columns to function calls.
   */
  public static DataType DESCRIPTOR();
}

/**
 * A hint that marks argument as a managed state.
 */
public @interface StateHint {

  /**
   * The name of the state.
   *
   * <p>This can be used to provide a descriptive name for the state.
   * And reference it e.g. in {@code ctx.clearState}.
   */
  String name() default "";

  /**
   * The data type hint for the argument.
   *
   * <p>This can be used to provide additional information about the expected data type of the
   * argument. The {@link DataTypeHint} annotation can be used to specify the data type explicitly
   * or provide hints for the reflection-based extraction of the data type.
   */
  DataTypeHint type() default @DataTypeHint();

  // we can add state TTL parameters here as well
}

/**
 * A hint that provides additional information about an argument.
 *
 */
public @interface ArgumentHint {

    /**
     * Traits that describe the argument. A scalar argument by default.
     */
    ArgumentTrait[] value() default {ArgumentTrait.SCALAR};

    /**
     * The name of the argument.
     *
     * <p>This can be used to provide a descriptive name for the argument.
     */
    String name() default "";

    /**
     * Specifies whether the argument is optional or required.
     *
     * <p>If set to {@code true}, the argument is considered optional.And if the user does not
     * specify this parameter when calling, 'null' will be passed in. By default, an argument is
     * considered required.
     */
    boolean isOptional() default false;

    /**
     * The data type hint for the argument.
     *
     * <p>This can be used to provide additional information about the expected data type of the
     * argument. The {@link DataTypeHint} annotation can be used to specify the data type explicitly
     * or provide hints for the reflection-based extraction of the data type.
     */
    DataTypeHint type() default @DataTypeHint();
}

package org.apache.flink.table.annotation;

/**
 * Traits that specify the usage of an argument.
 * Parameters for ArgumentHint.
 */
public enum ArgumentTrait {
    SCALAR(null),
    TABLE_AS_SET(null),
    TABLE_AS_ROW(null),
    OPTIONAL_PARTITION_BY(TABLE_AS_SET),
    SUPPORTS_NESTED_PARTITION_BY(TABLE_AS_SET),
    REQUIRES_ORDER_BY(TABLE_AS_SET),
    SUPPORTS_NESTED_ORDER_BY(TABLE_AS_SET),
    OPTIONAL_COPARTITION(OPTIONAL_PARTITION_BY),
    REQUIRES_TIME_ATTRIBUTE(TABLE_AS_SET);

    private final Set<ArgumentTrait> requires;

    ArgumentTrait(ArgumentTrait... requires) {
        this.requires = Set.of(requires);
    }
}

public class OneMatchJoin extends ProcessFunction<JoinResult> {

  public class JoinState {
    public boolean matched = false;
    public Customer customer;
    public Order order;
  }

  public class JoinResult {
    public Customer customer;
    public Order order;
  }

  public class Customer {
    public String name;
    public String address;
  }

  public class Order {
    public Long orderId;
    public Double price;
  }

  public void eval(
    Context ctx,
    @StateHint JoinState state,
    @ArgumentHint(TABLE_AS_SET) Customer customer,
    @ArgumentHint(TABLE_AS_SET) Order order) {

    if (state.matched) {
      return;
    }

    // Process left input
    if (customer != null) {
      if (state.order != null) {
        matchFound(state, customer, state.order);
      } else {
        state.customer = customer;
      }
    }
    // Process right input
    if (order != null) {
      if (state.customer != null) {
        matchFound(state, state.customer, order);
      } else {
        state.order = order;
      }
    }
  }

  private void matchFound(JoinState state, Customer customer, Order order) {
    state.matched = true;
    state.customer = null;
    state.order = null;

    JoinResult result = new JoinResult();
    result.customer = customer;
    result.order = order;
    collect(result);
  }
}

SELECT * FROM
FROM
  OneMatchJoin(
    customer => TABLE(data1) PARTITION BY customer_id,
    order => TABLE(data2) PARTITION BY customer_id COPARTITION(data1, data2)
  )

public class BuiltInOneMatchJoin extends ProcessFunction<Row> {

  public TypeInference getTypeInference() {
    return TypeInference.newBuilder()
      .staticArguments(         StaticArgument.table(
          "left", Row.class, false,
          TABLE_AS_SET, REQUIRES_TIME_ATTRIBUTE),
        StaticArgument.table(
          "right", Row.class, false,
          TABLE_AS_SET, REQUIRES_TIME_ATTRIBUTE),
        StaticArgument.scalar(
          "ttl", DataTypes.INTERVAL(SECOND(3)), false))
      .stateTypeStrategy("state", callContext -> {
        return Optional.of(
          DataTypes.ROW(
            FIELD("matched", DataTypes.BOOLEAN())
            FIELD("left", callContext.getArgumentDataTypes(0)),
            FIELD("right", callContext.getArgumentDataTypes(1))
        );
      })
      .outputTypeStrategy(callContext -> {
        return Optional.of(
          return Optional.of(
            DataTypes.ROW(
              FIELD("left", callContext.getArgumentDataTypes(0)),
              FIELD("right", callContext.getArgumentDataTypes(1))
            );
        );
      });
  }

  
  public void eval(Context ctx, Row state, Row left, Row right, Duration ttl) {
    Timers<Instant> timers = ctx.getTimersAs(Instant.class);
    timers.registerOnTime("timeout", timers.currentTimestamp().plus(ttl));

    if (Boolean.TRUE.equals(state.getField("matched"))) {
      return;
    }

    // Process left input
    if (left != null) {
      if (state.getField("right") != null) {
        matchFound(state, left, state.getField("right"));
      } else {
        state.setField("left", left);
      }
    }
    // Process right input
    if (right != null) {
      if (state.getField("left") != null) {
        matchFound(state, state.getField("left"), right);
      } else {
        state.setField("right", right);
    }
  }

  public void onTime(Context ctx, Row state) {
    ctx.clearAll();
  }

  private void matchFound(Row state, Row left, Row right) {
    state.setField("matched", true);
    state.setField("left", null);
    state.setField("right", null);

    collect(Row.of(left, right));
  }
}

SELECT * FROM
FROM
  BuiltInOneMatchJoin(
    left => TABLE(customers) PARTITION BY customer_id,
    right => TABLE(orders) PARTITION BY customer_id COPARTITION(customers, orders),
    ttl => INTERVAL '5' MINUTES,
    on_time => DESCRIPTOR(c_rowtime, o_rowtime) 
  )

@DataTypeHint("ROW<s STRING>")
public class TimedLastValue extends ProcessFunction<Row> {

  @Override
  public ChangelogMode getOutputChangelogMode() {
    return ChangelogMode.all();
  }

  public static class LastValue {
    public Instant time;
    public String value;
  }

  public void eval(
      Context ctx,
      @StateHint LastValue state,
      @ArgumentHint(TABLE_AS_SET, REQUIRES_TIME_ATTRIBUTE) Row input) {
    Timers<Instant> timers = ctx.getTimersAs(Instant.class);
    Instant inputTime = timers.currentTimestamp();
    String inputValue = input.getFieldAs("value");
    // State is empty
    if (state.time == null) {
      state.time = inputTime;
      state.value = inputValue;
      collect(Row.ofKind(INSERT, inputValue));
    }
    // State is not empty
    else {
      if (inputTime.isAfter(state.time)) {
        collect(Row.ofKind(UPDATE_BEFORE, state.value));
        collect(Row.ofKind(UPDATE_AFTER, inputValue));
        state.time = inputTime;
        state.value = inputValue;
      }
    }
  }
}

SELECT customer_id, s FROM
FROM
  TimedLastValue(
    input => TABLE(customers) PARTITION BY customer_id,
    on_time => DESCRIPTOR(c_rowtime) 
  )

public class BuiltInChangelog extends ProcessFunction<Row> {
  
  @Override
  public TypeInference getTypeInference() {
    return TypeInference.newBuilder()
      .staticArgument(StaticArgument.table("input", Row.class, false))
      .outputStrategy(callContext -> {
        List<DataTypes.Field> fields = Stream
          .concat(
            Stream.of(DataTypes.STRING()),
            DataType.getFields(callContext.getArgumentTypes(0)).stream()))
          .collect(toList())
        return Optional.of(ROW(fields));
      })
      .build();
  }

  @Override
  public ChangelogMode getInputChangelogMode() {
    return ChangelogMode.all();
  }

  public void eval(Row input) {
    Row flag = Row.of(input.getKind().shortString());
    collect(Row.join(flag, input));
  }
}

SELECT f0 AS changeflag, c1, c2 FROM BuiltInChangelog(customers);

EXECUTE STATEMENT SET BEGIN
   INSERT INTO main SELECT a, b, target_table FROM FunctionWithSideOutput(input => data, uid = 'only_once') WHERE target_table = 'main'; 
   INSERT INTO side SELECT a, b, target_table FROM FunctionWithSideOutput(input => data, uid = 'only_once') WHERE target_table = 'side'; 
END;

SELECT * FROM
FROM
  BuiltInOneMatchJoin(
    left => TABLE(customers) PARTITION BY customer_id,
    right => TABLE(orders) PARTITION BY customer_id COPARTITION(customers, orders),
    ttl => INTERVAL '5' MINUTES,
    on_time => DESCRIPTOR(c_rowtime, o_rowtime) 
  )

// PTF call without named args
env
    .from(
      call(
        BuiltInOneMatchJoin.class,
        orders.partitionBy($("customer_id")).asArgument("orderById"),
        payments.partitionBy($("order_id")).copartitionWith("orderById"),
        Duration.ofHours(2)
      )
    )
    .execute()
    .print();

// PTF call with named args
env
    .from(
      call(
        BuiltInOneMatchJoin.class,
        customers
          .partitionBy($("customer_id"))
          .asArgument("left"),
        orders
          .partitionBy($("customer_id"))
          .copartitionWith("left")
          .asArgument("orders"),
        lit(Duration.ofMinutes(5)))
          .asArgument("ttl")
        descriptor("c_rowtime", "o_rowtime")
          .asArgument("on_time")
      )
    )
    .execute()
    .print();

SELECT user_id, count
FROM
  CountWithTimeout(

    -- Split the input table into partitions per virtual processor
    input => TABLE(data) PARTITION BY user_id,

    -- Let the system know what "timers" and "time"  means
    on_time => DESCRIPTOR(rowtime),

    -- Support for query evolution
    uid => 'main-counting'
  )

env
  .from("orders")
  .partitionBy($("user_id"))
  // Same signatures as call() for avoiding one level of nesting expressions.
  // The table will be passed as the first argument of the PTF.
  .process(
    BuiltInOneMatchJoin.class,
    descriptor("rowtime").asArgument("on_time"),
    lit("main-counting").asArgument("uid"))