1. Motivation

Deduplication is one of the most common transformations users ask Flink SQL to perform. Today the only way to express keep-first deduplication is a ROW_NUMBER() over-window filtered to the first row:

SELECT [column_list]
FROM table_name
QUALIFY ROW_NUMBER() OVER (
  [PARTITION BY col1[, col2...]] ORDER BY time_attr [ASC|DESC]
) = 1;

As documented in the Flink deduplication query documentation, ASC keeps the first row and DESC keeps the last row. This pattern has the problem:

• It is verbose and easy to get wrong. The pattern must be written exactly for the planner to recognize it as deduplication, and the intent is buried in a windowed rank.

DEDUPLICATE_KEEP_FIRST is a built-in Process Table Function (PTF) that takes an insert-only or updating input and always produces an insert-only output, keeping the first record per key. It supports two ordering modes:

• a watermarkless mode that keeps the first record observed for a key, requiring no watermark or event-time attribute; and
• an event-time mode that deterministically keeps the record with the earliest event time and emits it once a watermark makes the choice final.

The function covers several use cases, for example:

• Keeping the first record seen per key without defining watermarks.
• Deterministically keeping the earliest record per key by event time.

2. Background

This design builds on existing PTF-framework behavior.

2.1 Set semantics and partitioning

This function uses set semantics: rows are correlated by the PARTITION BY key and per-key state may be kept, with all rows for a key routed to the same processor so the function scales out by key. Without PARTITION BY, it runs a single instance (parallelism 1) over the whole input.

2.2 on_time and watermarks

on_time is a standard PTF time argument. When supplied with a rowtime descriptor, the framework provides the event-time attribute and watermark progress that let the function order records and decide when a choice is final. When omitted, the function has no time concept and keeps the first record observed for a key (arrival order).

2.3 State TTL

State TTL in Flink today is processing-time based and configured through table.exec.state.ttl. It defines a minimum idle-retention period; cleanup may occur sometime after that period. There is no event-time TTL in the engine today.

3. Public Interfaces

A new built-in SQL function DEDUPLICATE_KEEP_FIRST, implemented as a set-semantic Process Table Function. Its input may be insert-only or updating; its output is insert-only in every case. It is also available through the Table API PTF call surface.

Complete signature:

DEDUPLICATE_KEEP_FIRST(
    input                  => TABLE(t) [PARTITION BY <cols>],   -- required table argument
    on_time                => DESCRIPTOR(<rowtime>),            -- optional
    state_ttl              => <interval>,                        -- optional
    reset_ttl_on_duplicate => <boolean>,                         -- optional
    uid                    => <string>                           -- optional (implicit system argument)
)

Parameters:

• input (required) — the source table (insert-only or updating), optionally partitioned with PARTITION BY.
• on_time (optional) — a descriptor for a rowtime attribute. Enables deterministic event-time ordering, when omitted the function uses watermarkless first-arrival ordering. Requires insert-only input — supplying on_time with an updating input is a validation error (enforced by the framework), so updating inputs run in watermarkless mode only.
• state_ttl (optional) — a processing-time retention interval for the deduplication state. If not specified, falls back to table.exec.state.ttl, which retains state indefinitely at its default of 0 (see Section 4.1.2.3).
• reset_ttl_on_duplicate (optional) — whether a later duplicate refreshes the key's state_ttl; defaults to true, so retention tracks the most recent occurrence of a key (see Section 4.1.2.4).

The signature also carries the framework's implicit uid argument (auto-appended to every PTF; see Section 5). The parameters are detailed in Section 4.1.2.

4. Proposed Changes

4.1 DEDUPLICATE_KEEP_FIRST PTF

4.1.1 Default signature

The simplest invocation has no partition key and no ordering column. All rows form a single global partition, so the function emits only the first record it observes, once. With the default table.exec.state.ttl of 0, state is retained indefinitely:

SELECT * FROM DEDUPLICATE_KEEP_FIRST(input => TABLE(user_events))

