You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "VasShabu (via GitHub)" <gi...@apache.org> on 2026/10/02 13:40:14 UTC
[PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
VasShabu opened a new pull request, #29365:
URL: https://github.com/apache/flink/pull/29365
## 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?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change the checkbox below to `[X]` and replace the placeholder in the "Generated-by"
line with the tool name and version. Otherwise remove the "Generated-by" line.
See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed. See
AGENTS.md for the full guidance.
-->
- [X] Yes (please specify the tool below)
Generated-by: [Claude Opus 4.8]
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4167279018
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.api.common.typeutils.TypeSerializer;
+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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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.util.stream.IntStream;
+
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
Review Comment:
`@Internal`, javadoc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4256930161
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+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;
+ private static final String SEEN_STATE_NAME = "seen";
+ private static final String CANDIDATE_STATE_NAME = "candidate";
+
+ 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());
+ };
+
+ public static final StateTypeStrategy SEEN_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ return Optional.of(
+ ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ return callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ }
+ };
+
+ public static final StateTypeStrategy CANDIDATE_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext 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> candidateFields =
+ Arrays.stream(
+ ChangelogTypeStrategyUtils.computeOutputIndices(
+ tableSemantics))
+ .mapToObj(inputFields::get)
+ .collect(Collectors.toCollection(ArrayList::new));
+ // The trailing timestamp field is read positionally at runtime, so its name
+ // only needs to avoid clashing with a payload column of the same name.
+ final Set<String> takenNames =
+ candidateFields.stream()
+ .map(Field::getName)
+ .collect(Collectors.toSet());
+ String timestampField = "event_time";
+ for (int i = 0; takenNames.contains(timestampField); i++) {
+ timestampField = "event_time_" + i;
+ }
+ candidateFields.add(DataTypes.FIELD(timestampField, DataTypes.BIGINT()));
+ return Optional.of(
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4167267594
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.table.api.DataTypes;
+import org.apache.flink.table.api.DataTypes.Field;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class DeduplicateKeepFirstTypeStrategy {
Review Comment:
seems
`@Internal` and javadoc are still missing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313333244
##########
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. `INTERVAL '0'` also retains state indefinitely, but excludes the TTL from the state layout, so a retention cannot be introduced later without losing the state. See [State TTL]({{< ref "docs/dev/table/functions/ptfs" >}}#state-ttl). |
Review Comment:
sure done.
##########
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. `INTERVAL '0'` also retains state indefinitely, but excludes the TTL from the state layout, so a retention cannot be introduced later without losing the state. See [State TTL]({{< ref "docs/dev/table/functions/ptfs" >}}#state-ttl). |
Review Comment:
sure. done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313367112
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java:
##########
@@ -0,0 +1,560 @@
+/*
+ * 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.Duration;
+import java.time.Instant;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+
+/** {@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 TABLE_API =
Review Comment:
sounds good, dropped
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on PR #29365:
URL: https://github.com/apache/flink/pull/29365#issuecomment-6444453097
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr merged PR #29365:
URL: https://github.com/apache/flink/pull/29365
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4207880641
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
Review Comment:
```suggestion
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.
```
##########
docs/content.zh/docs/sql/functions/built-in-functions.md:
##########
@@ -147,6 +147,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 insert-only stream (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. |
Review Comment:
```suggestion
| `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. |
```
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
Review Comment:
```suggestion
The input may be an append-only or updating table; the output is append-only in every case.
```
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>]
+)
+```
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (insert-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. |
Review Comment:
```suggestion
| `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. |
```
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (first arrival, or earliest event time with `on_time`) |
Review Comment:
```suggestion
| [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`) |
```
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>]
+)
+```
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (insert-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. |
+| `on_time` | No | A `DESCRIPTOR` naming a single rowtime attribute. 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 insert-only input; combining `on_time` with an updating input is rejected at planning time. |
+| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for the per-key deduplication state. If omitted, falls back to `table.exec.state.ttl`, which retains state indefinitely at its default of 0. After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. `INTERVAL '0'` disables retention. |
Review Comment:
> `INTERVAL '0'` disables retention
is this true? 0 in PTFs should mean we disable TTL and since the state layout does not even contain a timestamp, one can not enabled it as part of an evolution step.
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>]
+)
+```
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (insert-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. |
+| `on_time` | No | A `DESCRIPTOR` naming a single rowtime attribute. 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 insert-only input; combining `on_time` with an updating input is rejected at planning time. |
+| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for the per-key deduplication state. If omitted, falls back to `table.exec.state.ttl`, which retains state indefinitely at its default of 0. After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. `INTERVAL '0'` disables retention. |
+| `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`. |
+
+### 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.
Review Comment:
```suggestion
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.
```
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
Review Comment:
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.
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>]
+)
+```
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (insert-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. |
+| `on_time` | No | A `DESCRIPTOR` naming a single rowtime attribute. 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 insert-only input; combining `on_time` with an updating input is rejected at planning time. |
+| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for the per-key deduplication state. If omitted, falls back to `table.exec.state.ttl`, which retains state indefinitely at its default of 0. After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. `INTERVAL '0'` disables retention. |
+| `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`. |
+
+### 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.
+
+```
+[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.
+
+#### Whole input, no PARTITION BY
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Alice', action:'click']
+-- +I[user_name:'Bob', action:'view']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events
Review Comment:
we should rather partition by all columns in this case, don't show examples that are difficult to scale.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java:
##########
@@ -142,6 +142,10 @@ public static InputTypeStrategy plainJsonPath(final InputTypeStrategy signatures
public static final InputTypeStrategy FROM_CHANGELOG_INPUT_TYPE_STRATEGY =
FromChangelogTypeStrategy.INPUT_TYPE_STRATEGY;
+ /** Input strategy for {@link BuiltInFunctionDefinitions#DEDUPLICATE_KEEP_FIRST}. */
+ public static final InputTypeStrategy DEDUPLICATE_KEEP_FIRST_INPUT_TYPE_STRATEGY =
Review Comment:
No need to add it here.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+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;
+ private static final String SEEN_STATE_NAME = "seen";
+ private static final String CANDIDATE_STATE_NAME = "candidate";
+
+ 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());
+ };
+
+ public static final StateTypeStrategy SEEN_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ return Optional.of(
+ ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ return callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ }
+ };
+
+ public static final StateTypeStrategy CANDIDATE_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext 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> candidateFields =
+ Arrays.stream(
+ ChangelogTypeStrategyUtils.computeOutputIndices(
+ tableSemantics))
+ .mapToObj(inputFields::get)
+ .collect(Collectors.toCollection(ArrayList::new));
+ // The trailing timestamp field is read positionally at runtime, so its name
+ // only needs to avoid clashing with a payload column of the same name.
+ final Set<String> takenNames =
+ candidateFields.stream()
+ .map(Field::getName)
+ .collect(Collectors.toSet());
+ String timestampField = "event_time";
+ for (int i = 0; takenNames.contains(timestampField); i++) {
Review Comment:
can we just solve duplicates by x$0, x$1, etc?
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>]
+)
+```
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (insert-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. |
+| `on_time` | No | A `DESCRIPTOR` naming a single rowtime attribute. 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 insert-only input; combining `on_time` with an updating input is rejected at planning time. |
+| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for the per-key deduplication state. If omitted, falls back to `table.exec.state.ttl`, which retains state indefinitely at its default of 0. After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. `INTERVAL '0'` disables retention. |
+| `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`. |
+
+### 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.
+
+```
+[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.
+
+#### Whole input, no PARTITION BY
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Alice', action:'click']
+-- +I[user_name:'Bob', action:'view']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+```
+
+Without `PARTITION BY` the whole input is a single group at parallelism 1, so only the first row of the entire stream is emitted; every later row is dropped regardless of its content.
+
+#### 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.
+
+#### 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. `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.
Review Comment:
people might not know +I or -U
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+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;
+ private static final String SEEN_STATE_NAME = "seen";
+ private static final String CANDIDATE_STATE_NAME = "candidate";
+
+ 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());
+ };
+
+ public static final StateTypeStrategy SEEN_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ return Optional.of(
+ ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ return callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ }
+ };
+
+ public static final StateTypeStrategy CANDIDATE_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext 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> candidateFields =
+ Arrays.stream(
+ ChangelogTypeStrategyUtils.computeOutputIndices(
+ tableSemantics))
+ .mapToObj(inputFields::get)
+ .collect(Collectors.toCollection(ArrayList::new));
+ // The trailing timestamp field is read positionally at runtime, so its name
+ // only needs to avoid clashing with a payload column of the same name.
+ final Set<String> takenNames =
+ candidateFields.stream()
+ .map(Field::getName)
+ .collect(Collectors.toSet());
+ String timestampField = "event_time";
+ for (int i = 0; takenNames.contains(timestampField); i++) {
Review Comment:
or even better: just nest one more time
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>]
+)
+```
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (insert-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. |
+| `on_time` | No | A `DESCRIPTOR` naming a single rowtime attribute. 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 insert-only input; combining `on_time` with an updating input is rejected at planning time. |
Review Comment:
```suggestion
| `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. |
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+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;
+ private static final String SEEN_STATE_NAME = "seen";
+ private static final String CANDIDATE_STATE_NAME = "candidate";
+
+ 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());
+ };
+
+ public static final StateTypeStrategy SEEN_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ return Optional.of(
+ ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ return callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ }
+ };
+
+ public static final StateTypeStrategy CANDIDATE_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext 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> candidateFields =
+ Arrays.stream(
+ ChangelogTypeStrategyUtils.computeOutputIndices(
+ tableSemantics))
+ .mapToObj(inputFields::get)
+ .collect(Collectors.toCollection(ArrayList::new));
+ // The trailing timestamp field is read positionally at runtime, so its name
+ // only needs to avoid clashing with a payload column of the same name.
+ final Set<String> takenNames =
+ candidateFields.stream()
+ .map(Field::getName)
+ .collect(Collectors.toSet());
+ String timestampField = "event_time";
+ for (int i = 0; takenNames.contains(timestampField); i++) {
+ timestampField = "event_time_" + i;
+ }
+ candidateFields.add(DataTypes.FIELD(timestampField, DataTypes.BIGINT()));
+ return Optional.of(
Review Comment:
We don't always need this state right?
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>]
+)
+```
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (insert-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. |
+| `on_time` | No | A `DESCRIPTOR` naming a single rowtime attribute. 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 insert-only input; combining `on_time` with an updating input is rejected at planning time. |
+| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for the per-key deduplication state. If omitted, falls back to `table.exec.state.ttl`, which retains state indefinitely at its default of 0. After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. `INTERVAL '0'` disables retention. |
Review Comment:
so there is a difference between 0 and optional. Optional meaning one can still change it later. 0 disabled for ever.
##########
docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md:
##########
@@ -0,0 +1,187 @@
+---
+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, append-in / append-out alternative to the `ROW_NUMBER()` [deduplication]({{< ref "docs/sql/reference/queries/deduplication" >}}) pattern.
+
+| Function | Description |
+|:---------|:------------|
+| [DEDUPLICATE_KEEP_FIRST](#deduplicate_keep_first) | Keeps the first record per key as an insert-only stream (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 insert-only or updating; the output is insert-only in every case.
+
+### Syntax
+
+```sql
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE source_table [PARTITION BY key_col [, key_col2 ...]],
+ [on_time => DESCRIPTOR(rowtime_column),]
+ [state_ttl => <interval>,]
+ [reset_ttl_on_duplicate => <boolean>]
+)
+```
+
+### Parameters
+
+| Parameter | Required | Description |
+|:----------|:---------|:------------|
+| `input` | Yes | The input table (insert-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. |
+| `on_time` | No | A `DESCRIPTOR` naming a single rowtime attribute. 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 insert-only input; combining `on_time` with an updating input is rejected at planning time. |
+| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for the per-key deduplication state. If omitted, falls back to `table.exec.state.ttl`, which retains state indefinitely at its default of 0. After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. `INTERVAL '0'` disables retention. |
+| `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`. |
+
+### 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.
+
+```
+[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.
+
+#### Whole input, no PARTITION BY
+
+```sql
+-- Input (append-only):
+-- +I[user_name:'Vas', action:'login']
+-- +I[user_name:'Alice', action:'click']
+-- +I[user_name:'Bob', action:'view']
+
+SELECT * FROM DEDUPLICATE_KEEP_FIRST(
+ input => TABLE user_events
+)
+
+-- Output (insert-only):
+-- +I[user_name:'Vas', action:'login']
+```
+
+Without `PARTITION BY` the whole input is a single group at parallelism 1, so only the first row of the entire stream is emitted; every later row is dropped regardless of its content.
+
+#### 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.
+
+#### 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. `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.
Review Comment:
link "updating changelog" to a page, currently PTF page until Gustavos PR is in
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4314968423
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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;
+
+/** Implementation of the {@code DEDUPLICATE_KEEP_FIRST} process table function. */
+@Internal
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int CANDIDATE_PAYLOAD = 0;
+
+ private static final int CANDIDATE_EVENT_TIME = 1;
+
+ private final int[] outputIndices;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ final LogicalType[] payloadTypes = new LogicalType[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ payloadGetters[i] = RowData.createFieldGetter(payloadTypes[i], outputIndices[i]);
+ }
+ candidateSerializer =
+ InternalSerializers.create(RowType.of(RowType.of(payloadTypes), new BigIntType()));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
+ // a parameter for every declared argument, so it is unused here.
+ final RowKind kind = input.getRowKind();
+ if (kind == RowKind.UPDATE_BEFORE || kind == RowKind.DELETE) {
+ return;
+ }
+
+ final TimeContext<Long> context = ctx.timeContext(Long.class);
+ final Long rowtime = context.time();
+
+ if (rowtime != null) { // watermark mode
+ final Long watermark = context.tableWatermark();
+ if (watermark != null && rowtime < watermark) {
+ return;
+ }
+
+ if (seen.getValue() == null) { // not yet emitted → buffer the earliest candidate
+ 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(
+ candidateSerializer.copy(materializeCandidate(input, rowtime)));
+ context.registerOnTime(rowtime);
+ }
+ } else { // already emitted → drop, optionally refresh the TTL
+ if (resetTtlOnDuplicate == null || resetTtlOnDuplicate) {
+ seen.setValue(true);
Review Comment:
Yeah I have confirmed that this work with claude + with the flink sqlclient
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4166714940
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.api.common.typeutils.TypeSerializer;
+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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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.util.stream.IntStream;
+
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int[] outputIndices;
+
+ private final int eventTimeIndex;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient ProjectedRowData emitProjection;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.eventTimeIndex = outputIndices.length;
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ emitProjection = ProjectedRowData.from(IntStream.range(0, eventTimeIndex).toArray());
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadGetters[i] =
+ RowData.createFieldGetter(
+ inputRowType.getTypeAt(outputIndices[i]), outputIndices[i]);
+ }
+
+ final LogicalType[] candidateTypes = new LogicalType[eventTimeIndex + 1];
+ for (int i = 0; i < eventTimeIndex; i++) {
+ candidateTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ }
+ candidateTypes[eventTimeIndex] = new BigIntType();
+ candidateSerializer = InternalSerializers.create(RowType.of(candidateTypes));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
+ // a parameter for every declared argument, so it is unused here.
+ final Long rowtime = ctx.timeContext(Long.class).time();
+
+ if (rowtime != null) { // watermark mode
+ final Long watermark = ctx.timeContext(Long.class).tableWatermark();
+ if (watermark != null && rowtime < watermark) {
+ return;
+ }
+
+ if (seen.getValue() == null) { // not yet emitted → buffer the earliest candidate
+ final RowData currLowest = candidate.getValue();
+ if (currLowest == null || rowtime < currLowest.getLong(eventTimeIndex)) {
+ if (currLowest != null) {
+ ctx.timeContext(Long.class).clearTimer(currLowest.getLong(eventTimeIndex));
+ }
+ // object reuse may overwrite the input's memory before the timer fires
+ candidate.setValue(
+ candidateSerializer.copy(materializeCandidate(input, rowtime)));
+ ctx.timeContext(Long.class).registerOnTime(rowtime);
+ }
+ } else { // already emitted → drop, optionally refresh the TTL
+ if (resetTtlOnDuplicate == null || resetTtlOnDuplicate) {
+ seen.setValue(true);
+ }
+ }
+
+ } else { // watermarkless mode
+ if (seen.getValue() == null) {
+ projectedOutput.replaceRow(input);
+ projectedOutput.setRowKind(RowKind.INSERT);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4167222267
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java:
##########
@@ -142,6 +142,10 @@ public static InputTypeStrategy plainJsonPath(final InputTypeStrategy signatures
public static final InputTypeStrategy FROM_CHANGELOG_INPUT_TYPE_STRATEGY =
FromChangelogTypeStrategy.INPUT_TYPE_STRATEGY;
+ /** Input strategy for {@link BuiltInFunctionDefinitions#DEDUPLICATE_KEEP_FIRST}. */
+ public static final InputTypeStrategy DEDUPLICATE_KEEP_FIRST_TYPE_STRATEGY =
Review Comment:
```suggestion
public static final InputTypeStrategy DEDUPLICATE_KEEP_FIRST_INPUT_TYPE_STRATEGY =
```
please follow existing pattern in namings
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4389032852
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstSemanticTests.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+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
+ protected void applyDefaultEnvironmentOptions(TableConfig config) {
+ super.applyDefaultEnvironmentOptions(config);
+ config.set(
Review Comment:
I used FROM_CHANGELOG as a template for the semantic test file, so was just copied over as boilerplate.
Later on FROM_CHANGELOG removed this setting
https://github.com/apache/flink/pull/29378
Will drop it from keep_first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4167206721
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.api.common.typeutils.TypeSerializer;
+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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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.util.stream.IntStream;
+
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int[] outputIndices;
+
+ private final int eventTimeIndex;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient ProjectedRowData emitProjection;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.eventTimeIndex = outputIndices.length;
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ emitProjection = ProjectedRowData.from(IntStream.range(0, eventTimeIndex).toArray());
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadGetters[i] =
+ RowData.createFieldGetter(
+ inputRowType.getTypeAt(outputIndices[i]), outputIndices[i]);
+ }
+
+ final LogicalType[] candidateTypes = new LogicalType[eventTimeIndex + 1];
+ for (int i = 0; i < eventTimeIndex; i++) {
+ candidateTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ }
+ candidateTypes[eventTimeIndex] = new BigIntType();
+ candidateSerializer = InternalSerializers.create(RowType.of(candidateTypes));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
+ // a parameter for every declared argument, so it is unused here.
+ final Long rowtime = ctx.timeContext(Long.class).time();
+
+ if (rowtime != null) { // watermark mode
+ final Long watermark = ctx.timeContext(Long.class).tableWatermark();
+ if (watermark != null && rowtime < watermark) {
+ return;
+ }
+
+ if (seen.getValue() == null) { // not yet emitted → buffer the earliest candidate
+ final RowData currLowest = candidate.getValue();
+ if (currLowest == null || rowtime < currLowest.getLong(eventTimeIndex)) {
Review Comment:
same: reading same value twice in hot path
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4388551190
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstSemanticTests.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+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
+ protected void applyDefaultEnvironmentOptions(TableConfig config) {
+ super.applyDefaultEnvironmentOptions(config);
+ config.set(
Review Comment:
What is the reason for this setting?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java:
##########
@@ -0,0 +1,480 @@
+/*
+ * 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 EVENT_TIME_EMITTED_KEY_DROPS_LATER_ROWS =
Review Comment:
This one is already covered in the case above. We should not add to many tests, therefore I would suggest to drop.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313383792
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java:
##########
@@ -0,0 +1,560 @@
+/*
+ * 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.Duration;
+import java.time.Instant;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+
+/** {@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 TABLE_API =
+ TableTestProgram.of(
+ "deduplicate-keep-first-table-api",
+ "the generic Table API process() call invokes the PTF, matching "
+ + "the SQL form")
+ .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())
+ .runTableApi(
+ env ->
+ env.from("user_events")
+ .partitionBy($("user_name"))
+ .process("DEDUPLICATE_KEEP_FIRST"),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_WITH_ARGS =
+ TableTestProgram.of(
+ "deduplicate-keep-first-table-api-with-args",
+ "the Table API process() call passes the optional state_ttl and "
+ + "reset_ttl_on_duplicate as named arguments")
+ .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())
+ .runTableApi(
+ env ->
+ env.from("user_events")
+ .partitionBy($("user_name"))
+ .process(
+ "DEDUPLICATE_KEEP_FIRST",
+ lit(Duration.ofSeconds(5))
+ .asArgument("state_ttl"),
+ lit(false)
+ .asArgument("reset_ttl_on_duplicate")),
+ "sink")
+ .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("region STRING", "user_name STRING", "action STRING")
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "EU", "Vas", "login"),
+ Row.ofKind(RowKind.INSERT, "US", "Vas", "signup"),
+ Row.ofKind(RowKind.INSERT, "EU", "Alice", "view"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY (region, user_name))")
+ .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_MULTI_KEY =
Review Comment:
dropped
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4167017843
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.api.common.typeutils.TypeSerializer;
+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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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.util.stream.IntStream;
+
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int[] outputIndices;
+
+ private final int eventTimeIndex;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient ProjectedRowData emitProjection;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.eventTimeIndex = outputIndices.length;
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ emitProjection = ProjectedRowData.from(IntStream.range(0, eventTimeIndex).toArray());
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadGetters[i] =
+ RowData.createFieldGetter(
+ inputRowType.getTypeAt(outputIndices[i]), outputIndices[i]);
+ }
+
+ final LogicalType[] candidateTypes = new LogicalType[eventTimeIndex + 1];
+ for (int i = 0; i < eventTimeIndex; i++) {
+ candidateTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ }
+ candidateTypes[eventTimeIndex] = new BigIntType();
+ candidateSerializer = InternalSerializers.create(RowType.of(candidateTypes));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
+ // a parameter for every declared argument, so it is unused here.
+ final Long rowtime = ctx.timeContext(Long.class).time();
Review Comment:
it is runtime code
is there a reason we have multiple calls to `ctx.timeContext(Long.class)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313362532
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.config.ExecutionConfigOptions;
+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.Duration;
+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)))
+ .producedAfterRestore(
+ Row.of("Bob", "later", Instant.ofEpochMilli(100)))
Review Comment:
added three rows before the restore and emits all three after it.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.config.ExecutionConfigOptions;
+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.Duration;
+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)))
+ .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"))
+ .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 TTL_EXPIRED_RESTORE =
+ TableTestProgram.of(
+ "deduplicate-keep-first-ttl-expired-restore",
+ "the 'seen' state in the savepoint has outlived state_ttl by the time "
+ + "the test restores it, so an already-seen key is emitted "
Review Comment:
dropped
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4285320995
##########
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. `INTERVAL '0'` also retains state indefinitely, but excludes the TTL from the state layout, so a retention cannot be introduced later without losing the state. See [State TTL]({{< ref "docs/dev/table/functions/ptfs" >}}#state-ttl). |
Review Comment:
```suggestion
| `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 retention cannot be introduced later without losing the state. See [State TTL]({{< ref "docs/dev/table/functions/ptfs" >}}#state-ttl). |
```
just to avoid confusion with the previous sentence
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * 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.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+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;
+ private static final String SEEN_STATE_NAME = "seen";
+ private static final String CANDIDATE_STATE_NAME = "candidate";
+
+ 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());
+ };
+
+ public static final StateTypeStrategy SEEN_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ return Optional.of(
+ ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ return callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ }
+ };
+
+ public static final StateTypeStrategy CANDIDATE_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(ARG_INPUT)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "First argument must be a table for DEDUPLICATE_KEEP_FIRST."));
+ // 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()));
Review Comment:
let's not use boolean but a correct type, allows for switching the mode in the future. it doesn't add any additional state.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * 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.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+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;
+ private static final String SEEN_STATE_NAME = "seen";
+ private static final String CANDIDATE_STATE_NAME = "candidate";
+
+ 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());
+ };
+
+ public static final StateTypeStrategy SEEN_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ return Optional.of(
+ ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ return callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ }
+ };
+
+ public static final StateTypeStrategy CANDIDATE_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(ARG_INPUT)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "First argument must be a table for DEDUPLICATE_KEEP_FIRST."));
+ // 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()));
+ }
+ final List<Field> inputFields = DataType.getFields(tableSemantics.dataType());
+ final List<Field> payloadFields =
+ Arrays.stream(
+ ChangelogTypeStrategyUtils.computeOutputIndices(
+ tableSemantics))
+ .mapToObj(inputFields::get)
+ .collect(Collectors.toList());
+ return Optional.of(
+ ValueView.newValueViewDataType(
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "payload", DataTypes.ROW(payloadFields)),
+ DataTypes.FIELD("event_time", DataTypes.BIGINT()))));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ // 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 Optional.of(Duration.ZERO);
+ }
+ };
+
+ public static LinkedHashMap<String, StateTypeStrategy> stateTypeStrategies() {
+ final LinkedHashMap<String, StateTypeStrategy> strategies = new LinkedHashMap<>();
+ strategies.put(SEEN_STATE_NAME, SEEN_STATE_TYPE_STRATEGY);
+ strategies.put(CANDIDATE_STATE_NAME, CANDIDATE_STATE_TYPE_STRATEGY);
+ return strategies;
+ }
+
+ 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 'table.exec.state.ttl'.",
+ stateTtl.get());
Review Comment:
```suggestion
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());
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.config.ExecutionConfigOptions;
+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.Duration;
+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)))
+ .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"))
+ .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 TTL_EXPIRED_RESTORE =
+ TableTestProgram.of(
+ "deduplicate-keep-first-ttl-expired-restore",
+ "the 'seen' state in the savepoint has outlived state_ttl by the time "
+ + "the test restores it, so an already-seen key is emitted "
Review Comment:
this sounds like the test is not deterministic? drop?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.config.ExecutionConfigOptions;
+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.Duration;
+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)))
+ .producedAfterRestore(
+ Row.of("Bob", "later", Instant.ofEpochMilli(100)))
Review Comment:
add a couple of rows more, not just 2.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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;
+
+/** Implementation of the {@code DEDUPLICATE_KEEP_FIRST} process table function. */
+@Internal
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int CANDIDATE_PAYLOAD = 0;
+
+ private static final int CANDIDATE_EVENT_TIME = 1;
+
+ private final int[] outputIndices;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ final LogicalType[] payloadTypes = new LogicalType[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ payloadGetters[i] = RowData.createFieldGetter(payloadTypes[i], outputIndices[i]);
+ }
+ candidateSerializer =
+ InternalSerializers.create(RowType.of(RowType.of(payloadTypes), new BigIntType()));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
+ // a parameter for every declared argument, so it is unused here.
+ final RowKind kind = input.getRowKind();
+ if (kind == RowKind.UPDATE_BEFORE || kind == RowKind.DELETE) {
+ return;
+ }
+
+ final TimeContext<Long> context = ctx.timeContext(Long.class);
+ final Long rowtime = context.time();
+
+ if (rowtime != null) { // watermark mode
+ final Long watermark = context.tableWatermark();
+ if (watermark != null && rowtime < watermark) {
+ return;
+ }
+
+ if (seen.getValue() == null) { // not yet emitted → buffer the earliest candidate
+ 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(
+ candidateSerializer.copy(materializeCandidate(input, rowtime)));
+ context.registerOnTime(rowtime);
+ }
+ } else { // already emitted → drop, optionally refresh the TTL
+ if (resetTtlOnDuplicate == null || resetTtlOnDuplicate) {
+ seen.setValue(true);
Review Comment:
did you verify that this indeed causes resetTtlOnDuplicate?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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;
+
+/** Implementation of the {@code DEDUPLICATE_KEEP_FIRST} process table function. */
+@Internal
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int CANDIDATE_PAYLOAD = 0;
+
+ private static final int CANDIDATE_EVENT_TIME = 1;
+
+ private final int[] outputIndices;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ final LogicalType[] payloadTypes = new LogicalType[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ payloadGetters[i] = RowData.createFieldGetter(payloadTypes[i], outputIndices[i]);
+ }
+ candidateSerializer =
+ InternalSerializers.create(RowType.of(RowType.of(payloadTypes), new BigIntType()));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
+ // a parameter for every declared argument, so it is unused here.
+ final RowKind kind = input.getRowKind();
+ if (kind == RowKind.UPDATE_BEFORE || kind == RowKind.DELETE) {
+ return;
+ }
+
+ final TimeContext<Long> context = ctx.timeContext(Long.class);
+ final Long rowtime = context.time();
+
+ if (rowtime != null) { // watermark mode
+ final Long watermark = context.tableWatermark();
+ if (watermark != null && rowtime < watermark) {
+ return;
+ }
+
+ if (seen.getValue() == null) { // not yet emitted → buffer the earliest candidate
+ 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(
+ candidateSerializer.copy(materializeCandidate(input, rowtime)));
Review Comment:
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.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java:
##########
@@ -0,0 +1,560 @@
+/*
+ * 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.Duration;
+import java.time.Instant;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+
+/** {@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 TABLE_API =
Review Comment:
we can drop all table API tests, we can assume the framework supports Table API well when SQL does.
##########
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. `INTERVAL '0'` also retains state indefinitely, but excludes the TTL from the state layout, so a retention cannot be introduced later without losing the state. See [State TTL]({{< ref "docs/dev/table/functions/ptfs" >}}#state-ttl). |
+| `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.
Review Comment:
Do we have a test for this as well?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java:
##########
@@ -0,0 +1,560 @@
+/*
+ * 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.Duration;
+import java.time.Instant;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+
+/** {@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 TABLE_API =
+ TableTestProgram.of(
+ "deduplicate-keep-first-table-api",
+ "the generic Table API process() call invokes the PTF, matching "
+ + "the SQL form")
+ .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())
+ .runTableApi(
+ env ->
+ env.from("user_events")
+ .partitionBy($("user_name"))
+ .process("DEDUPLICATE_KEEP_FIRST"),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_WITH_ARGS =
+ TableTestProgram.of(
+ "deduplicate-keep-first-table-api-with-args",
+ "the Table API process() call passes the optional state_ttl and "
+ + "reset_ttl_on_duplicate as named arguments")
+ .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())
+ .runTableApi(
+ env ->
+ env.from("user_events")
+ .partitionBy($("user_name"))
+ .process(
+ "DEDUPLICATE_KEEP_FIRST",
+ lit(Duration.ofSeconds(5))
+ .asArgument("state_ttl"),
+ lit(false)
+ .asArgument("reset_ttl_on_duplicate")),
+ "sink")
+ .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("region STRING", "user_name STRING", "action STRING")
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "EU", "Vas", "login"),
+ Row.ofKind(RowKind.INSERT, "US", "Vas", "signup"),
+ Row.ofKind(RowKind.INSERT, "EU", "Alice", "view"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY (region, user_name))")
Review Comment:
```suggestion
+ "input => TABLE user_events PARTITION BY (user_name, region))")
```
challenge the implementation
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.config.ExecutionConfigOptions;
+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.Duration;
+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)))
+ .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"))
+ .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 TTL_EXPIRED_RESTORE =
+ TableTestProgram.of(
+ "deduplicate-keep-first-ttl-expired-restore",
+ "the 'seen' state in the savepoint has outlived state_ttl by the time "
+ + "the test restores it, so an already-seen key is emitted "
+ + "again in a new deduplication period")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedBeforeRestore(
+ Row.of("Vas", "login"), Row.of("Vas", "click"))
+ .producedAfterRestore(Row.of("Vas", "logout"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedBeforeRestore(
+ Row.ofKind(RowKind.INSERT, "Vas", "login"))
+ .consumedAfterRestore(
+ Row.ofKind(RowKind.INSERT, "Vas", "logout"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY user_name, "
+ + "state_ttl => INTERVAL '10' SECOND)")
+ .build();
+
+ public static final TableTestProgram GLOBAL_TTL_EXPIRED_RESTORE =
Review Comment:
same here: most likely not deterministic
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java:
##########
@@ -0,0 +1,560 @@
+/*
+ * 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.Duration;
+import java.time.Instant;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+
+/** {@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 TABLE_API =
+ TableTestProgram.of(
+ "deduplicate-keep-first-table-api",
+ "the generic Table API process() call invokes the PTF, matching "
+ + "the SQL form")
+ .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())
+ .runTableApi(
+ env ->
+ env.from("user_events")
+ .partitionBy($("user_name"))
+ .process("DEDUPLICATE_KEEP_FIRST"),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_WITH_ARGS =
+ TableTestProgram.of(
+ "deduplicate-keep-first-table-api-with-args",
+ "the Table API process() call passes the optional state_ttl and "
+ + "reset_ttl_on_duplicate as named arguments")
+ .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())
+ .runTableApi(
+ env ->
+ env.from("user_events")
+ .partitionBy($("user_name"))
+ .process(
+ "DEDUPLICATE_KEEP_FIRST",
+ lit(Duration.ofSeconds(5))
+ .asArgument("state_ttl"),
+ lit(false)
+ .asArgument("reset_ttl_on_duplicate")),
+ "sink")
+ .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("region STRING", "user_name STRING", "action STRING")
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "EU", "Vas", "login"),
+ Row.ofKind(RowKind.INSERT, "US", "Vas", "signup"),
+ Row.ofKind(RowKind.INSERT, "EU", "Alice", "view"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY (region, user_name))")
+ .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_MULTI_KEY =
Review Comment:
semantic tests are expensive, let's also drop this. we should avoid to many tests
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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;
+
+/** Implementation of the {@code DEDUPLICATE_KEEP_FIRST} process table function. */
+@Internal
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int CANDIDATE_PAYLOAD = 0;
+
+ private static final int CANDIDATE_EVENT_TIME = 1;
+
+ private final int[] outputIndices;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ final LogicalType[] payloadTypes = new LogicalType[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ payloadGetters[i] = RowData.createFieldGetter(payloadTypes[i], outputIndices[i]);
+ }
+ candidateSerializer =
+ InternalSerializers.create(RowType.of(RowType.of(payloadTypes), new BigIntType()));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
+ // a parameter for every declared argument, so it is unused here.
+ final RowKind kind = input.getRowKind();
+ if (kind == RowKind.UPDATE_BEFORE || kind == RowKind.DELETE) {
+ return;
+ }
+
+ final TimeContext<Long> context = ctx.timeContext(Long.class);
Review Comment:
no need to get time context for watermarkless execution. we can derive a boolean flag in the constructor to know the mode early.
##########
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. `INTERVAL '0'` also retains state indefinitely, but excludes the TTL from the state layout, so a retention cannot be introduced later without losing the state. See [State TTL]({{< ref "docs/dev/table/functions/ptfs" >}}#state-ttl). |
+| `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.
Review Comment:
```suggestion
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.
```
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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;
+
+/** Implementation of the {@code DEDUPLICATE_KEEP_FIRST} process table function. */
+@Internal
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int CANDIDATE_PAYLOAD = 0;
+
+ private static final int CANDIDATE_EVENT_TIME = 1;
+
+ private final int[] outputIndices;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ final LogicalType[] payloadTypes = new LogicalType[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ payloadGetters[i] = RowData.createFieldGetter(payloadTypes[i], outputIndices[i]);
+ }
+ candidateSerializer =
+ InternalSerializers.create(RowType.of(RowType.of(payloadTypes), new BigIntType()));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
Review Comment:
```suggestion
// state_ttl and reset_ttl_on_duplicate is applied at plan time via getTimeToLive(); the PTF codegen still requires
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313401448
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java:
##########
@@ -0,0 +1,560 @@
+/*
+ * 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.Duration;
+import java.time.Instant;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+
+/** {@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 TABLE_API =
+ TableTestProgram.of(
+ "deduplicate-keep-first-table-api",
+ "the generic Table API process() call invokes the PTF, matching "
+ + "the SQL form")
+ .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())
+ .runTableApi(
+ env ->
+ env.from("user_events")
+ .partitionBy($("user_name"))
+ .process("DEDUPLICATE_KEEP_FIRST"),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_WITH_ARGS =
+ TableTestProgram.of(
+ "deduplicate-keep-first-table-api-with-args",
+ "the Table API process() call passes the optional state_ttl and "
+ + "reset_ttl_on_duplicate as named arguments")
+ .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())
+ .runTableApi(
+ env ->
+ env.from("user_events")
+ .partitionBy($("user_name"))
+ .process(
+ "DEDUPLICATE_KEEP_FIRST",
+ lit(Duration.ofSeconds(5))
+ .asArgument("state_ttl"),
+ lit(false)
+ .asArgument("reset_ttl_on_duplicate")),
+ "sink")
+ .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("region STRING", "user_name STRING", "action STRING")
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, "EU", "Vas", "login"),
+ Row.ofKind(RowKind.INSERT, "US", "Vas", "signup"),
+ Row.ofKind(RowKind.INSERT, "EU", "Alice", "view"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY (region, user_name))")
Review Comment:
yep makes sense 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313354769
##########
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. `INTERVAL '0'` also retains state indefinitely, but excludes the TTL from the state layout, so a retention cannot be introduced later without losing the state. See [State TTL]({{< ref "docs/dev/table/functions/ptfs" >}}#state-ttl). |
+| `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.
Review Comment:
There was no test for ts == watermark, and the code only dropped ts < watermark, so I changed it to ts <= watermark. EVENT_TIME_LATE_DROPPED now has a row at exactly the watermark, and it fails with the old <.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313365050
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.config.ExecutionConfigOptions;
+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.Duration;
+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)))
+ .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"))
+ .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 TTL_EXPIRED_RESTORE =
+ TableTestProgram.of(
+ "deduplicate-keep-first-ttl-expired-restore",
+ "the 'seen' state in the savepoint has outlived state_ttl by the time "
+ + "the test restores it, so an already-seen key is emitted "
+ + "again in a new deduplication period")
+ .setupTableSource(
+ SourceTestStep.newBuilder("user_events")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .producedBeforeRestore(
+ Row.of("Vas", "login"), Row.of("Vas", "click"))
+ .producedAfterRestore(Row.of("Vas", "logout"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(USER_EVENTS_SCHEMA)
+ .consumedBeforeRestore(
+ Row.ofKind(RowKind.INSERT, "Vas", "login"))
+ .consumedAfterRestore(
+ Row.ofKind(RowKind.INSERT, "Vas", "logout"))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST("
+ + "input => TABLE user_events PARTITION BY user_name, "
+ + "state_ttl => INTERVAL '10' SECOND)")
+ .build();
+
+ public static final TableTestProgram GLOBAL_TTL_EXPIRED_RESTORE =
Review Comment:
dropped
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4185122131
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.api.common.typeutils.TypeSerializer;
+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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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.util.stream.IntStream;
+
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int[] outputIndices;
+
+ private final int eventTimeIndex;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient ProjectedRowData emitProjection;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.eventTimeIndex = outputIndices.length;
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ emitProjection = ProjectedRowData.from(IntStream.range(0, eventTimeIndex).toArray());
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadGetters[i] =
+ RowData.createFieldGetter(
+ inputRowType.getTypeAt(outputIndices[i]), outputIndices[i]);
+ }
+
+ final LogicalType[] candidateTypes = new LogicalType[eventTimeIndex + 1];
+ for (int i = 0; i < eventTimeIndex; i++) {
+ candidateTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ }
+ candidateTypes[eventTimeIndex] = new BigIntType();
+ candidateSerializer = InternalSerializers.create(RowType.of(candidateTypes));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
+ // a parameter for every declared argument, so it is unused here.
+ final Long rowtime = ctx.timeContext(Long.class).time();
Review Comment:
Hoisted into a single context
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29365:
URL: https://github.com/apache/flink/pull/29365#issuecomment-5953784165
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8a0f24498aa1e30cbf4430d74ac199fb85afee77",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8a0f24498aa1e30cbf4430d74ac199fb85afee77",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8a0f24498aa1e30cbf4430d74ac199fb85afee77 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4167272882
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.table.api.DataTypes;
+import org.apache.flink.table.api.DataTypes.Field;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public 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 final String SEEN_STATE_NAME = "seen";
+
+ public static final StateTypeStrategy SEEN_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ return Optional.of(
+ ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ return callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ }
+ };
+
+ private static final String CANDIDATE_STATE_NAME = "candidate";
Review Comment:
can we have one line constants at the very top?
it will simplify reading
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313357142
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * 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.api.dataview.ValueView;
+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.StateTypeStrategy;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+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;
+ private static final String SEEN_STATE_NAME = "seen";
+ private static final String CANDIDATE_STATE_NAME = "candidate";
+
+ 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());
+ };
+
+ public static final StateTypeStrategy SEEN_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ return Optional.of(
+ ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull()));
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext callContext) {
+ return callContext.getArgumentValue(ARG_STATE_TTL, Duration.class);
+ }
+ };
+
+ public static final StateTypeStrategy CANDIDATE_STATE_TYPE_STRATEGY =
+ new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext) {
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(ARG_INPUT)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "First argument must be a table for DEDUPLICATE_KEEP_FIRST."));
+ // 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()));
Review Comment:
sounds good will change accordingly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40896][table]Deduplicate keep first implementation [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #29365:
URL: https://github.com/apache/flink/pull/29365#discussion_r4313386111
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.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.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.strategies.ChangelogTypeStrategyUtils;
+import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy;
+import org.apache.flink.table.types.logical.BigIntType;
+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;
+
+/** Implementation of the {@code DEDUPLICATE_KEEP_FIRST} process table function. */
+@Internal
+public class DeduplicateKeepFirst extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int CANDIDATE_PAYLOAD = 0;
+
+ private static final int CANDIDATE_EVENT_TIME = 1;
+
+ private final int[] outputIndices;
+
+ private final RowType inputRowType;
+
+ private transient ProjectedRowData projectedOutput;
+
+ private transient RowData.FieldGetter[] payloadGetters;
+
+ private transient TypeSerializer<RowData> candidateSerializer;
+
+ public DeduplicateKeepFirst(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.DEDUPLICATE_KEEP_FIRST, context);
+ final CallContext callContext = context.getCallContext();
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(DeduplicateKeepFirstTypeStrategy.ARG_INPUT)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ this.outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
+ this.inputRowType = (RowType) tableSemantics.dataType().getLogicalType();
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedOutput = ProjectedRowData.from(outputIndices);
+ payloadGetters = new RowData.FieldGetter[outputIndices.length];
+ final LogicalType[] payloadTypes = new LogicalType[outputIndices.length];
+ for (int i = 0; i < outputIndices.length; i++) {
+ payloadTypes[i] = inputRowType.getTypeAt(outputIndices[i]);
+ payloadGetters[i] = RowData.createFieldGetter(payloadTypes[i], outputIndices[i]);
+ }
+ candidateSerializer =
+ InternalSerializers.create(RowType.of(RowType.of(payloadTypes), new BigIntType()));
+ }
+
+ 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 {
+ // state_ttl is applied at plan time via getTimeToLive(); the PTF codegen still requires
Review Comment:
added to the sentence
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org