This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 4fd7447f3fc [FLINK-40896][table] Add built-in DEDUPLICATE_KEEP_FIRST
process table function
4fd7447f3fc is described below
commit 4fd7447f3fcd761c47c1eef84a90afefa0bd8161
Author: Vasudev Kelappassery <[email protected]>
AuthorDate: Tue Oct 27 16:31:13 2026 +0000
[FLINK-40896][table] Add built-in DEDUPLICATE_KEEP_FIRST process table
function
This closes #29365.
---
.../docs/sql/functions/built-in-functions.md | 1 +
.../reference/queries/deduplicate-keep-first.md | 192 +++++++++
.../docs/sql/functions/built-in-functions.md | 1 +
.../reference/queries/deduplicate-keep-first.md | 192 +++++++++
.../functions/BuiltInFunctionDefinitions.java | 25 ++
.../DeduplicateKeepFirstTypeStrategy.java | 124 ++++++
.../stream/DeduplicateKeepFirstRestoreTest.java | 65 +++
.../DeduplicateKeepFirstRestoreTestPrograms.java | 98 +++++
.../stream/DeduplicateKeepFirstSemanticTests.java | 50 +++
.../stream/DeduplicateKeepFirstTestPrograms.java | 456 +++++++++++++++++++++
...cate-keep-first-event-time-pending-restore.json | 352 ++++++++++++++++
.../savepoint/_metadata | Bin 0 -> 18455 bytes
.../plan/deduplicate-keep-first-keyed-restore.json | 143 +++++++
.../savepoint/_metadata | Bin 0 -> 13722 bytes
.../functions/ptf/DeduplicateKeepFirst.java | 258 ++++++++++++
.../functions/ptf/DeduplicateKeepFirstTest.java | 174 ++++++++
16 files changed, 2131 insertions(+)
diff --git a/docs/content.zh/docs/sql/functions/built-in-functions.md
b/docs/content.zh/docs/sql/functions/built-in-functions.md
index 242cb4b0358..79115623e0a 100644
--- a/docs/content.zh/docs/sql/functions/built-in-functions.md
+++ b/docs/content.zh/docs/sql/functions/built-in-functions.md
@@ -289,6 +289,7 @@ Table functions can be used in two ways: as stand-alone
inputs, where they are i
| `FROM_CHANGELOG(input => TABLE t [, ...])` | Converts an append-only table
with an explicit operation column into a dynamic table. See Changelog
Conversion for the full list of arguments, semantics, and usage.
|
| `TO_CHANGELOG(input => TABLE t [, ...])` | Converts a dynamic table into
an append-only table with an explicit operation column. See Changelog
Conversion for the full list of arguments, semantics, and usage.
|
| `SNAPSHOT(input => TABLE t [, ...])` | Returns the current state of a
dynamic table `t`. `SNAPSHOT` can only be used in a `LATERAL` context and not
as a stand-alone table function. See [LATERAL SNAPSHOT join]({{< ref
"docs/sql/reference/queries/joins" >}}#lateral-snapshot-join) for the full list
of arguments, the join semantics, and usage. |
+| `DEDUPLICATE_KEEP_FIRST(input => TABLE t [, ...])` | Keeps the first record
per key as an append-only table (first arrival, or earliest event time with
`on_time`). See [Deduplicate Keep First]({{< ref
"docs/sql/reference/queries/deduplicate-keep-first" >}}#deduplicate_keep_first)
for the full list of arguments, semantics, and usage. |
To implement your own table functions, see [user-defined table functions]({{<
ref "docs/dev/table/functions/udfs" >}}#table-functions).
diff --git
a/docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md
b/docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md
new file mode 100644
index 00000000000..4e307552e4f
--- /dev/null
+++ b/docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md
@@ -0,0 +1,192 @@
+---
+title: "Deduplicate Keep First"
+weight: 16
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Deduplicate Keep First
+
+{{< label Streaming >}}
+
+Flink SQL provides the `DEDUPLICATE_KEEP_FIRST` process table function (PTF)
for removing duplicate rows while keeping only the **first** record per key.
Its output is always **insert-only**, which makes it a concise alternative to
the `ROW_NUMBER()` [deduplication]({{< ref
"docs/sql/reference/queries/deduplication" >}}) pattern with more configuration
options.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record
per key as an append-only table (first arrival, or earliest event time with
`on_time`) |
+
+## DEDUPLICATE_KEEP_FIRST
+
+The `DEDUPLICATE_KEEP_FIRST` PTF keeps the first record per key and drops
every later record for that key. It supports two ordering modes:
+
+* **Processing time (default):** without `on_time`, the first record
*observed* for a key is emitted; later records for that key are dropped while
its state exists. This requires no watermark or event-time attribute.
+* **Event time:** with `on_time`, the record with the *smallest event time*
per key is kept and emitted once the watermark makes the choice 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 as
late.
+
+The input may be an append-only or updating table; the output is append-only
in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY (key_col [, key_col2 ...])],
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>,]
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [uid => <string>]
+)
+```
+
+`on_time` and `uid` are system arguments and always come last in the
signature, which matters when arguments are passed by position.
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (append-only or updating). Use `PARTITION
BY` to deduplicate per key; all rows for a key are routed to the same parallel
instance. Without `PARTITION BY`, the whole input is a single group processed
at parallelism 1, so only the first row of the entire stream is emitted and
every later row is dropped regardless of its content. |
+| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for
the per-key deduplication state. After a key's state expires, a later record
for that key starts a new deduplication period and may be emitted again. If
omitted, falls back to `table.exec.state.ttl`; at its default of 0, state is
retained indefinitely, but a retention can still be introduced later.
`state_ttl => INTERVAL '0'` also retains state indefinitely, but excludes the
TTL from the state layout, so a retent [...]
+| `reset_ttl_on_duplicate` | No | Whether a later duplicate refreshes the
key's `state_ttl`. Defaults to `TRUE`, so retention tracks the most recent
occurrence of a key. Only meaningful when a TTL applies, whether set through
`state_ttl` or inherited from `table.exec.state.ttl`. |
+| `on_time` | No | A `DESCRIPTOR` naming a single watermarked column. When
provided, the row with the smallest event time per key is kept and emitted once
the watermark passes that timestamp (late rows are dropped). When omitted, the
function keeps the first row observed for a key (arrival order). Requires
append-only input; combining `on_time` with an updating input is rejected at
planning time. |
+| `uid` | No | A unique identifier for the function's state in checkpoints and
savepoints. Defaults to the function name, so it is required when
`DEDUPLICATE_KEEP_FIRST` is called more than once in the same job. See [Query
Evolution with UIDs]({{< ref "docs/dev/table/functions/ptfs"
>}}#query-evolution-with-uids). |
+
+### Output Schema
+
+When `PARTITION BY` is used, the partition key columns are prepended to the
output, followed by the remaining (non-key) input columns. In event-time mode a
rowtime column is appended for (potentially following) time-based operations.
+
+```
+[partition_key_columns] + [remaining_input_columns]
+```
+
+### Examples
+
+#### Keyed keep-first (processing time)
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'click']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY user_name
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+```
+
+The first row for `Vas` is emitted; the later `click` on the same key is
dropped.
+
+#### Exact duplicates (all columns as key)
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'click']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY (user_name, action)
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'click']
+```
+
+Partitioning by every column drops only exact duplicate rows while still
distributing the work across parallel instances.
+
+#### Bounded state with state_ttl
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'click']
+-- +I[user_name:'Vas', action:'view']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY user_name,
+ state_ttl => INTERVAL '5' SECOND,
+ reset_ttl_on_duplicate => FALSE
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+```
+
+`state_ttl` bounds how long per-key state is retained (processing time). While
the state exists, duplicates are dropped and only the first row per key is
emitted; after the state expires, a later record for the key starts a new
deduplication period. With the default `reset_ttl_on_duplicate => TRUE`, each
dropped duplicate refreshes the retention window; with `FALSE`, retention is
measured from the first record for the key.
+
+#### Earliest by event time
+
+```sql
+-- Source declares: WATERMARK FOR ts AS ts - INTERVAL '10' SECOND
+-- Input (ts = event time):
+-- +I[user_name:'Vas', action:'login', ts:3.000]
+-- +I[user_name:'Vas', action:'click', ts:1.000]
+-- +I[user_name:'Vas', action:'view', ts:5.000]
+
+SELECT user_name, action FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY user_name,
+ on_time => DESCRIPTOR(ts)
+)
+
+-- Output (insert-only, once the watermark finalizes the earliest):
+-- +I[user_name:'Vas', action:'click']
+```
+
+With `on_time`, the record with the smallest event time per key is kept (here
`click` at `ts 1.000`, not `login` which arrived first) and emitted once the
watermark passes that timestamp. Late events (i.e. events with timestamp
smaller or equal to the received watermark) are not taken into consideration.
+
+#### Updating input
+
+```sql
+-- Input (updating changelog):
+-- -D[user_name:'Vas', action:'logout'] key not seen yet -> ignored
+-- +I[user_name:'Vas', action:'login'] first record for the key -> emitted
+-- +I[user_name:'Vas', action:'login'] duplicate -> swallowed
+-- -U[user_name:'Vas', action:'login'] retraction -> swallowed
+-- +U[user_name:'Vas', action:'click'] update -> swallowed
+-- -D[user_name:'Vas', action:'click'] delete -> swallowed
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY user_name
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+```
+
+The input may be an [updating changelog]({{< ref
"docs/dev/table/functions/ptfs" >}}#change-data-capture-basics) containing
inserts (`+I`), update-befores (`-U`), update-afters (`+U`), and deletes
(`-D`). `DEDUPLICATE_KEEP_FIRST` keeps the first `+I` or `+U` observed per key
and swallows every later change to that key (duplicate, `-U`, `+U`, `-D`), so
the output stays insert-only. 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 nex [...]
+
+#### Table API
+
+```java
+Table userEvents = ...;
+
+// Keep the first row per key (processing time).
+Table result =
userEvents.partitionBy($("user_name")).process("DEDUPLICATE_KEEP_FIRST");
+
+// With optional arguments.
+Table result = userEvents
+ .partitionBy($("user_name"))
+ .process(
+ "DEDUPLICATE_KEEP_FIRST",
+ lit(Duration.ofSeconds(5)).asArgument("state_ttl"),
+ lit(false).asArgument("reset_ttl_on_duplicate"));
+```
+
+{{< top >}}
diff --git a/docs/content/docs/sql/functions/built-in-functions.md
b/docs/content/docs/sql/functions/built-in-functions.md
index 40b54b2869b..01bf2070579 100644
--- a/docs/content/docs/sql/functions/built-in-functions.md
+++ b/docs/content/docs/sql/functions/built-in-functions.md
@@ -292,6 +292,7 @@ Table functions can be used in two ways: as stand-alone
inputs, where they are i
| `FROM_CHANGELOG(input => TABLE t [, ...])` | Converts an append-only table
with an explicit operation column into a dynamic table. See [Changelog
Conversion]({{< ref "docs/sql/reference/queries/changelog" >}}#from_changelog)
for the full list of arguments, semantics, and usage.
|
| `TO_CHANGELOG(input => TABLE t [, ...])` | Converts a dynamic table into
an append-only table with an explicit operation column. See [Changelog
Conversion]({{< ref "docs/sql/reference/queries/changelog" >}}#to_changelog)
for the full list of arguments, semantics, and usage.
|
| `SNAPSHOT(input => TABLE t [, ...])` | Returns the current state of a
dynamic table `t`. `SNAPSHOT` can only be used in a `LATERAL` context and not
as a stand-alone table function. See [LATERAL SNAPSHOT join]({{< ref
"docs/sql/reference/queries/joins" >}}#lateral-snapshot-join) for the full list
of arguments, the join semantics, and usage. |
+| `DEDUPLICATE_KEEP_FIRST(input => TABLE t [, ...])` | Keeps the first record
per key as an append-only table (first arrival, or earliest event time with
`on_time`). See [Deduplicate Keep First]({{< ref
"docs/sql/reference/queries/deduplicate-keep-first" >}}#deduplicate_keep_first)
for the full list of arguments, semantics, and usage. |
To implement your own table functions, see [user-defined table functions]({{<
ref "docs/dev/table/functions/udfs" >}}#table-functions).
diff --git a/docs/content/docs/sql/reference/queries/deduplicate-keep-first.md
b/docs/content/docs/sql/reference/queries/deduplicate-keep-first.md
new file mode 100644
index 00000000000..4e307552e4f
--- /dev/null
+++ b/docs/content/docs/sql/reference/queries/deduplicate-keep-first.md
@@ -0,0 +1,192 @@
+---
+title: "Deduplicate Keep First"
+weight: 16
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Deduplicate Keep First
+
+{{< label Streaming >}}
+
+Flink SQL provides the `DEDUPLICATE_KEEP_FIRST` process table function (PTF)
for removing duplicate rows while keeping only the **first** record per key.
Its output is always **insert-only**, which makes it a concise alternative to
the `ROW_NUMBER()` [deduplication]({{< ref
"docs/sql/reference/queries/deduplication" >}}) pattern with more configuration
options.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record
per key as an append-only table (first arrival, or earliest event time with
`on_time`) |
+
+## DEDUPLICATE_KEEP_FIRST
+
+The `DEDUPLICATE_KEEP_FIRST` PTF keeps the first record per key and drops
every later record for that key. It supports two ordering modes:
+
+* **Processing time (default):** without `on_time`, the first record
*observed* for a key is emitted; later records for that key are dropped while
its state exists. This requires no watermark or event-time attribute.
+* **Event time:** with `on_time`, the record with the *smallest event time*
per key is kept and emitted once the watermark makes the choice 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 as
late.
+
+The input may be an append-only or updating table; the output is append-only
in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY (key_col [, key_col2 ...])],
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>,]
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [uid => <string>]
+)
+```
+
+`on_time` and `uid` are system arguments and always come last in the
signature, which matters when arguments are passed by position.
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (append-only or updating). Use `PARTITION
BY` to deduplicate per key; all rows for a key are routed to the same parallel
instance. Without `PARTITION BY`, the whole input is a single group processed
at parallelism 1, so only the first row of the entire stream is emitted and
every later row is dropped regardless of its content. |
+| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for
the per-key deduplication state. After a key's state expires, a later record
for that key starts a new deduplication period and may be emitted again. If
omitted, falls back to `table.exec.state.ttl`; at its default of 0, state is
retained indefinitely, but a retention can still be introduced later.
`state_ttl => INTERVAL '0'` also retains state indefinitely, but excludes the
TTL from the state layout, so a retent [...]
+| `reset_ttl_on_duplicate` | No | Whether a later duplicate refreshes the
key's `state_ttl`. Defaults to `TRUE`, so retention tracks the most recent
occurrence of a key. Only meaningful when a TTL applies, whether set through
`state_ttl` or inherited from `table.exec.state.ttl`. |
+| `on_time` | No | A `DESCRIPTOR` naming a single watermarked column. When
provided, the row with the smallest event time per key is kept and emitted once
the watermark passes that timestamp (late rows are dropped). When omitted, the
function keeps the first row observed for a key (arrival order). Requires
append-only input; combining `on_time` with an updating input is rejected at
planning time. |
+| `uid` | No | A unique identifier for the function's state in checkpoints and
savepoints. Defaults to the function name, so it is required when
`DEDUPLICATE_KEEP_FIRST` is called more than once in the same job. See [Query
Evolution with UIDs]({{< ref "docs/dev/table/functions/ptfs"
>}}#query-evolution-with-uids). |
+
+### Output Schema
+
+When `PARTITION BY` is used, the partition key columns are prepended to the
output, followed by the remaining (non-key) input columns. In event-time mode a
rowtime column is appended for (potentially following) time-based operations.
+
+```
+[partition_key_columns] + [remaining_input_columns]
+```
+
+### Examples
+
+#### Keyed keep-first (processing time)
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'click']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY user_name
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+```
+
+The first row for `Vas` is emitted; the later `click` on the same key is
dropped.
+
+#### Exact duplicates (all columns as key)
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'click']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY (user_name, action)
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'click']
+```
+
+Partitioning by every column drops only exact duplicate rows while still
distributing the work across parallel instances.
+
+#### Bounded state with state_ttl
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Vas', action:'click']
+-- +I[user_name:'Vas', action:'view']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY user_name,
+ state_ttl => INTERVAL '5' SECOND,
+ reset_ttl_on_duplicate => FALSE
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+```
+
+`state_ttl` bounds how long per-key state is retained (processing time). While
the state exists, duplicates are dropped and only the first row per key is
emitted; after the state expires, a later record for the key starts a new
deduplication period. With the default `reset_ttl_on_duplicate => TRUE`, each
dropped duplicate refreshes the retention window; with `FALSE`, retention is
measured from the first record for the key.
+
+#### Earliest by event time
+
+```sql
+-- Source declares: WATERMARK FOR ts AS ts - INTERVAL '10' SECOND
+-- Input (ts = event time):
+-- +I[user_name:'Vas', action:'login', ts:3.000]
+-- +I[user_name:'Vas', action:'click', ts:1.000]
+-- +I[user_name:'Vas', action:'view', ts:5.000]
+
+SELECT user_name, action FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY user_name,
+ on_time => DESCRIPTOR(ts)
+)
+
+-- Output (insert-only, once the watermark finalizes the earliest):
+-- +I[user_name:'Vas', action:'click']
+```
+
+With `on_time`, the record with the smallest event time per key is kept (here
`click` at `ts 1.000`, not `login` which arrived first) and emitted once the
watermark passes that timestamp. Late events (i.e. events with timestamp
smaller or equal to the received watermark) are not taken into consideration.
+
+#### Updating input
+
+```sql
+-- Input (updating changelog):
+-- -D[user_name:'Vas', action:'logout'] key not seen yet -> ignored
+-- +I[user_name:'Vas', action:'login'] first record for the key -> emitted
+-- +I[user_name:'Vas', action:'login'] duplicate -> swallowed
+-- -U[user_name:'Vas', action:'login'] retraction -> swallowed
+-- +U[user_name:'Vas', action:'click'] update -> swallowed
+-- -D[user_name:'Vas', action:'click'] delete -> swallowed
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events PARTITION BY user_name
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+```
+
+The input may be an [updating changelog]({{< ref
"docs/dev/table/functions/ptfs" >}}#change-data-capture-basics) containing
inserts (`+I`), update-befores (`-U`), update-afters (`+U`), and deletes
(`-D`). `DEDUPLICATE_KEEP_FIRST` keeps the first `+I` or `+U` observed per key
and swallows every later change to that key (duplicate, `-U`, `+U`, `-D`), so
the output stays insert-only. 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 nex [...]
+
+#### Table API
+
+```java
+Table userEvents = ...;
+
+// Keep the first row per key (processing time).
+Table result =
userEvents.partitionBy($("user_name")).process("DEDUPLICATE_KEEP_FIRST");
+
+// With optional arguments.
+Table result = userEvents
+ .partitionBy($("user_name"))
+ .process(
+ "DEDUPLICATE_KEEP_FIRST",
+ lit(Duration.ofSeconds(5)).asArgument("state_ttl"),
+ lit(false).asArgument("reset_ttl_on_duplicate"));
+```
+
+{{< top >}}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 9e993e26429..6c09b633d81 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -38,6 +38,7 @@ import
org.apache.flink.table.types.inference.StaticArgumentTrait;
import org.apache.flink.table.types.inference.TraitCondition;
import org.apache.flink.table.types.inference.TypeStrategies;
import
org.apache.flink.table.types.inference.strategies.ArrayOfStringArgumentTypeStrategy;
+import
org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
import
org.apache.flink.table.types.inference.strategies.FromChangelogTypeStrategy;
import
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies;
import
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies;
@@ -939,6 +940,30 @@ public final class BuiltInFunctionDefinitions {
"org.apache.flink.table.runtime.functions.ptf.FromChangelogFunction")
.build();
+ public static final BuiltInFunctionDefinition DEDUPLICATE_KEEP_FIRST =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("DEDUPLICATE_KEEP_FIRST")
+ .kind(PROCESS_TABLE)
+ .staticArguments(
+ StaticArgument.table(
+ "input",
+ Row.class,
+ false,
+ EnumSet.of(
+ StaticArgumentTrait.TABLE,
+
StaticArgumentTrait.SET_SEMANTIC_TABLE,
+
StaticArgumentTrait.OPTIONAL_PARTITION_BY,
+
StaticArgumentTrait.SUPPORT_UPDATES)),
+ StaticArgument.scalar(
+ "state_ttl",
DataTypes.INTERVAL(DataTypes.SECOND()), true),
+ StaticArgument.scalar(
+ "reset_ttl_on_duplicate",
DataTypes.BOOLEAN(), true))
+
.inputTypeStrategy(DeduplicateKeepFirstTypeStrategy.INPUT_TYPE_STRATEGY)
+
.outputTypeStrategy(DeduplicateKeepFirstTypeStrategy.OUTPUT_TYPE_STRATEGY)
+ .runtimeClass(
+
"org.apache.flink.table.runtime.functions.ptf.DeduplicateKeepFirst")
+ .build();
+
/**
* Built-in proxy function for the LATERAL SNAPSHOT temporal join.
*
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java
new file mode 100644
index 00000000000..3ea1167da87
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.DataTypes.Field;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Type strategies for the {@code DEDUPLICATE_KEEP_FIRST} process table
function. */
+@Internal
+public final class DeduplicateKeepFirstTypeStrategy {
+
+ public static final int ARG_INPUT = 0;
+ public static final int ARG_STATE_TTL = 1;
+ public static final int ARG_RESET_TTL_ON_DUPLICATE = 2;
+
+ public static final InputTypeStrategy INPUT_TYPE_STRATEGY =
+ new ValidationOnlyInputTypeStrategy() {
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ final CallContext callContext, final boolean
throwOnFailure) {
+ final Optional<List<DataType>> stateTtlError =
+ validateStateTtl(callContext, throwOnFailure);
+ if (stateTtlError.isPresent()) {
+ return stateTtlError;
+ }
+
+ final Optional<List<DataType>> resetTtlError =
+ validateResetTtlOnDuplicate(callContext,
throwOnFailure);
+ if (resetTtlError.isPresent()) {
+ return resetTtlError;
+ }
+
+ return Optional.of(callContext.getArgumentDataTypes());
+ }
+ };
+
+ public static final TypeStrategy OUTPUT_TYPE_STRATEGY =
+ callContext -> {
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(ARG_INPUT)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "First argument must
be a table for DEDUPLICATE_KEEP_FIRST."));
+
+ final List<Field> inputFields =
DataType.getFields(tableSemantics.dataType());
+ final List<Field> outputFields =
+ Arrays.stream(
+
ChangelogTypeStrategyUtils.computeOutputIndices(
+ tableSemantics))
+ .mapToObj(inputFields::get)
+ .collect(Collectors.toList());
+ return Optional.of(DataTypes.ROW(outputFields).notNull());
+ };
+
+ private static Optional<List<DataType>> validateStateTtl(
+ final CallContext callContext, final boolean throwOnFailure) {
+ if (callContext.isArgumentNull(ARG_STATE_TTL)) {
+ return Optional.empty();
+ }
+ if (!callContext.isArgumentLiteral(ARG_STATE_TTL)) {
+ return callContext.fail(
+ throwOnFailure,
+ "The 'state_ttl' argument must be a constant INTERVAL
literal.");
+ }
+ final Optional<Duration> stateTtl =
+ callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ if (stateTtl.isPresent() && stateTtl.get().isNegative()) {
+ return callContext.fail(
+ throwOnFailure,
+ "The 'state_ttl' argument must not be negative, but was
'%s'. "
+ + "Use a positive INTERVAL to bound state
retention, INTERVAL '0' to "
+ + "retain state indefinitely without a TTL, or
omit the argument to "
+ + "fall back to session defaults.",
+ stateTtl.get());
+ }
+ return Optional.empty();
+ }
+
+ private static Optional<List<DataType>> validateResetTtlOnDuplicate(
+ final CallContext callContext, final boolean throwOnFailure) {
+ if (callContext.isArgumentNull(ARG_RESET_TTL_ON_DUPLICATE)) {
+ return Optional.empty();
+ }
+ if (!callContext.isArgumentLiteral(ARG_RESET_TTL_ON_DUPLICATE)) {
+ return callContext.fail(
+ throwOnFailure,
+ "The 'reset_ttl_on_duplicate' argument must be a constant
BOOLEAN literal.");
+ }
+ return Optional.empty();
+ }
+
+ private DeduplicateKeepFirstTypeStrategy() {}
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTest.java
new file mode 100644
index 00000000000..fa9ad95f085
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTest.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+/** Restore tests for the built-in DEDUPLICATE_KEEP_FIRST PTF. */
+public class DeduplicateKeepFirstRestoreTest extends RestoreTestBase {
+
+ // Bounds the wait for the pending-candidate savepoint trigger so a stuck
source fails fast.
+ private static final long SAVEPOINT_READY_TIMEOUT_MILLIS =
TimeUnit.MINUTES.toMillis(5);
+
+ public DeduplicateKeepFirstRestoreTest() {
+ super(StreamExecProcessTableFunction.class);
+ }
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return List.of(
+ DeduplicateKeepFirstRestoreTestPrograms.KEYED_RESTORE,
+
DeduplicateKeepFirstRestoreTestPrograms.EVENT_TIME_PENDING_RESTORE);
+ }
+
+ @Override
+ protected void awaitSavepointReady(TableTestProgram program,
List<CompletableFuture<?>> futures)
+ throws Exception {
+ if (program !=
DeduplicateKeepFirstRestoreTestPrograms.EVENT_TIME_PENDING_RESTORE) {
+ super.awaitSavepointReady(program, futures);
+ return;
+ }
+ // The candidate is buffered but not emitted before restore, so there
is no sink output to
+ // gate on. Instead wait until the source has emitted its
before-restore rows;
+ // stop-with-savepoint then drains them into keyed state before
snapshotting.
+ for (SourceTestStep source : program.getSetupSourceTestSteps()) {
+ final int count = source.dataBeforeRestore.size();
+ if (count > 0) {
+ TestValuesTableFactory.awaitSourceEmitted(source.name, count)
+ .get(SAVEPOINT_READY_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS);
+ }
+ }
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java
new file mode 100644
index 00000000000..f33d91892d0
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import java.time.Instant;
+
+/** Restore {@link TableTestProgram}s for the built-in DEDUPLICATE_KEEP_FIRST
PTF. */
+public class DeduplicateKeepFirstRestoreTestPrograms {
+
+ private static final String[] USER_EVENTS_SCHEMA = {"user_name STRING",
"action STRING"};
+
+ private static final String[] TIMED_EVENTS_SCHEMA = {
+ "user_name STRING",
+ "action STRING",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND"
+ };
+
+ public static final TableTestProgram KEYED_RESTORE =
+ TableTestProgram.of(
+ "deduplicate-keep-first-keyed-restore",
+ "the keep-first 'seen' state restores via compiled
plan + savepoint: "
+ + "an already-seen key stays deduplicated
after restore, and a "
+ + "key first observed after restore is
emitted")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedBeforeRestore(
+ Row.of("Vas", "login"),
Row.of("Vas", "click"))
+ .producedAfterRestore(
+ Row.of("Vas", "logout"),
Row.of("Alice", "hello"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedBeforeRestore(
+ Row.ofKind(RowKind.INSERT, "Vas",
"login"))
+ .consumedAfterRestore(
+ Row.ofKind(RowKind.INSERT,
"Alice", "hello"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name)")
+ .build();
+
+ public static final TableTestProgram EVENT_TIME_PENDING_RESTORE =
+ TableTestProgram.of(
+
"deduplicate-keep-first-event-time-pending-restore",
+ "an event-time candidate buffered but not yet
finalized (its timer has "
+ + "not fired) survives the savepoint:
after restore, the "
+ + "watermark advances and the buffered
candidate is emitted")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(TIMED_EVENTS_SCHEMA)
+ .producedBeforeRestore(
+ Row.of("Bob", "first",
Instant.ofEpochMilli(5)),
+ Row.of("Alice", "first",
Instant.ofEpochMilli(5)),
+ Row.of("Carol", "first",
Instant.ofEpochMilli(5)))
+ .producedAfterRestore(
+ Row.of("Bob", "later",
Instant.ofEpochMilli(100)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedBeforeRestore(new Row[0])
+ .consumedAfterRestore(
+ Row.ofKind(RowKind.INSERT, "Bob",
"first"),
+ Row.ofKind(RowKind.INSERT,
"Alice", "first"),
+ Row.ofKind(RowKind.INSERT,
"Carol", "first"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT user_name, action FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name, "
+ + "on_time => DESCRIPTOR(ts))")
+ .build();
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstSemanticTests.java
new file mode 100644
index 00000000000..89e8367cee4
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstSemanticTests.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.stream;
+
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.SemanticTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.List;
+
+/** Semantic tests for the built-in DEDUPLICATE_KEEP_FIRST process table
function. */
+public class DeduplicateKeepFirstSemanticTests extends SemanticTestBase {
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return List.of(
+ DeduplicateKeepFirstTestPrograms.KEYED,
+ DeduplicateKeepFirstTestPrograms.MULTI_KEY,
+ DeduplicateKeepFirstTestPrograms.MULTI_COLUMN_KEY,
+ DeduplicateKeepFirstTestPrograms.NULL_KEY,
+ DeduplicateKeepFirstTestPrograms.NO_PARTITION_BY,
+ DeduplicateKeepFirstTestPrograms.RESET_WITH_ZERO_TTL,
+ DeduplicateKeepFirstTestPrograms.NO_TTL_RESET,
+ DeduplicateKeepFirstTestPrograms.WHOLE_ROW_KEY,
+ DeduplicateKeepFirstTestPrograms.EVENT_TIME,
+
DeduplicateKeepFirstTestPrograms.EVENT_TIME_CANDIDATE_FIELD_NAMES,
+ DeduplicateKeepFirstTestPrograms.EVENT_TIME_LATE_DROPPED,
+ DeduplicateKeepFirstTestPrograms.UPDATING_INPUT,
+ DeduplicateKeepFirstTestPrograms.UNSEEN_RETRACTION_IGNORED,
+ DeduplicateKeepFirstTestPrograms.UPDATING_EVENT_TIME_FAILS,
+ DeduplicateKeepFirstTestPrograms.NEGATIVE_TTL_FAILS,
+ DeduplicateKeepFirstTestPrograms.NON_LITERAL_TTL_FAILS,
+ DeduplicateKeepFirstTestPrograms.NON_LITERAL_RESET_FAILS);
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java
new file mode 100644
index 00000000000..2ecebb7e798
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import java.time.Instant;
+
+/** {@link TableTestProgram} definitions for testing the built-in
DEDUPLICATE_KEEP_FIRST PTF. */
+public class DeduplicateKeepFirstTestPrograms {
+
+ private static final String[] USER_EVENTS_SCHEMA = {"user_name STRING",
"action STRING"};
+
+ private static final String[] TIMED_EVENTS_SCHEMA = {
+ "user_name STRING",
+ "action STRING",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL '10' SECOND"
+ };
+
+ public static final TableTestProgram KEYED =
+ TableTestProgram.of(
+ "deduplicate-keep-first-keyed",
+ "only the first row per PARTITION BY key is
emitted")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedValues(Row.of("Vas", "login"),
Row.of("Vas", "click"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(Row.ofKind(RowKind.INSERT,
"Vas", "login"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name)")
+ .build();
+
+ public static final TableTestProgram MULTI_KEY =
+ TableTestProgram.of(
+ "deduplicate-keep-first-multi-key",
+ "each PARTITION BY key keeps its own first row,
independently of "
+ + "interleaving with other keys")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedValues(
+ Row.of("Vas", "login"),
+ Row.of("Alice", "click"),
+ Row.of("Vas", "click"),
+ Row.of("Alice", "view"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "Vas",
"login"),
+ Row.ofKind(RowKind.INSERT,
"Alice", "click"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name)")
+ .build();
+
+ public static final TableTestProgram MULTI_COLUMN_KEY =
+ TableTestProgram.of(
+ "deduplicate-keep-first-multi-column-key",
+ "a multi-column PARTITION BY keeps the first row
per composite key; "
+ + "rows differing in either key column
form distinct groups")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema("region STRING", "user_name
STRING", "action STRING")
+ .producedValues(
+ Row.of("EU", "Vas", "login"),
+ Row.of("EU", "Vas", "click"),
+ Row.of("US", "Vas", "signup"),
+ Row.of("EU", "Alice", "view"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("user_name STRING", "region
STRING", "action STRING")
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "Vas",
"EU", "login"),
+ Row.ofKind(RowKind.INSERT, "Vas",
"US", "signup"),
+ Row.ofKind(RowKind.INSERT,
"Alice", "EU", "view"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
"
+ + "(user_name, region))")
+ .build();
+
+ public static final TableTestProgram NULL_KEY =
+ TableTestProgram.of(
+ "deduplicate-keep-first-null-key",
+ "a NULL PARTITION BY key forms its own dedup
group, keeping only "
+ + "its first row")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedValues(
+ Row.of(null, "login"),
+ Row.of("Vas", "click"),
+ Row.of(null, "click"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, null,
"login"),
+ Row.ofKind(RowKind.INSERT, "Vas",
"click"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name)")
+ .build();
+
+ public static final TableTestProgram NO_PARTITION_BY =
+ TableTestProgram.of(
+ "deduplicate-keep-first-no-partition-by",
+ "without PARTITION BY the whole input is one
group: only the "
+ + "very first row is emitted, regardless
of its content")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedValues(
+ Row.of("Vas", "login"),
+ Row.of("Alice", "click"),
+ Row.of("Bob", "view"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(Row.ofKind(RowKind.INSERT,
"Vas", "login"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST(input => TABLE user_events)")
+ .build();
+
+ public static final TableTestProgram RESET_WITH_ZERO_TTL =
+ TableTestProgram.of(
+ "deduplicate-keep-first-reset-with-zero-ttl",
+ "reset_ttl_on_duplicate=TRUE with
state_ttl=INTERVAL '0' (TTL "
+ + "disabled): only the first row per key
is emitted")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedValues(
+ Row.of("Vas", "login"),
+ Row.of("Vas", "click"),
+ Row.of("Vas", "view"),
+ Row.of("Vas", "logout"),
+ Row.of("Vas", "login"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(Row.ofKind(RowKind.INSERT,
"Vas", "login"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name, "
+ + "state_ttl => INTERVAL '0' SECOND, "
+ + "reset_ttl_on_duplicate => TRUE)")
+ .build();
+
+ public static final TableTestProgram NO_TTL_RESET =
+ TableTestProgram.of(
+ "deduplicate-keep-first-no-ttl-reset",
+ "reset_ttl_on_duplicate=FALSE still emits only the
first row per key")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedValues(
+ Row.of("Vas", "login"),
+ Row.of("Vas", "click"),
+ Row.of("Vas", "view"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(Row.ofKind(RowKind.INSERT,
"Vas", "login"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name, "
+ + "state_ttl => INTERVAL '5' SECOND, "
+ + "reset_ttl_on_duplicate => FALSE)")
+ .build();
+
+ public static final TableTestProgram WHOLE_ROW_KEY =
+ TableTestProgram.of(
+ "deduplicate-keep-first-whole-row-key",
+ "PARTITION BY over every column deduplicates
exact-duplicate rows, "
+ + "keeping one row per distinct payload")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedValues(
+ Row.of("Vas", "login"),
+ Row.of("Vas", "login"),
+ Row.of("Vas", "click"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "Vas",
"login"),
+ Row.ofKind(RowKind.INSERT, "Vas",
"click"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
(user_name, action))")
+ .build();
+
+ public static final TableTestProgram EVENT_TIME =
+ TableTestProgram.of(
+ "deduplicate-keep-first-event-time",
+ "with on_time, the row with the smallest event
time is kept per key, "
+ + "not the first to arrive")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(TIMED_EVENTS_SCHEMA)
+ .producedValues(
+ Row.of("Vas", "login",
Instant.ofEpochMilli(3000)),
+ Row.of("Vas", "click",
Instant.ofEpochMilli(1000)),
+ Row.of("Vas", "view",
Instant.ofEpochMilli(5000)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("user_name STRING", "action
STRING")
+ .consumedValues(Row.ofKind(RowKind.INSERT,
"Vas", "click"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT user_name, action FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name, "
+ + "on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram EVENT_TIME_CANDIDATE_FIELD_NAMES =
+ TableTestProgram.of(
+
"deduplicate-keep-first-event-time-candidate-field-names",
+ "with on_time, input columns named like the
buffered candidate's "
+ + "fields are emitted unchanged for the
earliest row")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(
+ "user_name STRING",
+ "payload STRING",
+ "event_time BIGINT",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'10' SECOND")
+ .producedValues(
+ Row.of("Vas", "login", 3L,
Instant.ofEpochMilli(3000)),
+ Row.of("Vas", "click", 1L,
Instant.ofEpochMilli(1000)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_name STRING",
+ "payload STRING",
+ "event_time BIGINT")
+ .consumedValues(Row.ofKind(RowKind.INSERT,
"Vas", "click", 1L))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT user_name, payload,
event_time "
+ + "FROM DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name, "
+ + "on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram EVENT_TIME_LATE_DROPPED =
+ TableTestProgram.of(
+ "deduplicate-keep-first-event-time-late-dropped",
+ "a row at or below the watermark is dropped, even
though its event "
+ + "time is earlier than the
already-finalized candidate; a late "
+ + "first row for an unseen key is dropped
without marking the "
+ + "key as seen")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(TIMED_EVENTS_SCHEMA)
+ .producedValues(
+ Row.of("Bob", "early",
Instant.ofEpochMilli(1000)),
+ Row.of("Bob", "advance",
Instant.ofEpochMilli(20000)),
+ Row.of("Bob", "late",
Instant.ofEpochMilli(500)),
+ Row.of("Carol", "late",
Instant.ofEpochMilli(500)),
+ Row.of(
+ "Dave",
+ "at-watermark",
+
Instant.ofEpochMilli(10000)),
+ Row.of("Dave", "on-time",
Instant.ofEpochMilli(15000)),
+ Row.of("Carol", "on-time",
Instant.ofEpochMilli(30000)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("user_name STRING", "action
STRING")
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "Bob",
"early"),
+ Row.ofKind(RowKind.INSERT, "Dave",
"on-time"),
+ Row.ofKind(RowKind.INSERT,
"Carol", "on-time"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT user_name, action FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name, "
+ + "on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram UPDATING_INPUT =
+ TableTestProgram.of(
+ "deduplicate-keep-first-updating-input",
+ "an updating input is deduplicated into an
insert-only result: a -D "
+ + "before the key is seen is ignored, the
first +I is emitted "
+ + "once and every later change (duplicate,
-U, +U, -D) is "
+ + "swallowed")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .addOption("changelog-mode", "I,UB,UA,D")
+ .producedValues(
+ Row.ofKind(RowKind.DELETE, "Vas",
"logout"),
+ Row.ofKind(RowKind.INSERT, "Vas",
"login"),
+ Row.ofKind(RowKind.INSERT, "Vas",
"login"),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"Vas", "login"),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Vas", "click"),
+ Row.ofKind(RowKind.DELETE, "Vas",
"click"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(Row.ofKind(RowKind.INSERT,
"Vas", "login"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name)")
+ .build();
+
+ public static final TableTestProgram UNSEEN_RETRACTION_IGNORED =
+ TableTestProgram.of(
+ "deduplicate-keep-first-unseen-retraction-ignored",
+ "a -U or -D for a key that has not been seen is
ignored and does not "
+ + "mark the key as seen, so the next +I or
+U for that key is "
+ + "emitted as an insert")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .addOption("changelog-mode", "I,UB,UA,D")
+ .producedValues(
+ Row.ofKind(RowKind.DELETE, "Vas",
"logout"),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"Bob", "login"),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Bob", "click"),
+ Row.ofKind(RowKind.INSERT, "Vas",
"login"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "Bob",
"click"),
+ Row.ofKind(RowKind.INSERT, "Vas",
"login"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name)")
+ .build();
+
+ public static final TableTestProgram UPDATING_EVENT_TIME_FAILS =
+ TableTestProgram.of(
+ "deduplicate-keep-first-updating-event-time-fails",
+ "event-time mode (on_time) is rejected for
updating input, because a "
+ + "watermark asserts no earlier record
will arrive while an "
+ + "update can invalidate that")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(TIMED_EVENTS_SCHEMA)
+ .addOption("changelog-mode", "I,UB,UA,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "Vas",
"login", null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("user_name STRING", "action
STRING")
+ .consumedValues(new Row[0])
+ .build())
+ .runFailingSql(
+ "INSERT INTO sink SELECT user_name, action FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name, "
+ + "on_time => DESCRIPTOR(ts))",
+ ValidationException.class,
+ "not supported for PTFs that consume or produce
updates")
+ .build();
+
+ public static final TableTestProgram NEGATIVE_TTL_FAILS =
+ invalidArgumentProgram(
+ "deduplicate-keep-first-negative-ttl-fails",
+ "a negative state_ttl is rejected at planning time",
+ "state_ttl => INTERVAL '-1' SECOND",
+ "The 'state_ttl' argument must not be negative");
+
+ public static final TableTestProgram NON_LITERAL_TTL_FAILS =
+ invalidArgumentProgram(
+ "deduplicate-keep-first-non-literal-ttl-fails",
+ "a state_ttl that is not a literal is rejected at planning
time",
+ "state_ttl => INTERVAL '1' SECOND * 2",
+ "The 'state_ttl' argument must be a constant INTERVAL
literal.");
+
+ public static final TableTestProgram NON_LITERAL_RESET_FAILS =
+ invalidArgumentProgram(
+ "deduplicate-keep-first-non-literal-reset-fails",
+ "a reset_ttl_on_duplicate that is not a literal is
rejected at planning time",
+ "reset_ttl_on_duplicate => 1 = 1",
+ "The 'reset_ttl_on_duplicate' argument must be a constant
BOOLEAN literal.");
+
+ private static TableTestProgram invalidArgumentProgram(
+ final String id, final String description, final String argument,
final String error) {
+ return TableTestProgram.of(id, description)
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedValues(Row.of("Vas", "login"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedValues(new Row[0])
+ .build())
+ .runFailingSql(
+ "INSERT INTO sink SELECT * FROM
DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY
user_name, "
+ + argument
+ + ")",
+ ValidationException.class,
+ error)
+ .build();
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/plan/deduplicate-keep-first-event-time-pending-restore.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/plan/deduplicate-keep-first-event-time-pending-restore.json
new file mode 100644
index 00000000000..0f0841b7164
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/plan/deduplicate-keep-first-event-time-pending-restore.json
@@ -0,0 +1,352 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 5,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`user_events`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "dataType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ],
+ "watermarkSpecs" : [ {
+ "rowtimeAttribute" : "ts",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$-$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "1",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "serializableString" : "`ts` - INTERVAL '0.001' SECOND"
+ }
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "WatermarkPushDown",
+ "watermarkExpr" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$-$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "1",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "rowtimeExpr" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "idleTimeoutMillis" : -1,
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "name" : "user_name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "watermarkParams" : {
+ "emitStrategy" : "ON_EVENT",
+ "alignGroupName" : null,
+ "alignMaxDrift" : "PT0S",
+ "alignUpdateInterval" : "PT1S",
+ "sourceIdleTimeout" : -1
+ }
+ } ]
+ },
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "user_name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, user_events, watermark=[-(ts, 1:INTERVAL SECOND)],
watermarkEmitStrategy=[on-event]]], fields=[user_name, action, ts])"
+ }, {
+ "id" : 6,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "user_name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "Exchange(distribution=[hash[user_name]])"
+ }, {
+ "id" : 7,
+ "type" : "stream-exec-process-table-function_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "user_name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ }, {
+ "name" : "rowtime",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" :
"ProcessTableFunction(invocation=[DEDUPLICATE_KEEP_FIRST(TABLE(#0) PARTITION
BY($0), DEFAULT(), DEFAULT(), DESCRIPTOR(_UTF-16LE'ts'), DEFAULT())],
uid=[DEDUPLICATE_KEEP_FIRST], select=[user_name,action,ts,rowtime],
rowType=[RecordType(VARCHAR(2147483647) user_name, VARCHAR(2147483647) action,
TIMESTAMP_LTZ(3) *ROWTIME* ts, TIMESTAMP_LTZ(3) *ROWTIME* rowtime)])",
+ "uid" : "DEDUPLICATE_KEEP_FIRST",
+ "functionCall" : {
+ "kind" : "CALL",
+ "internalName" : "$DEDUPLICATE_KEEP_FIRST$1",
+ "operands" : [ {
+ "kind" : "TABLE_ARG_CALL",
+ "inputIndex" : 0,
+ "partitionKeys" : [ 0 ],
+ "type" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "name" : "user_name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ }
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$DEFAULT$1",
+ "type" : "INTERVAL SECOND(6)"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$DEFAULT$1",
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "internalName" : "$DESCRIPTOR$1",
+ "operands" : [ {
+ "kind" : "LITERAL",
+ "value" : "ts",
+ "type" : "CHAR(2) NOT NULL"
+ } ],
+ "type" : "DESCRIPTOR NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$DEFAULT$1",
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "name" : "user_name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ }, {
+ "name" : "rowtime",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ }
+ },
+ "inputChangelogModes" : [ [ "INSERT" ] ],
+ "outputChangelogMode" : [ "INSERT" ],
+ "inputUpsertKeys" : [ [ ] ]
+ }, {
+ "id" : 8,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action`
VARCHAR(2147483647)>",
+ "description" : "Calc(select=[user_name, action])"
+ }, {
+ "id" : 9,
+ "type" : "stream-exec-sink_2",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT" ],
+ "upsertMaterializeStrategy" : "ADAPTIVE",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.sink],
fields=[user_name, action])"
+ } ],
+ "edges" : [ {
+ "source" : 5,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 6,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 7,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 8,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/savepoint/_metadata
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/savepoint/_metadata
new file mode 100644
index 00000000000..37b4d901412
Binary files /dev/null and
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/savepoint/_metadata
differ
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/plan/deduplicate-keep-first-keyed-restore.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/plan/deduplicate-keep-first-keyed-restore.json
new file mode 100644
index 00000000000..2204a83fdc8
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/plan/deduplicate-keep-first-keyed-restore.json
@@ -0,0 +1,143 @@
+{
+ "flinkVersion" : "2.5",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`user_events`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, user_events]], fields=[user_name, action])"
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[user_name]])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-process-table-function_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action`
VARCHAR(2147483647)>",
+ "description" :
"ProcessTableFunction(invocation=[DEDUPLICATE_KEEP_FIRST(TABLE(#0) PARTITION
BY($0), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())],
uid=[DEDUPLICATE_KEEP_FIRST], select=[user_name,action],
rowType=[RecordType(VARCHAR(2147483647) user_name, VARCHAR(2147483647)
action)])",
+ "uid" : "DEDUPLICATE_KEEP_FIRST",
+ "functionCall" : {
+ "kind" : "CALL",
+ "internalName" : "$DEDUPLICATE_KEEP_FIRST$1",
+ "operands" : [ {
+ "kind" : "TABLE_ARG_CALL",
+ "inputIndex" : 0,
+ "partitionKeys" : [ 0 ],
+ "type" : "ROW<`user_name` VARCHAR(2147483647), `action`
VARCHAR(2147483647)> NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$DEFAULT$1",
+ "type" : "INTERVAL SECOND(6)"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$DEFAULT$1",
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$DEFAULT$1",
+ "type" : "DESCRIPTOR"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$DEFAULT$1",
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "ROW<`user_name` VARCHAR(2147483647), `action`
VARCHAR(2147483647)> NOT NULL"
+ },
+ "inputChangelogModes" : [ [ "INSERT" ] ],
+ "outputChangelogMode" : [ "INSERT" ],
+ "inputUpsertKeys" : [ [ ] ]
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-sink_2",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "action",
+ "dataType" : "VARCHAR(2147483647)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT" ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action`
VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.sink],
fields=[user_name, action])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/savepoint/_metadata
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/savepoint/_metadata
new file mode 100644
index 00000000000..208bdc2d99f
Binary files /dev/null and
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/savepoint/_metadata
differ
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java
new file mode 100644
index 00000000000..7744f43a67f
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.ptf;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.ValueView;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.utils.ProjectedRowData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.runtime.functions.BuiltInSpecializedFunction;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.StateTypeStrategy;
+import
org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.RowKind;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static
org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy.ARG_INPUT;
+import static
org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy.ARG_STATE_TTL;
+
+/**
+ * Implementation of the {@code DEDUPLICATE_KEEP_FIRST} process table function.
+ *
+ * <p>Specializes into {@link ProcessingTime} or {@link EventTime} depending
on whether the input
+ * declares a time column, because only event time needs to buffer a candidate
row.
+ */
+@Internal
+public class DeduplicateKeepFirst extends BuiltInSpecializedFunction {
+
+ public DeduplicateKeepFirst() {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST);
+ }
+
+ @Override
+ public UserDefinedFunction specialize(final SpecializedContext context) {
+ return tableSemantics(context).timeColumn() == -1
+ ? new ProcessingTime(context)
+ : new EventTime(context);
+ }
+
+ private static TableSemantics tableSemantics(final SpecializedContext
context) {
+ return context.getCallContext()
+ .getTableSemantics(ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument
expected."));
+ }
+
+ private static StateTypeStrategy seenState(final SpecializedContext
context) {
+ return StateTypeStrategy.of(
+ callContext ->
+
Optional.of(ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull())),
+ context.getCallContext()
+ .getArgumentValue(ARG_STATE_TTL, Duration.class)
+ .orElse(null));
+ }
+
+ private static StateTypeStrategy candidateState(final SpecializedContext
context) {
+ final TableSemantics tableSemantics = tableSemantics(context);
+ final List<DataTypes.Field> inputFields =
DataType.getFields(tableSemantics.dataType());
+ final DataType payload =
+ DataTypes.ROW(
+ Arrays.stream(
+
ChangelogTypeStrategyUtils.computeOutputIndices(
+ tableSemantics))
+ .mapToObj(inputFields::get)
+ .toArray(DataTypes.Field[]::new));
+ final DataType candidate =
+ DataTypes.ROW(
+ DataTypes.FIELD("payload", payload),
+ DataTypes.FIELD("event_time", DataTypes.BIGINT()));
+ // The candidate buffer is cleared by its timer, not by retention;
disable its TTL so a
+ // global table.exec.state.ttl cannot expire it before the timer fires.
+ return StateTypeStrategy.of(
+ callContext ->
Optional.of(ValueView.newValueViewDataType(candidate)),
+ Duration.ZERO);
+ }
+
+ private abstract static class Base extends
BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ protected final int[] outputIndices;
+
+ protected final RowType inputRowType;
+
+ Base(
+ final SpecializedContext context,
+ final LinkedHashMap<String, StateTypeStrategy>
stateStrategies) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context,
stateStrategies);
+ final TableSemantics tableSemantics = tableSemantics(context);
+ this.outputIndices =
ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.inputRowType = (RowType)
tableSemantics.dataType().getLogicalType();
+ }
+ }
+
+ /** Processing-time mode: emits the first arriving row per key. */
+ @Internal
+ public static class ProcessingTime extends Base {
+
+ private static final long serialVersionUID = 1L;
+
+ private transient ProjectedRowData projectedOutput;
+
+ public ProcessingTime(final SpecializedContext context) {
+ super(context, internalState(Map.entry("seen",
seenState(context))));
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ }
+
+ public void eval(
+ final Context ctx,
+ final ValueView<Boolean> seen,
+ final RowData input,
+ @Nullable final Long stateTtl,
+ @Nullable final Boolean resetTtlOnDuplicate)
+ throws Exception {
+ // stateTtl is applied as the state's TTL at planning time.
+ final RowKind kind = input.getRowKind();
+ if (kind == RowKind.UPDATE_BEFORE || kind == RowKind.DELETE) {
+ return;
+ }
+
+ if (seen.getValue() == null) {
+ projectedOutput.replaceRow(input);
+ projectedOutput.setRowKind(RowKind.INSERT);
+ collect(projectedOutput);
+ seen.setValue(true);
+ } else if (resetTtlOnDuplicate == null || resetTtlOnDuplicate) {
+ seen.setValue(true);
+ }
+ }
+ }
+
+ /** Event-time mode: emits the row with the earliest event time per key. */
+ @Internal
+ public static class EventTime extends Base {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int CANDIDATE_PAYLOAD = 0;
+
+ private static final int CANDIDATE_EVENT_TIME = 1;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<Object>[] payloadSerializers;
+
+ public EventTime(final SpecializedContext context) {
+ super(
+ context,
+ internalState(
+ Map.entry("seen", seenState(context)),
+ Map.entry("candidate", candidateState(context))));
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ payloadSerializers = new TypeSerializer[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ final LogicalType payloadType =
inputRowType.getTypeAt(outputIndices[i]);
+ payloadGetters[i] = RowData.createFieldGetter(payloadType,
outputIndices[i]);
+ payloadSerializers[i] =
InternalSerializers.create(payloadType);
+ }
+ }
+
+ public void eval(
+ final Context ctx,
+ final ValueView<Boolean> seen,
+ final ValueView<RowData> candidate,
+ final RowData input,
+ @Nullable final Long stateTtl,
+ @Nullable final Boolean resetTtlOnDuplicate)
+ throws Exception {
+ final TimeContext<Long> context = ctx.timeContext(Long.class);
+ final Long rowtime = context.time();
+ final Long watermark = context.tableWatermark();
+ if (watermark != null && rowtime <= watermark) {
+ return;
+ }
+
+ if (seen.getValue() == null) {
+ final RowData currLowest = candidate.getValue();
+ final Long currentLowestEventTime =
+ currLowest == null ? null :
currLowest.getLong(CANDIDATE_EVENT_TIME);
+ if (currentLowestEventTime == null || rowtime <
currentLowestEventTime) {
+ if (currentLowestEventTime != null) {
+ context.clearTimer(currentLowestEventTime);
+ }
+ // object reuse may overwrite the input's memory before
the timer fires
+ candidate.setValue(materializeCandidate(input, rowtime));
+ context.registerOnTime(rowtime);
+ }
+ } else if (resetTtlOnDuplicate == null || resetTtlOnDuplicate) {
+ seen.setValue(true);
+ }
+ }
+
+ public void onTimer(
+ final OnTimerContext ctx,
+ final ValueView<Boolean> seen,
+ final ValueView<RowData> candidate)
+ throws Exception {
+ final RowData currLowest = candidate.getValue();
+ if (currLowest == null) {
+ return;
+ }
+ collect(currLowest.getRow(CANDIDATE_PAYLOAD,
outputIndices.length));
+ candidate.clear();
+ seen.setValue(true);
+ }
+
+ private GenericRowData materializeCandidate(final RowData input, final
long rowtime) {
+ final GenericRowData payload = new
GenericRowData(outputIndices.length);
+ for (int i = 0; i < outputIndices.length; i++) {
+ final Object field = payloadGetters[i].getFieldOrNull(input);
+ payload.setField(i, field == null ? null :
payloadSerializers[i].copy(field));
+ }
+ return GenericRowData.of(payload, rowtime);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirstTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirstTest.java
new file mode 100644
index 00000000000..8d82d969980
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirstTest.java
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.ptf;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.ValueView;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinition;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import
org.apache.flink.table.functions.SpecializedFunction.ExpressionEvaluator;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.StateTypeStrategy;
+import org.apache.flink.table.types.inference.utils.CallContextMock;
+import org.apache.flink.table.types.inference.utils.TableSemanticsMock;
+import org.apache.flink.table.types.utils.DataTypeFactoryMock;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the state that each specialization of {@link
DeduplicateKeepFirst} declares. */
+class DeduplicateKeepFirstTest {
+
+ private static final DataType INPUT_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("user_name", DataTypes.STRING()),
+ DataTypes.FIELD("action", DataTypes.STRING()),
+ DataTypes.FIELD("ts", DataTypes.TIMESTAMP_LTZ(3)));
+
+ private static final DataType OUTPUT_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("user_name", DataTypes.STRING()),
+ DataTypes.FIELD("action", DataTypes.STRING()),
+ DataTypes.FIELD("ts", DataTypes.TIMESTAMP_LTZ(3)))
+ .notNull();
+
+ @Test
+ void testProcessingTimeDeclaresOnlySeenState() {
+ final CallContextMock callContext = callContext(-1,
Duration.ofSeconds(5));
+ final UserDefinedFunction function = specialize(callContext);
+
+
assertThat(function).isInstanceOf(DeduplicateKeepFirst.ProcessingTime.class);
+ final Map<String, StateTypeStrategy> state = stateStrategies(function);
+ assertThat(state).containsOnlyKeys("seen");
+ assertThat(state.get("seen").inferType(callContext))
+
.contains(ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+
assertThat(state.get("seen").getTimeToLive(callContext)).contains(Duration.ofSeconds(5));
+
+ final CallContextMock withoutTtl = callContext(-1, null);
+
assertThat(stateStrategies(specialize(withoutTtl)).get("seen").getTimeToLive(withoutTtl))
+ .isEmpty();
+ }
+
+ @Test
+ void testEventTimeDeclaresSeenAndCandidateState() {
+ final CallContextMock callContext = callContext(2,
Duration.ofSeconds(5));
+ final UserDefinedFunction function = specialize(callContext);
+
+
assertThat(function).isInstanceOf(DeduplicateKeepFirst.EventTime.class);
+ final Map<String, StateTypeStrategy> state = stateStrategies(function);
+ assertThat(state.keySet()).containsExactly("seen", "candidate");
+
assertThat(state.get("seen").getTimeToLive(callContext)).contains(Duration.ofSeconds(5));
+
+ final DataType candidate =
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "payload",
+ DataTypes.ROW(
+ DataTypes.FIELD("action",
DataTypes.STRING()),
+ DataTypes.FIELD("ts",
DataTypes.TIMESTAMP_LTZ(3)))),
+ DataTypes.FIELD("event_time", DataTypes.BIGINT()));
+ assertThat(state.get("candidate").inferType(callContext))
+ .contains(
+ ValueView.newValueViewDataType(
+ DataTypeUtils.toInternalDataType(candidate)));
+ // The timer, not retention, clears the candidate, so state_ttl must
not apply to it.
+
assertThat(state.get("candidate").getTimeToLive(callContext)).contains(Duration.ZERO);
+ }
+
+ private static Map<String, StateTypeStrategy> stateStrategies(
+ final UserDefinedFunction function) {
+ return function.getTypeInference(new
DataTypeFactoryMock()).getStateTypeStrategies();
+ }
+
+ private static UserDefinedFunction specialize(final CallContext
callContext) {
+ return new DeduplicateKeepFirst()
+ .specialize(
+ new SpecializedContext() {
+ @Override
+ public CallContext getCallContext() {
+ return callContext;
+ }
+
+ @Override
+ public ReadableConfig getConfiguration() {
+ return new Configuration();
+ }
+
+ @Override
+ public ClassLoader getBuiltInClassLoader() {
+ return
DeduplicateKeepFirstTest.class.getClassLoader();
+ }
+
+ @Override
+ public ExpressionEvaluator createEvaluator(
+ Expression expression,
+ DataType outputDataType,
+ DataTypes.Field... args) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExpressionEvaluator createEvaluator(
+ String sqlExpression,
+ DataType outputDataType,
+ DataTypes.Field... args) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExpressionEvaluator createEvaluator(
+ BuiltInFunctionDefinition function,
+ DataType outputDataType,
+ DataType... args) {
+ throw new UnsupportedOperationException();
+ }
+ });
+ }
+
+ private static CallContextMock callContext(final int timeColumn, final
Duration stateTtl) {
+ final CallContextMock callContext = new CallContextMock();
+ callContext.typeFactory = new DataTypeFactoryMock();
+ callContext.functionDefinition =
BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST;
+ callContext.argumentDataTypes =
+ List.of(INPUT_TYPE, DataTypes.INTERVAL(DataTypes.SECOND()),
DataTypes.BOOLEAN());
+ callContext.argumentValues =
+ Arrays.asList(Optional.empty(), Optional.ofNullable(stateTtl),
Optional.empty());
+ callContext.outputDataType = Optional.of(OUTPUT_TYPE);
+ callContext.tableSemantics =
+ Map.of(
+ 0,
+ new TableSemanticsMock(
+ INPUT_TYPE, new int[] {0}, new int[0],
timeColumn, null));
+ return callContext;
+ }
+}