Input
-- +I[id: 1, name: 'Vas']
-- +I[id: 1, name: 'Bob']

Output
-- +I[id: 1, name: 'Vas']

All arguments other than input are optional; each one added changes the behavior as described in Section 4.1.2.

4.1.2 Parameters

4.1.2.1 Partitioning (PARTITION BY)

• PARTITION BY: (Optional) Distributes processing by key so all records for a key are routed to the same instance and the function scales out.
  • Default: omitted — all rows form a single global partition, the function runs at parallelism 1, and at most one row is ever emitted.     - One or more columns are allowed. Listing all payload columns makes exact-duplicate deduplication scalable (see Section 4.1.3.4).
  • NULL is valid in PARTITION BY columns — following GROUP BY semantics, all rows whose partition key is NULL form a single partition with its own state (NULLs are treated as equal); they are neither rejected nor dropped.

4.1.2.2 Ordering and determinism (on_time)

• on_time: (Optional) A DESCRIPTOR for a rowtime attribute that turns on deterministic event-time selection. The deduplication key is always the PARTITION BY columns (or a single global partition when omitted); on_time does not change the key — it only changes which record is kept among rows with the same key and when it is emitted.
  • Default: omitted — watermarkless mode. The first record observed for a key is emitted; later records for that key are dropped while its state exists. The result does not depend on watermarks or an event-time attribute. This is non-deterministic when distinct rows share a key (first arrival wins), which is acceptable because the output stays append-only and, for exact-duplicate payloads, the choice is unobservable.
  • When supplied — event-time mode. Among rows with the same key, the one with the smallest event-time value is retained and emitted once the watermark passes that timestamp so the choice is final. A record arriving later but carrying an earlier event time replaces the candidate before finalization; records that can no longer become the earliest are dropped.
  • Event-time mode requires insert-only input; combining on_time with an updating input is rejected at planning time, because a watermark asserts no earlier record will arrive while an update can invalidate that at any time.

4.1.2.2.1 Example: watermark-gated earliest selection

SELECT * FROM DEDUPLICATE_KEEP_FIRST(
    input   => TABLE(user_events) PARTITION BY user_id,
    on_time => DESCRIPTOR(rowtime)
)

-- source table declares: WATERMARK FOR rowtime AS rowtime - INTERVAL '5' MINUTES
Input (@ = processing-time arrival; rowtime = event time; watermark = max(rowtime) - 5 min)
-- +I[user_id: 'u1', event: 'click',  rowtime: 10:02]   @ 12:00   watermark → 09:57
-- +I[user_id: 'u1', event: 'view',   rowtime: 10:01]   @ 12:01   on time (>= 09:57); earlier event time → candidate
-- +I[user_id: 'u1', event: 'scroll', rowtime: 10:06]   @ 12:02   watermark → 10:01; later, not the candidate

Output
-- +I[user_id: 'u1', event: 'view', rowtime: 10:01]     @ 12:02   (watermark reached 10:01 → candidate finalized and emitted)

4.1.2.3 State retention (state_ttl)

• state_ttl: (Optional) Processing-time retention for the per-key deduplication state.
  • Default: if unset, the function uses the value of table.exec.state.ttl, as other PTFs do. With that option at its default of 0, state is retained indefinitely. An explicit state_ttl overrides it (e.g. INTERVAL '1' DAY).
  • After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. If the effective TTL is 0 (state retained indefinitely), a key is emitted exactly once and never re-emitted — the first record wins for its whole lifetime.
  • Set TTLs larger than the source's out-of-orderness bound. The state must still be present when a key's later duplicates and the watermark that finalizes them arrive. A TTL shorter than the source's bounded out-of-orderness (the lag between event time and the watermark), whether set through state_ttl or inherited from table.exec.state.ttl, can evict a key's state too early, so a later duplicate is treated as a new key and emitted again — silently breaking deduplication.
  • Orthogonal to the other parameters — combines with any use case in Section 4.1.3.

