-- Support of the ORDER BY clause in PTFs
SELECT * FROM MyPtf(input => TABLE t PARTITION BY k ORDER BY ts)

ValueView.value(): T
ValueView.update(T value)
ValueView.isEmpty(): boolean
ValueView.clear()

ArgumentTrait.BROADCAST_SEMANTIC_TABLE
ArgumentTrait.NOTIFY_STATEFUL_SETS

StaticArgumentTrait.BROADCAST_SEMANTIC_TABLE
StaticArgumentTrait.NOTIFY_STATEFUL_SETS

StateHint.value(): StateKind
StateKind.PER_SET
StateKind.BROADCAST

StateTypeStrategy.isBroadcast(CallContext callContext): boolean

SELECT * FROM MyPtf(input => TABLE t PARTITION BY k ORDER BY ts)

-- Given:
CREATE TABLE t (ts TIMESTAMP(3), name STRING, score INT, WATERMARK FOR ts AS ts)

-- Input Data:
13:00, Bob, 12
13:00, Bob, 33
14:00, Bob, 50
12:00, Bob, 100

-- Function call:
SELECT * FROM IdentityPtf(input => TABLE t PARTITION BY name, on_time => ts)

-- Current Implementation Output (simplified):
-- Data is dropped
13:00, Bob, 12
14:00, Bob, 50

-- Proposed Output (simplified):
-- Just forward as you would expect it from an identity function
13:00, Bob, 12
13:00, Bob, 33
14:00, Bob, 50
12:00, Bob, 100

-- Function call:
SELECT * FROM IdentityPtf(input => TABLE t PARTITION BY name ORDER BY ts, on_time => ts)

-- Proposed Output (simplified):
-- People expect "best-effort sorting" on a stream of data,
-- thus missing data is more acceptable
13:00, Bob, 12
14:00, Bob, 50

public class GreetingWithMemory extends ProcessTableFunction<String> {
  public static class CountState {
    public long counter = 0L;
  }
  public void eval(@StateHint CountState state, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
    state.counter++;
    collect("Hello " + input.getFieldAs("name") + ", your " + state.counter + " time?");
  }
}

/**
 * A {@link DataView} that provides {@link Supplier}-like functionality in state entries.
 *
 * <p>A {@link ValueView} can be backed by a Java heap object or can leverage Flink's state
 * backends depending on the context. In many unbounded data scenarios, the {@link ValueView}
 * delegates all calls to a {@link ValueState} instead.
 *
 * @param <T> Type fo the value
 */
@PublicEvolving
public class ValueView<T> implements DataView {

    private T value;

    /**
     * Returns the current value for this view. Returns {@code null} if no value has been set.
     *
     * @return The state value corresponding to the current input.
     */
    public T value() {
        return value;
    }

    /**
     * Updates the view to the given value.
     *
     * @param value The new value for the state, or null for clearing the value.
     */
    void update(T value) {
        this.value = value;
    }

    boolean isEmpty() {
        return value == null;
    }

    @Override
    public void clear() {
        value = null;
    }
}

public class GreetingWithMemory extends ProcessTableFunction<String> {
  public void eval(@StateHint ValueView<Long> state, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
    if (input.getFieldAs("condition").equals("true")) {
      // State is not read or deserialized if not required
      return;
    }
    Long newVal = 1;
    if (!state.isEmpty()) {
      newVal = state.get() + 1;
    }
    // State is only written out and serialized when instructed to do so
    state.update(newVal);
    
    collect("Hello " + input.getFieldAs("name") + ", your " + newVal + " time?");
  }
}

/**
 * A PTF that filters data based on a dynamic allowlist.
 */
public class RulePtf extends ProcessTableFunction<String> {

