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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