4.1.2.4 Duplicate TTL refresh (reset_ttl_on_duplicate)

• reset_ttl_on_duplicate: (Optional) Whether a dropped duplicate refreshes the key's state_ttl.
  • Default: true — retention tracks the most recent occurrence of a key, the position taken in the design discussion. See Section 9.     - Only meaningful when a TTL applies, whether set through state_ttl or inherited from table.exec.state.ttl. true keeps a continuously-duplicated key retained longer; false measures retention from the first record for the key regardless of later duplicates.

4.1.2.5 Complete signature

SELECT * FROM DEDUPLICATE_KEEP_FIRST(
    input                  => TABLE(user_events) PARTITION BY user_id,
    on_time                => DESCRIPTOR(rowtime),
    state_ttl              => INTERVAL '5' MINUTES,
    reset_ttl_on_duplicate => true,
    uid                    => 'user-events-dedup'
)

4.1.3 Use cases

Each use case is a combination of the parameters in Section 4.1.2; state_ttl combines freely with all of them.

event_ttl is deferred because Flink has no event-time TTL today (Section 8). Unkeyed deduplication with event-time ordering is also supported. It emits the single earliest-event-time row across the whole input once the watermark makes the choice final. It is omitted from the table because it is the same as Section 4.1.3.5 with one global partition.

4.1.3.1 Unkeyed, watermarkless

Keeping only the first record observed across the whole input, with no key and no time attribute.

• Input: insert-only stream; no PARTITION BY, no on_time.
• Output: insert-only; exactly one row, the first record observed.
• State: a single seen-marker.
• Key: none — all rows form one global partition; runs at parallelism 1.

SELECT * FROM DEDUPLICATE_KEEP_FIRST(input => TABLE(user_events))

Input
-- +I[id: 1, name: 'Vas']
-- +I[id: 1, name: 'Bob']

Output
-- +I[id: 1, name: 'Vas']

4.1.3.2 Keyed, watermarkless

Keeping the first record observed per key.

• Input: insert-only stream partitioned by one key column.
• Output: insert-only; the first record for each key.
• State: one seen-marker per key.
• Key: PARTITION BY user_name; scales out by key.

SELECT * FROM DEDUPLICATE_KEEP_FIRST(
    input => TABLE(user_events) PARTITION BY user_name
)

Input
-- +I[user_name: 'Vas', event: 'login']
-- +I[user_name: 'Vas', event: 'click']

Output
-- +I[user_name: 'Vas', event: 'login']

4.1.3.3 Keyed, watermarkless, state_ttl

As Section 4.1.3.2, with bounded state retention.

• Input: insert-only stream partitioned by one key column.
• Output: insert-only; the first record for each key.
• State: one seen-marker per key, retained for at least state_ttl of processing time; a later record for an expired key is emitted again.
• Key: PARTITION BY user_name.

SELECT * FROM DEDUPLICATE_KEEP_FIRST(
    input     => TABLE(user_events) PARTITION BY user_name,
    state_ttl => INTERVAL '5' MINUTES
)

Input (processing-time arrival)
-- +I[user_name: 'Vas', event: 'login']   @ 10:00
-- +I[user_name: 'Vas', event: 'login']   @ 10:03
-- +I[user_name: 'Vas', event: 'login']   @ 10:10

Output
-- +I[user_name: 'Vas', event: 'login']   (from 10:00)
-- +I[user_name: 'Vas', event: 'login']   (from 10:10, new period after the key expired 10:08)

The 10:03 duplicate is dropped; because reset_ttl_on_duplicate defaults to true, it refreshes the retention from 10:05 to 10:08. The 10:10 record then arrives after 10:08, so a new deduplication period begins and it is emitted.

4.1.3.4 Whole-row-as-key (explicit column list), watermarkless