    void eval(
            @StateHint(StateKind.BROADCAST) MapView<String, Double> badWords,
            @ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row data,
            @ArgumentHint(ArgumentTrait.BROADCAST_SEMANTIC_TABLE) Row rules)
            throws Exception {
        // Process broadcast input
        if (rules != null) {
            // Write access to broadcast state
            badWords.put(rules.getFieldAs("word"), rules.getFieldAs("score"));
            return;
        }
        // Process regular input
        if (data != null) {
            String sentence = data.getFieldAs("sentence");
            String[] words = sentence.toLowerCase(Locale.ROOT).split("\\s+");
            for (String w : words) {
                // Read access to broadcast state
                if (badWords.contains(w)) {
                    return;
                }
            }
            collect(sentence);
        }
    }
}

public enum ArgumentTrait {
    /**
     * An argument that accepts a table with broadcast semantics. This trait only applies to {@link
     * ProcessTableFunction} (PTF).
     *
     * <p>A broadcast table argument serves as a side input to PTFs with {@link #ROW_SEMANTIC_TABLE}
     * or {@link #SET_SEMANTIC_TABLE} table arguments. An incoming row for this table is sent to
     * each PTF executing operator, independent of the key context defined by the PARTITION BY
     * clause. If necessary, the PTF operator is able to store the broadcast information in state
     * entries of kind {@link StateKind#BROADCAST}. When processing rows of the main table(s), the
     * broadcast state entry is available for read access.
     *
     * <p>By default, a change to a broadcast state entry has no effect on previously processed
     * stateful sets of a {@link #SET_SEMANTIC_TABLE}. Use {@link #NOTIFY_STATEFUL_SETS} to
     * re-evaluate all stateful sets based on the broadcast information.
     */
    BROADCAST_SEMANTIC_TABLE(true, StaticArgumentTrait.BROADCAST_SEMANTIC_TABLE),
}

public @interface StateHint {

    /**
     * The kind of the state entry.
     *
     * <p>By default, the state entry is scoped to the set created by the PARTITION BY clause. It
     * uses Flink's keyed state.
     */
    StateKind value() default StateKind.PER_SET;
}

/** Kind of a {@link ProcessTableFunction}s state entry. */
public enum StateKind {

    /**
     * Defines a state entry scoped to the set created by the PARTITION BY clause. It uses Flink's
     * keyed state.
     *
     * <p>During processing, a PTF instance can access this state for all rows sharing the same key.
     * Both for reading and writing. However, the state cannot be accessed by other keys.
     */
    PER_SET,

    /**
     * Defines a state entry broadcast to all sets, independent of the PARTITION BY clause. It uses
     * Flink's operator state.
     *
     * <p>During processing, only a broadcasting table can write to this state. However, all PTF
     * instances can read it, regardless of their specific key context.
     */
    BROADCAST
}

public enum ArgumentTrait {
    /**
     * Notifies all existing stateful sets to re-evaluate based on the given broadcast row and the
     * potentially new state in state entries of kind {@link StateKind#BROADCAST}.
     *
     * <p>Note: This trait is only valid for {@link #BROADCAST_SEMANTIC_TABLE} arguments.
     */
    NOTIFY_STATEFUL_SETS(false, StaticArgumentTrait.NOTIFY_STATEFUL_SETS),
}

    /** A PTF that dedups data until broadcast cache invalidation. */
    public static class ControllableDedupPtf extends ProcessTableFunction<String> {

        void eval(
                @StateHint Tuple1<String> cache,
                @ArgumentHint(SET_SEMANTIC_TABLE) Row data,
                @ArgumentHint({BROADCAST_SEMANTIC_TABLE, NOTIFY_STATEFUL_SETS}) Row resetBroadcast)
                throws Exception {
            // Process regular data
            if (data != null) {
                // If cache is empty, remember it and emit
                if (cache.f0 == null) {
                    cache.f0 = data.getFieldAs("payload");
                    collect(cache.f0);
                }
                // Don't emit duplicate
                return;
            }
            // Notification of stateful sets
            else if (resetBroadcast != null && cache != null) {
                // Clean the cache for this key
                cache.f0 = null;
            }
        }
    }