[FLINK-40896][table]Deduplicate keep first implementation by VasShabu · Pull Request #29365 · apache/flink

VasShabu · 2026-10-02T13:40:11Z

What is the purpose of the change

Implement a new PTF for deduplication of a stream, currently Flink does not support such method, however people are able to get similar functionality with long and verbose SQL statements, this flip is to make this a mainline function and

Brief change log

Register the built-in DEDUPLICATE_KEEP_FIRST process table function in BuiltInFunctionDefinitions
Add the DeduplicateKeepFirst runtime implementation and its type strategy
Add docs and tests

Verifying this change

Added DeduplicateKeepFirstTypeStrategyTest for input validation and output type inference
Added DeduplicateKeepFirstSemanticTests for the function's behaviour
Added DeduplicateKeepFirstRestoreTest for state restore from a savepoint

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (don't know)
The runtime per-record code paths (performance sensitive): (yes)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (don't know)
The S3 file system connector: (don't know)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (docs)

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: [Claude Opus 4.8]

flinkbot · 2026-10-02T13:46:25Z

CI report:

b328f53 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

gustavodemorais

Thanks for the PR containing the full implementation, @VasShabu! This is definitely a very nice feature.

Before going into the detailed review: I wanted to suggest you to extend testing. We already have keep-first deduplication in operators/deduplicate. Could you take a look at ProcTimeDeduplicateKeepFirstRowFunctionTest (TTL expiry), RowTimeDeduplicateFunctionTest (rowtime + updating input) and DeduplicateITCase, and check which tests are still missing for the cases we support so we don't miss on coverage?

gustavodemorais · 2026-10-02T14:33:54Z

+        } else { // watermarkless mode
+            if (seen.getValue() == null) {
+                projectedOutput.replaceRow(input);
+                projectedOutput.setRowKind(RowKind.INSERT);


You're unconditionally accepting the first change for a key as the first appearance. Edge case: the first change for a key can be a deletion. Imagine we start reading a kafka topic a later offset and only see the deletion but not the original event.

-U and -D should never count as "first" because they only retract a row. Returning early on them means only +I or +U can become the kept row.

snuyanzin · 2026-10-02T15:36:00Z

+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class DeduplicateKeepFirstTypeStrategy {


seems
@Internal and javadoc are still missing

snuyanzin · 2026-10-02T15:37:18Z

+
+import java.util.stream.IntStream;
+
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {


@Internal, javadoc

twalthr · 2026-10-07T14:23:35Z

+-- +I[user_name:'Vas', action:'login']
+```
+
+The input may be an updating changelog. `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 next `+I` or `+U` for that key is still emitted. Event-time mode (`on_time`) is not supported with updating input.


link "updating changelog" to a page, currently PTF page until Gustavos PR is in

people might not know +I or -U

twalthr · 2026-10-07T14:25:59Z

+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+  input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+  [on_time => DESCRIPTOR(rowtime_column),]


we should also document "uid" argument that is required when more than one dedup is used. btw system args are at the end of the signature, this is important if positional args instead f named args are used.

twalthr · 2026-10-07T14:55:15Z

+                                    .map(Field::getName)
+                                    .collect(Collectors.toSet());
+                    String timestampField = "event_time";
+                    for (int i = 0; takenNames.contains(timestampField); i++) {


can we just solve duplicates by x$0, x$1, etc?

or even better: just nest one more time

twalthr · 2026-10-07T15:01:27Z

+                        timestampField = "event_time_" + i;
+                    }
+                    candidateFields.add(DataTypes.FIELD(timestampField, DataTypes.BIGINT()));
+                    return Optional.of(


We don't always need this state right?

We don't always need it no, however I changed its from a row to a boolean in the case when we are dealing with watermarkless mode as I dont think that we can completely remove it outright.

twalthr

Thank you @VasShabu. I did a full review now and left some comments.

twalthr · 2026-10-15T16:12:14Z

+            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 'table.exec.state.ttl'.",
+                    stateTtl.get());


Suggested change

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 'table.exec.state.ttl'.",

stateTtl.get());

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());

twalthr · 2026-10-15T16:13:36Z

+                    // Only event-time mode buffers a candidate. Without on_time a fixed type keeps
+                    // the unused state independent of the input schema across query evolution.
+                    if (tableSemantics.timeColumn() == -1) {
+                        return Optional.of(ValueView.newValueViewDataType(DataTypes.BOOLEAN()));


let's not use boolean but a correct type, allows for switching the mode in the future. it doesn't add any additional state.

sounds good will change accordingly

twalthr · 2026-10-15T16:39:04Z

+                    candidate.setValue(
+                            candidateSerializer.copy(materializeCandidate(input, rowtime)));


this should not be necessary, are there better ways? we have copied the row already. why do we need to serialize again, this would be expensive.

twalthr · 2026-10-15T16:40:22Z

+                }
+            } else { // already emitted → drop, optionally refresh the TTL
+                if (resetTtlOnDuplicate == null || resetTtlOnDuplicate) {
+                    seen.setValue(true);


did you verify that this indeed causes resetTtlOnDuplicate?

Yeah I have confirmed that this work with claude + with the flink sqlclient

…e function

…addressed -u -d issue, @internal etc

…docs

…didate state - Drop DEDUPLICATE_KEEP_FIRST rows with ts <= watermark, use one candidate state type in both modes, and derive the event-time mode once in the constructor - Trim the tests (drop Table API, multi-key and TTL-expiry cases, add restore rows) and polish the docs

…go into different functions given the input args

gustavodemorais reviewed Oct 2, 2026

View reviewed changes