Scalable exact-duplicate deduplication by listing all payload columns as the key. Replaces the equivalent ROW_NUMBER()-plus-TO_CHANGELOG query with a single append-in / append-out call.

• Input: insert-only stream partitioned by all payload columns.
• Output: insert-only; one row per distinct payload.
• State: one seen-marker per distinct payload.
• Key: PARTITION BY <all columns>; unlike Section 4.1.3.1 this stays on the keyed, scalable path.

SELECT * FROM DEDUPLICATE_KEEP_FIRST(
    input => TABLE(trades_with_dups) PARTITION BY trade_id, ticker, quantity, price
)

Input
-- +I[trade_id: 't1', ticker: 'AAPL', quantity: 1, price: 10]
-- +I[trade_id: 't1', ticker: 'AAPL', quantity: 1, price: 10]

Output
-- +I[trade_id: 't1', ticker: 'AAPL', quantity: 1, price: 10]

4.1.3.5 Keyed, event-time ordered, state_ttl (base case)

Keeping the earliest-event-time record per key, with bounded state retention.

• Input: insert-only stream with a rowtime attribute and watermarks, partitioned by one key column.
• Output: insert-only; the earliest-event-time record per key, emitted once the watermark makes the choice final.
• State: one candidate row and its event time per key, retained for at least state_ttl of processing time.
• Key: PARTITION BY user_id; scales out by key.

SELECT * FROM DEDUPLICATE_KEEP_FIRST(
    input     => TABLE(user_events) PARTITION BY user_id,
    on_time   => DESCRIPTOR(rowtime),
    state_ttl => INTERVAL '5' MINUTES
)

-- source table declares: WATERMARK FOR rowtime AS rowtime - INTERVAL '5' MINUTES
Input (@ = processing-time arrival; rowtime = event time; watermark = max(rowtime) - 5 min)
-- +I[user_id: 'u1', event: 'click',  rowtime: 10:02]   @ 12:00   watermark → 09:57
-- +I[user_id: 'u1', event: 'view',   rowtime: 10:01]   @ 12:01   on time (>= 09:57); earlier event time → candidate
-- +I[user_id: 'u1', event: 'scroll', rowtime: 10:06]   @ 12:02   watermark → 10:01; later, not the candidate

Output
-- +I[user_id: 'u1', event: 'view', rowtime: 10:01]     @ 12:02   (watermark reached 10:01 → candidate finalized and emitted)

A record that arrives after the key is finalized is late and dropped, even if its event time is earlier:

Input
-- +I[user_id: 'u1', event: 'signup', rowtime: 10:00]  @ 12:05   below watermark (10:01) → late, dropped

Output
-- (dropped — u1 already finalized; 10:00 is below the watermark)

4.1.3.6 Keyed, state_ttl with duplicate TTL refresh

Demonstrating reset_ttl_on_duplicate in its implementable form (paired with state_ttl).

• Input: insert-only stream partitioned by one key column.
• Output: insert-only; the first record for each key.
• State: one seen-marker per key. With reset_ttl_on_duplicate => true, each dropped duplicate refreshes the key's retention, so a continuously-duplicated key is never evicted; with false, the key is evicted state_ttl after its first record and the next record starts a new deduplication period.
• Key: PARTITION BY user_name.

SELECT * FROM DEDUPLICATE_KEEP_FIRST(
    input                  => TABLE(user_events) PARTITION BY user_name,
    state_ttl              => INTERVAL '5' MINUTES,
    reset_ttl_on_duplicate => true
)

The two outputs below run this same query on the same input, with reset_ttl_on_duplicate set to true and then false:

Input (@ = processing-time arrival)
-- +I[user_name: 'Vas', event: 'login']   @ 10:00
-- +I[user_name: 'Vas', event: 'login']   @ 10:03
-- +I[user_name: 'Vas', event: 'login']   @ 10:07
-- +I[user_name: 'Vas', event: 'login']   @ 10:10

Output with reset_ttl_on_duplicate => true   (each duplicate refreshes the 5-min TTL: 10:05 → 10:08 → 10:12)
-- +I[user_name: 'Vas', event: 'login']   (from 10:00)   -- key never expires; 10:03, 10:07, 10:10 all dropped

Output with reset_ttl_on_duplicate => false  (TTL fixed from the first record; expires 10:05)
-- +I[user_name: 'Vas', event: 'login']   (from 10:00)
-- +I[user_name: 'Vas', event: 'login']   (from 10:07, new period after expiry at 10:05)

The contrast is the point of the parameter: with true, a continuously-duplicated key keeps refreshing its retention and is never re-emitted; with false, retention is measured from the first record, so the 10:07 record falls outside the window and starts a new deduplication period.

4.1.3.7 Updating input, changes swallowed

Deduplicating an updating input into an insert-only result. The function keeps the first +I or +U observed per key and swallows every later change to that key. A -U or -D for a key that has not been seen yet is ignored and does not mark the key as seen, so the next +I or +U for that key is still emitted. This can happen when a changelog source starts from a later offset, or after this function's own state has expired.

• Input: updating stream (carries +I / -U / +U / -D), partitioned by one key column. (on_time is not permitted here, updating input runs in watermarkless mode only.)
• Output: insert-only; the first +I or +U observed for each key, never corrected or retracted.
• State: one seen-marker per key.
• Key: PARTITION BY user_name.

SELECT * FROM DEDUPLICATE_KEEP_FIRST(
    input => TABLE(user_events) PARTITION BY user_name
)

Input (updating changelog)
-- -D[user_name: 'Vas', event: 'logout']   key not seen yet → ignored, key not marked as seen
-- +I[user_name: 'Vas', event: 'login']    first +I/+U for Vas → emitted
-- +I[user_name: 'Vas', event: 'login']    key seen again → swallowed
-- -U[user_name: 'Vas', event: 'login']    retraction → swallowed
-- +U[user_name: 'Vas', event: 'click']    upsert → swallowed
-- -D[user_name: 'Vas', event: 'click']    delete → swallowed

Output
-- +I[user_name: 'Vas', event: 'login']

The first +I or +U observed for Vas (+I[Vas, login]) is emitted once as an insert. A -D that arrives before it is ignored. After that, every event on that key is swallowed — a later duplicate of the key, a retraction (-U), an upsert (-U/+U to click), and a delete (-D) all emit nothing. The function keeps the first value and never corrects or retracts it, so the output stays insert-only even though the input is updating.

5. Compatibility, Deprecation, and Migration Plan

This is a new function; there are no breaking changes and nothing is deprecated. Existing deduplication queries continue to work.

As a stateful operator, DEDUPLICATE_KEEP_FIRST participates in savepoint and compiled-plan compatibility from day one; the restore test in Section 6 is required. State restore across query evolution relies on the framework-provided uid argument, which is auto-appended to every PTF.

6. Test Plan

• Semantic tests covering: watermarkless first-arrival deduplication; single- and multi-column keys; unpartitioned (single global partition) deduplication; null-containing keys; event-time earliest selection; watermark-gated emission; a later record with an earlier event time replacing the candidate; state expiry with state_ttl; an updating input whose later changes (duplicate, -U/+U, -D) are all swallowed into an insert-only output; and a -D or -U for a key not yet seen, which is ignored without marking the key as seen.
• Planner / type-inference tests covering optional on_time validation, insert-only and updating input changelog modes with insert-only output, and output columns.
• Restore test with a committed compiled plan and savepoint, covering restore before and after the first record for a key and with an event-time candidate pending finalization.
• Documentation added to the SQL deduplication reference page (and its Chinese mirror), positioning the function as the append-in / append-out alternative to the ROW_NUMBER() approach.

7. Function Naming

The function is named DEDUPLICATE_KEEP_FIRST. Two principles drive it:

• No abbreviation, and no changelog mode in the name. Names spell out DEDUPLICATE, consistent with existing built-ins, and the input changelog mode is deliberately not part of the name. Whether the input is append-only or updating is a property of the input; the PTF framework derives the output changelog mode from it. KEEP_FIRST also mirrors the existing FIRST_VALUE / LAST_VALUE functions.
• One name per behavior, not per changelog mode. Because mode is derived, keep-first is a single function across append and updating input: it emits the first record per key and never retracts it, so its output is append. The companion DEDUPLICATE_KEEP_LAST is a separate future FLIP; keep-last must retract the previously emitted row, so its output is updating by necessity.

This FLIP proposes only DEDUPLICATE_KEEP_FIRST. It accepts insert-only or updating input and always produces insert-only output; the name does not bake in the changelog mode, so input support can widen further later without a rename.

8. Rejected and Deferred Alternatives

• Event-time TTL (event_ttl) — the deferred row in Section 4.1.3. Deferred. Retaining state by event time is more accurate for out-of-order and reprocessing workloads, but Flink has no event-time TTL today and adding one would require an engine change beyond this function. This FLIP relies on the existing processing-time state_ttl; reset_ttl_on_duplicate still applies to it (Section 4.1.3.6).
• Shipping a function-specific default state_ttl (e.g., 1 day or 7 days). Rejected. The function follows table.exec.state.ttl, as other PTFs do, so state is retained indefinitely unless the user configures a TTL.
• PARTITION BY ALL / PARTITION BY * / implicit whole-row partitioning. A nice feature to have but deferred to a next improvement. Exact-duplicate deduplication is expressed today by listing all payload columns (Section 4.1.3.4). A shorthand depends on Calcite support and on defined equality, null, serialization, and parallelism behavior.
• Ordering by an arbitrary non-time column. Deferred. on_time accepts only a rowtime attribute; a future ordered variant would need to define its ordering and buffering semantics.
• Expressing event-time ordering with an ORDER BY clause. Considered, and dropped in the design discussion in favour of handling the ordering inside the function via on_time. A set-semantic ORDER BY would let the framework sort the input and pass it to the function already ordered, but it buffers every record still inside the watermark window for each key, while keep-earliest only ever needs the single earliest row. Handling it internally keeps one row of state per key and needs no new syntax. This follows the precedent of Flink's other time-based PTFs — for example the temporal / lateral snapshot join — which take an on_time argument and sort internally rather than exposing ORDER BY. For keep-earliest the operator does not even sort: the ordering key is a single timestamp, so it just keeps the earliest row per key in state.
• A keep-last variant of this append function. Rejected: keeping the last record would require retracting the previously emitted one, which an append-only output cannot do. Keep-last over an updating output is a separate future function (DEDUPLICATE_KEEP_LAST; see Section 7), not a rejected one.
• Deriving updating output from updating input. Rejected: Updating pipelines are hard to handle and reason about; the more insert-only pipelines produced, the better.
• Event-time mode for updating input. Rejected: watermarks and updates are incompatible (a watermark finalizes a timestamp; an update can change or retract a past row afterward). This is the split that separated the original UPDATE_EARLIEST and APPEND_EARLIEST prototypes; with a single function, on_time is limited to insert-only input.

9. Open Design Points

• reset_ttl_on_duplicate. The default is proposed as true (retention tracks the most recent occurrence of a key). The point flagged for the discussion thread is whether exposing the toggle is warranted and whether true is right given the tradeoff: always refreshing means a continuously-duplicated key is never evicted, which some consider a drawback.
• Default state TTL duration — Decided: state_ttl falls back to table.exec.state.ttl, consistent with other PTFs.
• state_ttl parameter name. The name distinguishes it from a possible future event_ttl, but naming is not yet consistent across the first built-in PTFs (an existing time-based PTF calls the equivalent argument ttl). Reconcile the name across built-in PTFs before implementation.
• Function naming. The SQL function name (Section 7) is expected to be debated; confirm it — together with the updating siblings — before implementation.