diff --git a/docs/content.zh/docs/sql/functions/built-in-functions.md b/docs/content.zh/docs/sql/functions/built-in-functions.md index 242cb4b03582e8..79115623e0aa74 100644 --- a/docs/content.zh/docs/sql/functions/built-in-functions.md +++ b/docs/content.zh/docs/sql/functions/built-in-functions.md @@ -289,6 +289,7 @@ Table functions can be used in two ways: as stand-alone inputs, where they are i | `FROM_CHANGELOG(input => TABLE t [, ...])` | Converts an append-only table with an explicit operation column into a dynamic table. See Changelog Conversion for the full list of arguments, semantics, and usage. | | `TO_CHANGELOG(input => TABLE t [, ...])` | Converts a dynamic table into an append-only table with an explicit operation column. See Changelog Conversion for the full list of arguments, semantics, and usage. | | `SNAPSHOT(input => TABLE t [, ...])` | Returns the current state of a dynamic table `t`. `SNAPSHOT` can only be used in a `LATERAL` context and not as a stand-alone table function. See [LATERAL SNAPSHOT join]({{< ref "docs/sql/reference/queries/joins" >}}#lateral-snapshot-join) for the full list of arguments, the join semantics, and usage. | +| `DEDUPLICATE_KEEP_FIRST(input => TABLE t [, ...])` | Keeps the first record per key as an append-only table (first arrival, or earliest event time with `on_time`). See [Deduplicate Keep First]({{< ref "docs/sql/reference/queries/deduplicate-keep-first" >}}#deduplicate_keep_first) for the full list of arguments, semantics, and usage. | To implement your own table functions, see [user-defined table functions]({{< ref "docs/dev/table/functions/udfs" >}}#table-functions). diff --git a/docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md b/docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md new file mode 100644 index 00000000000000..4e307552e4ff21 --- /dev/null +++ b/docs/content.zh/docs/sql/reference/queries/deduplicate-keep-first.md @@ -0,0 +1,192 @@ +--- +title: "Deduplicate Keep First" +weight: 16 +type: docs +--- + + +# 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 => ,] + [reset_ttl_on_duplicate => ,] + [on_time => DESCRIPTOR(rowtime_column),] + [uid => ] +) +``` + +`on_time` and `uid` are system arguments and always come last in the signature, which matters when arguments are passed by position. + +### Parameters + +| Parameter | Required | Description | +|:----------|:---------|:------------| +| `input` | Yes | The input table (append-only or updating). Use `PARTITION BY` to deduplicate per key; all rows for a key are routed to the same parallel instance. Without `PARTITION BY`, the whole input is a single group processed at parallelism 1, so only the first row of the entire stream is emitted and every later row is dropped regardless of its content. | +| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for the per-key deduplication state. After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. If omitted, falls back to `table.exec.state.ttl`; at its default of 0, state is retained indefinitely, but a retention can still be introduced later. `state_ttl => INTERVAL '0'` also retains state indefinitely, but excludes the TTL from the state layout, so a 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. Late events (i.e. events with timestamp smaller or equal to the received watermark) are not taken into consideration. + +#### Updating input + +```sql +-- Input (updating changelog): +-- -D[user_name:'Vas', action:'logout'] key not seen yet -> ignored +-- +I[user_name:'Vas', action:'login'] first record for the key -> emitted +-- +I[user_name:'Vas', action:'login'] duplicate -> swallowed +-- -U[user_name:'Vas', action:'login'] retraction -> swallowed +-- +U[user_name:'Vas', action:'click'] update -> swallowed +-- -D[user_name:'Vas', action:'click'] delete -> swallowed + +SELECT * FROM DEDUPLICATE_KEEP_FIRST( + input => TABLE user_events PARTITION BY user_name +) + +-- Output (insert-only): +-- +I[user_name:'Vas', action:'login'] +``` + +The input may be an [updating changelog]({{< ref "docs/dev/table/functions/ptfs" >}}#change-data-capture-basics) containing inserts (`+I`), update-befores (`-U`), update-afters (`+U`), and deletes (`-D`). `DEDUPLICATE_KEEP_FIRST` keeps the first `+I` or `+U` observed per key and swallows every later change to that key (duplicate, `-U`, `+U`, `-D`), so the output stays insert-only. A `-U` or `-D` for a key that has not been seen yet is ignored and does not mark the key as seen, so the next `+I` or `+U` for that key is still emitted. Event-time mode (`on_time`) is not supported with updating input. + +#### Table API + +```java +Table userEvents = ...; + +// Keep the first row per key (processing time). +Table result = userEvents.partitionBy($("user_name")).process("DEDUPLICATE_KEEP_FIRST"); + +// With optional arguments. +Table result = userEvents + .partitionBy($("user_name")) + .process( + "DEDUPLICATE_KEEP_FIRST", + lit(Duration.ofSeconds(5)).asArgument("state_ttl"), + lit(false).asArgument("reset_ttl_on_duplicate")); +``` + +{{< top >}} diff --git a/docs/content/docs/sql/functions/built-in-functions.md b/docs/content/docs/sql/functions/built-in-functions.md index 40b54b2869b29c..01bf207057959f 100644 --- a/docs/content/docs/sql/functions/built-in-functions.md +++ b/docs/content/docs/sql/functions/built-in-functions.md @@ -292,6 +292,7 @@ Table functions can be used in two ways: as stand-alone inputs, where they are i | `FROM_CHANGELOG(input => TABLE t [, ...])` | Converts an append-only table with an explicit operation column into a dynamic table. See [Changelog Conversion]({{< ref "docs/sql/reference/queries/changelog" >}}#from_changelog) for the full list of arguments, semantics, and usage. | | `TO_CHANGELOG(input => TABLE t [, ...])` | Converts a dynamic table into an append-only table with an explicit operation column. See [Changelog Conversion]({{< ref "docs/sql/reference/queries/changelog" >}}#to_changelog) for the full list of arguments, semantics, and usage. | | `SNAPSHOT(input => TABLE t [, ...])` | Returns the current state of a dynamic table `t`. `SNAPSHOT` can only be used in a `LATERAL` context and not as a stand-alone table function. See [LATERAL SNAPSHOT join]({{< ref "docs/sql/reference/queries/joins" >}}#lateral-snapshot-join) for the full list of arguments, the join semantics, and usage. | +| `DEDUPLICATE_KEEP_FIRST(input => TABLE t [, ...])` | Keeps the first record per key as an append-only table (first arrival, or earliest event time with `on_time`). See [Deduplicate Keep First]({{< ref "docs/sql/reference/queries/deduplicate-keep-first" >}}#deduplicate_keep_first) for the full list of arguments, semantics, and usage. | To implement your own table functions, see [user-defined table functions]({{< ref "docs/dev/table/functions/udfs" >}}#table-functions). diff --git a/docs/content/docs/sql/reference/queries/deduplicate-keep-first.md b/docs/content/docs/sql/reference/queries/deduplicate-keep-first.md new file mode 100644 index 00000000000000..4e307552e4ff21 --- /dev/null +++ b/docs/content/docs/sql/reference/queries/deduplicate-keep-first.md @@ -0,0 +1,192 @@ +--- +title: "Deduplicate Keep First" +weight: 16 +type: docs +--- + + +# 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 => ,] + [reset_ttl_on_duplicate => ,] + [on_time => DESCRIPTOR(rowtime_column),] + [uid => ] +) +``` + +`on_time` and `uid` are system arguments and always come last in the signature, which matters when arguments are passed by position. + +### Parameters + +| Parameter | Required | Description | +|:----------|:---------|:------------| +| `input` | Yes | The input table (append-only or updating). Use `PARTITION BY` to deduplicate per key; all rows for a key are routed to the same parallel instance. Without `PARTITION BY`, the whole input is a single group processed at parallelism 1, so only the first row of the entire stream is emitted and every later row is dropped regardless of its content. | +| `state_ttl` | No | An `INTERVAL` giving the processing-time retention for the per-key deduplication state. After a key's state expires, a later record for that key starts a new deduplication period and may be emitted again. If omitted, falls back to `table.exec.state.ttl`; at its default of 0, state is retained indefinitely, but a retention can still be introduced later. `state_ttl => INTERVAL '0'` also retains state indefinitely, but excludes the TTL from the state layout, so a 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. Late events (i.e. events with timestamp smaller or equal to the received watermark) are not taken into consideration. + +#### Updating input + +```sql +-- Input (updating changelog): +-- -D[user_name:'Vas', action:'logout'] key not seen yet -> ignored +-- +I[user_name:'Vas', action:'login'] first record for the key -> emitted +-- +I[user_name:'Vas', action:'login'] duplicate -> swallowed +-- -U[user_name:'Vas', action:'login'] retraction -> swallowed +-- +U[user_name:'Vas', action:'click'] update -> swallowed +-- -D[user_name:'Vas', action:'click'] delete -> swallowed + +SELECT * FROM DEDUPLICATE_KEEP_FIRST( + input => TABLE user_events PARTITION BY user_name +) + +-- Output (insert-only): +-- +I[user_name:'Vas', action:'login'] +``` + +The input may be an [updating changelog]({{< ref "docs/dev/table/functions/ptfs" >}}#change-data-capture-basics) containing inserts (`+I`), update-befores (`-U`), update-afters (`+U`), and deletes (`-D`). `DEDUPLICATE_KEEP_FIRST` keeps the first `+I` or `+U` observed per key and swallows every later change to that key (duplicate, `-U`, `+U`, `-D`), so the output stays insert-only. A `-U` or `-D` for a key that has not been seen yet is ignored and does not mark the key as seen, so the next `+I` or `+U` for that key is still emitted. Event-time mode (`on_time`) is not supported with updating input. + +#### Table API + +```java +Table userEvents = ...; + +// Keep the first row per key (processing time). +Table result = userEvents.partitionBy($("user_name")).process("DEDUPLICATE_KEEP_FIRST"); + +// With optional arguments. +Table result = userEvents + .partitionBy($("user_name")) + .process( + "DEDUPLICATE_KEEP_FIRST", + lit(Duration.ofSeconds(5)).asArgument("state_ttl"), + lit(false).asArgument("reset_ttl_on_duplicate")); +``` + +{{< top >}} diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java index 03e4780e0183d0..a81de868f34233 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java @@ -23,6 +23,7 @@ import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.inference.InputTypeStrategy; +import org.apache.flink.table.types.inference.StateTypeStrategy; import org.apache.flink.table.types.inference.StaticArgument; import org.apache.flink.table.types.inference.TypeInference; import org.apache.flink.table.types.inference.TypeStrategy; @@ -31,6 +32,7 @@ import java.lang.reflect.Constructor; import java.util.Arrays; +import java.util.LinkedHashMap; import java.util.Locale; import java.util.Optional; import java.util.regex.Pattern; @@ -345,6 +347,12 @@ public Builder outputTypeStrategy(TypeStrategy outputTypeStrategy) { return this; } + public Builder stateTypeStrategies( + LinkedHashMap stateTypeStrategies) { + this.typeInferenceBuilder.stateTypeStrategies(stateTypeStrategies); + return this; + } + public Builder notDeterministic() { this.isDeterministic = false; return this; diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java index 9e993e2642991d..00d247c6e54934 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java @@ -38,6 +38,7 @@ import org.apache.flink.table.types.inference.TraitCondition; import org.apache.flink.table.types.inference.TypeStrategies; import org.apache.flink.table.types.inference.strategies.ArrayOfStringArgumentTypeStrategy; +import org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy; import org.apache.flink.table.types.inference.strategies.FromChangelogTypeStrategy; import org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies; import org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies; @@ -939,6 +940,31 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL) "org.apache.flink.table.runtime.functions.ptf.FromChangelogFunction") .build(); + public static final BuiltInFunctionDefinition DEDUPLICATE_KEEP_FIRST = + BuiltInFunctionDefinition.newBuilder() + .name("DEDUPLICATE_KEEP_FIRST") + .kind(PROCESS_TABLE) + .staticArguments( + StaticArgument.table( + "input", + Row.class, + false, + EnumSet.of( + StaticArgumentTrait.TABLE, + StaticArgumentTrait.SET_SEMANTIC_TABLE, + StaticArgumentTrait.OPTIONAL_PARTITION_BY, + StaticArgumentTrait.SUPPORT_UPDATES)), + StaticArgument.scalar( + "state_ttl", DataTypes.INTERVAL(DataTypes.SECOND()), true), + StaticArgument.scalar( + "reset_ttl_on_duplicate", DataTypes.BOOLEAN(), true)) + .inputTypeStrategy(DeduplicateKeepFirstTypeStrategy.INPUT_TYPE_STRATEGY) + .outputTypeStrategy(DeduplicateKeepFirstTypeStrategy.OUTPUT_TYPE_STRATEGY) + .stateTypeStrategies(DeduplicateKeepFirstTypeStrategy.stateTypeStrategies()) + .runtimeClass( + "org.apache.flink.table.runtime.functions.ptf.DeduplicateKeepFirst") + .build(); + /** * Built-in proxy function for the LATERAL SNAPSHOT temporal join. * diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java new file mode 100644 index 00000000000000..28232e0846d4c2 --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategy.java @@ -0,0 +1,184 @@ +/* + * 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"; + public static final String CANDIDATE_STATE_NAME = "candidate"; + + public static final InputTypeStrategy INPUT_TYPE_STRATEGY = + new ValidationOnlyInputTypeStrategy() { + @Override + public Optional> inferInputTypes( + final CallContext callContext, final boolean throwOnFailure) { + final Optional> stateTtlError = + validateStateTtl(callContext, throwOnFailure); + if (stateTtlError.isPresent()) { + return stateTtlError; + } + + final Optional> 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 inputFields = DataType.getFields(tableSemantics.dataType()); + final List 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 inferType(final CallContext callContext) { + return Optional.of( + ValueView.newValueViewDataType(DataTypes.BOOLEAN().notNull())); + } + + @Override + public Optional 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 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 inputFields = DataType.getFields(tableSemantics.dataType()); + final List 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 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 stateTypeStrategies() { + final LinkedHashMap 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> 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 stateTtl = + callContext.getArgumentValue(ARG_STATE_TTL, Duration.class); + if (stateTtl.isPresent() && stateTtl.get().isNegative()) { + return callContext.fail( + throwOnFailure, + "The 'state_ttl' argument must not be negative, but was '%s'. " + + "Use a positive INTERVAL to bound state retention, INTERVAL '0' to " + + "retain state indefinitely without a TTL, or omit the argument to " + + "fall back to session defaults.", + stateTtl.get()); + } + return Optional.empty(); + } + + private static Optional> validateResetTtlOnDuplicate( + final CallContext callContext, final boolean throwOnFailure) { + if (callContext.isArgumentNull(ARG_RESET_TTL_ON_DUPLICATE)) { + return Optional.empty(); + } + if (!callContext.isArgumentLiteral(ARG_RESET_TTL_ON_DUPLICATE)) { + return callContext.fail( + throwOnFailure, + "The 'reset_ttl_on_duplicate' argument must be a constant BOOLEAN literal."); + } + return Optional.empty(); + } + + private DeduplicateKeepFirstTypeStrategy() {} +} diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategyTest.java new file mode 100644 index 00000000000000..59ec5195982793 --- /dev/null +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/DeduplicateKeepFirstTypeStrategyTest.java @@ -0,0 +1,103 @@ +/* + * 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.dataview.ValueView; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.inference.utils.CallContextMock; +import org.apache.flink.table.types.inference.utils.TableSemanticsMock; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy.ARG_INPUT; +import static org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy.ARG_RESET_TTL_ON_DUPLICATE; +import static org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy.ARG_STATE_TTL; +import static org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy.CANDIDATE_STATE_TYPE_STRATEGY; +import static org.apache.flink.table.types.inference.strategies.DeduplicateKeepFirstTypeStrategy.SEEN_STATE_TYPE_STRATEGY; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for the state contract of {@link DeduplicateKeepFirstTypeStrategy}. */ +class DeduplicateKeepFirstTypeStrategyTest { + + private static final DataType INPUT_TYPE = + DataTypes.ROW( + DataTypes.FIELD("user_name", DataTypes.STRING()), + DataTypes.FIELD("action", DataTypes.STRING()), + DataTypes.FIELD("ts", DataTypes.TIMESTAMP_LTZ(3))); + + @Test + void testCandidateStateDisablesTtl() { + // The event-time candidate buffer is cleared by its timer, so it must declare a disabled + // TTL rather than inheriting the global state TTL and expiring before the timer fires. + assertThat(CANDIDATE_STATE_TYPE_STRATEGY.getTimeToLive(callContext(Duration.ofSeconds(5)))) + .contains(Duration.ZERO); + assertThat(CANDIDATE_STATE_TYPE_STRATEGY.getTimeToLive(callContext(null))) + .contains(Duration.ZERO); + } + + @Test + void testCandidateStateNestsPayloadInBothModes() { + final DataType expected = + ValueView.newValueViewDataType( + DataTypes.ROW( + DataTypes.FIELD( + "payload", + DataTypes.ROW( + DataTypes.FIELD("action", DataTypes.STRING()), + DataTypes.FIELD("ts", DataTypes.TIMESTAMP_LTZ(3)))), + DataTypes.FIELD("event_time", DataTypes.BIGINT()))); + assertThat(CANDIDATE_STATE_TYPE_STRATEGY.inferType(callContext(-1))).contains(expected); + assertThat(CANDIDATE_STATE_TYPE_STRATEGY.inferType(callContext(2))).contains(expected); + } + + @Test + void testSeenStateTtlFollowsStateTtlArgument() { + assertThat(SEEN_STATE_TYPE_STRATEGY.getTimeToLive(callContext(Duration.ofSeconds(5)))) + .contains(Duration.ofSeconds(5)); + assertThat(SEEN_STATE_TYPE_STRATEGY.getTimeToLive(callContext(null))).isEmpty(); + } + + private static CallContextMock callContext(Duration stateTtl) { + final CallContextMock callContext = new CallContextMock(); + final List> values = new ArrayList<>(); + for (int i = 0; i <= ARG_RESET_TTL_ON_DUPLICATE; i++) { + values.add(Optional.empty()); + } + values.set(ARG_STATE_TTL, Optional.ofNullable(stateTtl)); + callContext.argumentValues = values; + return callContext; + } + + private static CallContextMock callContext(final int timeColumn) { + final CallContextMock callContext = new CallContextMock(); + callContext.tableSemantics = + Map.of( + ARG_INPUT, + new TableSemanticsMock( + INPUT_TYPE, new int[] {0}, new int[0], timeColumn, null)); + return callContext; + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTest.java new file mode 100644 index 00000000000000..fa9ad95f08559e --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTest.java @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.plan.nodes.exec.stream; + +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase; +import org.apache.flink.table.test.program.SourceTestStep; +import org.apache.flink.table.test.program.TableTestProgram; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +/** Restore tests for the built-in DEDUPLICATE_KEEP_FIRST PTF. */ +public class DeduplicateKeepFirstRestoreTest extends RestoreTestBase { + + // Bounds the wait for the pending-candidate savepoint trigger so a stuck source fails fast. + private static final long SAVEPOINT_READY_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(5); + + public DeduplicateKeepFirstRestoreTest() { + super(StreamExecProcessTableFunction.class); + } + + @Override + public List programs() { + return List.of( + DeduplicateKeepFirstRestoreTestPrograms.KEYED_RESTORE, + DeduplicateKeepFirstRestoreTestPrograms.EVENT_TIME_PENDING_RESTORE); + } + + @Override + protected void awaitSavepointReady(TableTestProgram program, List> futures) + throws Exception { + if (program != DeduplicateKeepFirstRestoreTestPrograms.EVENT_TIME_PENDING_RESTORE) { + super.awaitSavepointReady(program, futures); + return; + } + // The candidate is buffered but not emitted before restore, so there is no sink output to + // gate on. Instead wait until the source has emitted its before-restore rows; + // stop-with-savepoint then drains them into keyed state before snapshotting. + for (SourceTestStep source : program.getSetupSourceTestSteps()) { + final int count = source.dataBeforeRestore.size(); + if (count > 0) { + TestValuesTableFactory.awaitSourceEmitted(source.name, count) + .get(SAVEPOINT_READY_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + } + } + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java new file mode 100644 index 00000000000000..f33d91892d0926 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstRestoreTestPrograms.java @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.plan.nodes.exec.stream; + +import org.apache.flink.table.test.program.SinkTestStep; +import org.apache.flink.table.test.program.SourceTestStep; +import org.apache.flink.table.test.program.TableTestProgram; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; + +import java.time.Instant; + +/** Restore {@link TableTestProgram}s for the built-in DEDUPLICATE_KEEP_FIRST PTF. */ +public class DeduplicateKeepFirstRestoreTestPrograms { + + private static final String[] USER_EVENTS_SCHEMA = {"user_name STRING", "action STRING"}; + + private static final String[] TIMED_EVENTS_SCHEMA = { + "user_name STRING", + "action STRING", + "ts TIMESTAMP_LTZ(3)", + "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND" + }; + + public static final TableTestProgram KEYED_RESTORE = + TableTestProgram.of( + "deduplicate-keep-first-keyed-restore", + "the keep-first 'seen' state restores via compiled plan + savepoint: " + + "an already-seen key stays deduplicated after restore, and a " + + "key first observed after restore is emitted") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedBeforeRestore( + Row.of("Vas", "login"), Row.of("Vas", "click")) + .producedAfterRestore( + Row.of("Vas", "logout"), Row.of("Alice", "hello")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedBeforeRestore( + Row.ofKind(RowKind.INSERT, "Vas", "login")) + .consumedAfterRestore( + Row.ofKind(RowKind.INSERT, "Alice", "hello")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name)") + .build(); + + public static final TableTestProgram EVENT_TIME_PENDING_RESTORE = + TableTestProgram.of( + "deduplicate-keep-first-event-time-pending-restore", + "an event-time candidate buffered but not yet finalized (its timer has " + + "not fired) survives the savepoint: after restore, the " + + "watermark advances and the buffered candidate is emitted") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(TIMED_EVENTS_SCHEMA) + .producedBeforeRestore( + Row.of("Bob", "first", Instant.ofEpochMilli(5)), + Row.of("Alice", "first", Instant.ofEpochMilli(5)), + Row.of("Carol", "first", Instant.ofEpochMilli(5))) + .producedAfterRestore( + Row.of("Bob", "later", Instant.ofEpochMilli(100))) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedBeforeRestore(new Row[0]) + .consumedAfterRestore( + Row.ofKind(RowKind.INSERT, "Bob", "first"), + Row.ofKind(RowKind.INSERT, "Alice", "first"), + Row.ofKind(RowKind.INSERT, "Carol", "first")) + .build()) + .runSql( + "INSERT INTO sink SELECT user_name, action FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name, " + + "on_time => DESCRIPTOR(ts))") + .build(); +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstSemanticTests.java new file mode 100644 index 00000000000000..8d21c8355b1d66 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstSemanticTests.java @@ -0,0 +1,60 @@ +/* + * 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( + OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY, + OptimizerConfigOptions.NonDeterministicUpdateStrategy.IGNORE); + } + + @Override + public List programs() { + return List.of( + DeduplicateKeepFirstTestPrograms.KEYED, + DeduplicateKeepFirstTestPrograms.MULTI_KEY, + DeduplicateKeepFirstTestPrograms.MULTI_COLUMN_KEY, + DeduplicateKeepFirstTestPrograms.NULL_KEY, + DeduplicateKeepFirstTestPrograms.NO_PARTITION_BY, + DeduplicateKeepFirstTestPrograms.RESET_WITH_ZERO_TTL, + DeduplicateKeepFirstTestPrograms.NO_TTL_RESET, + DeduplicateKeepFirstTestPrograms.WHOLE_ROW_KEY, + DeduplicateKeepFirstTestPrograms.EVENT_TIME, + DeduplicateKeepFirstTestPrograms.EVENT_TIME_CANDIDATE_FIELD_NAMES, + DeduplicateKeepFirstTestPrograms.EVENT_TIME_LATE_DROPPED, + DeduplicateKeepFirstTestPrograms.UPDATING_INPUT, + DeduplicateKeepFirstTestPrograms.UNSEEN_RETRACTION_IGNORED, + DeduplicateKeepFirstTestPrograms.UPDATING_EVENT_TIME_FAILS, + DeduplicateKeepFirstTestPrograms.NEGATIVE_TTL_FAILS, + DeduplicateKeepFirstTestPrograms.NON_LITERAL_TTL_FAILS, + DeduplicateKeepFirstTestPrograms.NON_LITERAL_RESET_FAILS); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java new file mode 100644 index 00000000000000..2ecebb7e7984dc --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicateKeepFirstTestPrograms.java @@ -0,0 +1,456 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.plan.nodes.exec.stream; + +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.test.program.SinkTestStep; +import org.apache.flink.table.test.program.SourceTestStep; +import org.apache.flink.table.test.program.TableTestProgram; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; + +import java.time.Instant; + +/** {@link TableTestProgram} definitions for testing the built-in DEDUPLICATE_KEEP_FIRST PTF. */ +public class DeduplicateKeepFirstTestPrograms { + + private static final String[] USER_EVENTS_SCHEMA = {"user_name STRING", "action STRING"}; + + private static final String[] TIMED_EVENTS_SCHEMA = { + "user_name STRING", + "action STRING", + "ts TIMESTAMP_LTZ(3)", + "WATERMARK FOR ts AS ts - INTERVAL '10' SECOND" + }; + + public static final TableTestProgram KEYED = + TableTestProgram.of( + "deduplicate-keep-first-keyed", + "only the first row per PARTITION BY key is emitted") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedValues(Row.of("Vas", "login"), Row.of("Vas", "click")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues(Row.ofKind(RowKind.INSERT, "Vas", "login")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name)") + .build(); + + public static final TableTestProgram MULTI_KEY = + TableTestProgram.of( + "deduplicate-keep-first-multi-key", + "each PARTITION BY key keeps its own first row, independently of " + + "interleaving with other keys") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedValues( + Row.of("Vas", "login"), + Row.of("Alice", "click"), + Row.of("Vas", "click"), + Row.of("Alice", "view")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues( + Row.ofKind(RowKind.INSERT, "Vas", "login"), + Row.ofKind(RowKind.INSERT, "Alice", "click")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name)") + .build(); + + public static final TableTestProgram MULTI_COLUMN_KEY = + TableTestProgram.of( + "deduplicate-keep-first-multi-column-key", + "a multi-column PARTITION BY keeps the first row per composite key; " + + "rows differing in either key column form distinct groups") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema("region STRING", "user_name STRING", "action STRING") + .producedValues( + Row.of("EU", "Vas", "login"), + Row.of("EU", "Vas", "click"), + Row.of("US", "Vas", "signup"), + Row.of("EU", "Alice", "view")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("user_name STRING", "region STRING", "action STRING") + .consumedValues( + Row.ofKind(RowKind.INSERT, "Vas", "EU", "login"), + Row.ofKind(RowKind.INSERT, "Vas", "US", "signup"), + Row.ofKind(RowKind.INSERT, "Alice", "EU", "view")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY " + + "(user_name, region))") + .build(); + + public static final TableTestProgram NULL_KEY = + TableTestProgram.of( + "deduplicate-keep-first-null-key", + "a NULL PARTITION BY key forms its own dedup group, keeping only " + + "its first row") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedValues( + Row.of(null, "login"), + Row.of("Vas", "click"), + Row.of(null, "click")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues( + Row.ofKind(RowKind.INSERT, null, "login"), + Row.ofKind(RowKind.INSERT, "Vas", "click")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name)") + .build(); + + public static final TableTestProgram NO_PARTITION_BY = + TableTestProgram.of( + "deduplicate-keep-first-no-partition-by", + "without PARTITION BY the whole input is one group: only the " + + "very first row is emitted, regardless of its content") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedValues( + Row.of("Vas", "login"), + Row.of("Alice", "click"), + Row.of("Bob", "view")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues(Row.ofKind(RowKind.INSERT, "Vas", "login")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(input => TABLE user_events)") + .build(); + + public static final TableTestProgram RESET_WITH_ZERO_TTL = + TableTestProgram.of( + "deduplicate-keep-first-reset-with-zero-ttl", + "reset_ttl_on_duplicate=TRUE with state_ttl=INTERVAL '0' (TTL " + + "disabled): only the first row per key is emitted") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedValues( + Row.of("Vas", "login"), + Row.of("Vas", "click"), + Row.of("Vas", "view"), + Row.of("Vas", "logout"), + Row.of("Vas", "login")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues(Row.ofKind(RowKind.INSERT, "Vas", "login")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name, " + + "state_ttl => INTERVAL '0' SECOND, " + + "reset_ttl_on_duplicate => TRUE)") + .build(); + + public static final TableTestProgram NO_TTL_RESET = + TableTestProgram.of( + "deduplicate-keep-first-no-ttl-reset", + "reset_ttl_on_duplicate=FALSE still emits only the first row per key") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedValues( + Row.of("Vas", "login"), + Row.of("Vas", "click"), + Row.of("Vas", "view")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues(Row.ofKind(RowKind.INSERT, "Vas", "login")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name, " + + "state_ttl => INTERVAL '5' SECOND, " + + "reset_ttl_on_duplicate => FALSE)") + .build(); + + public static final TableTestProgram WHOLE_ROW_KEY = + TableTestProgram.of( + "deduplicate-keep-first-whole-row-key", + "PARTITION BY over every column deduplicates exact-duplicate rows, " + + "keeping one row per distinct payload") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedValues( + Row.of("Vas", "login"), + Row.of("Vas", "login"), + Row.of("Vas", "click")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues( + Row.ofKind(RowKind.INSERT, "Vas", "login"), + Row.ofKind(RowKind.INSERT, "Vas", "click")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY (user_name, action))") + .build(); + + public static final TableTestProgram EVENT_TIME = + TableTestProgram.of( + "deduplicate-keep-first-event-time", + "with on_time, the row with the smallest event time is kept per key, " + + "not the first to arrive") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(TIMED_EVENTS_SCHEMA) + .producedValues( + Row.of("Vas", "login", Instant.ofEpochMilli(3000)), + Row.of("Vas", "click", Instant.ofEpochMilli(1000)), + Row.of("Vas", "view", Instant.ofEpochMilli(5000))) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("user_name STRING", "action STRING") + .consumedValues(Row.ofKind(RowKind.INSERT, "Vas", "click")) + .build()) + .runSql( + "INSERT INTO sink SELECT user_name, action FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name, " + + "on_time => DESCRIPTOR(ts))") + .build(); + + public static final TableTestProgram EVENT_TIME_CANDIDATE_FIELD_NAMES = + TableTestProgram.of( + "deduplicate-keep-first-event-time-candidate-field-names", + "with on_time, input columns named like the buffered candidate's " + + "fields are emitted unchanged for the earliest row") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema( + "user_name STRING", + "payload STRING", + "event_time BIGINT", + "ts TIMESTAMP_LTZ(3)", + "WATERMARK FOR ts AS ts - INTERVAL '10' SECOND") + .producedValues( + Row.of("Vas", "login", 3L, Instant.ofEpochMilli(3000)), + Row.of("Vas", "click", 1L, Instant.ofEpochMilli(1000))) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema( + "user_name STRING", + "payload STRING", + "event_time BIGINT") + .consumedValues(Row.ofKind(RowKind.INSERT, "Vas", "click", 1L)) + .build()) + .runSql( + "INSERT INTO sink SELECT user_name, payload, event_time " + + "FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name, " + + "on_time => DESCRIPTOR(ts))") + .build(); + + public static final TableTestProgram EVENT_TIME_LATE_DROPPED = + TableTestProgram.of( + "deduplicate-keep-first-event-time-late-dropped", + "a row at or below the watermark is dropped, even though its event " + + "time is earlier than the already-finalized candidate; a late " + + "first row for an unseen key is dropped without marking the " + + "key as seen") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(TIMED_EVENTS_SCHEMA) + .producedValues( + Row.of("Bob", "early", Instant.ofEpochMilli(1000)), + Row.of("Bob", "advance", Instant.ofEpochMilli(20000)), + Row.of("Bob", "late", Instant.ofEpochMilli(500)), + Row.of("Carol", "late", Instant.ofEpochMilli(500)), + Row.of( + "Dave", + "at-watermark", + Instant.ofEpochMilli(10000)), + Row.of("Dave", "on-time", Instant.ofEpochMilli(15000)), + Row.of("Carol", "on-time", Instant.ofEpochMilli(30000))) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("user_name STRING", "action STRING") + .consumedValues( + Row.ofKind(RowKind.INSERT, "Bob", "early"), + Row.ofKind(RowKind.INSERT, "Dave", "on-time"), + Row.ofKind(RowKind.INSERT, "Carol", "on-time")) + .build()) + .runSql( + "INSERT INTO sink SELECT user_name, action FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name, " + + "on_time => DESCRIPTOR(ts))") + .build(); + + public static final TableTestProgram UPDATING_INPUT = + TableTestProgram.of( + "deduplicate-keep-first-updating-input", + "an updating input is deduplicated into an insert-only result: a -D " + + "before the key is seen is ignored, the first +I is emitted " + + "once and every later change (duplicate, -U, +U, -D) is " + + "swallowed") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .addOption("changelog-mode", "I,UB,UA,D") + .producedValues( + Row.ofKind(RowKind.DELETE, "Vas", "logout"), + Row.ofKind(RowKind.INSERT, "Vas", "login"), + Row.ofKind(RowKind.INSERT, "Vas", "login"), + Row.ofKind(RowKind.UPDATE_BEFORE, "Vas", "login"), + Row.ofKind(RowKind.UPDATE_AFTER, "Vas", "click"), + Row.ofKind(RowKind.DELETE, "Vas", "click")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues(Row.ofKind(RowKind.INSERT, "Vas", "login")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name)") + .build(); + + public static final TableTestProgram UNSEEN_RETRACTION_IGNORED = + TableTestProgram.of( + "deduplicate-keep-first-unseen-retraction-ignored", + "a -U or -D for a key that has not been seen is ignored and does not " + + "mark the key as seen, so the next +I or +U for that key is " + + "emitted as an insert") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .addOption("changelog-mode", "I,UB,UA,D") + .producedValues( + Row.ofKind(RowKind.DELETE, "Vas", "logout"), + Row.ofKind(RowKind.UPDATE_BEFORE, "Bob", "login"), + Row.ofKind(RowKind.UPDATE_AFTER, "Bob", "click"), + Row.ofKind(RowKind.INSERT, "Vas", "login")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues( + Row.ofKind(RowKind.INSERT, "Bob", "click"), + Row.ofKind(RowKind.INSERT, "Vas", "login")) + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name)") + .build(); + + public static final TableTestProgram UPDATING_EVENT_TIME_FAILS = + TableTestProgram.of( + "deduplicate-keep-first-updating-event-time-fails", + "event-time mode (on_time) is rejected for updating input, because a " + + "watermark asserts no earlier record will arrive while an " + + "update can invalidate that") + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(TIMED_EVENTS_SCHEMA) + .addOption("changelog-mode", "I,UB,UA,D") + .producedValues( + Row.ofKind(RowKind.INSERT, "Vas", "login", null)) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("user_name STRING", "action STRING") + .consumedValues(new Row[0]) + .build()) + .runFailingSql( + "INSERT INTO sink SELECT user_name, action FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name, " + + "on_time => DESCRIPTOR(ts))", + ValidationException.class, + "not supported for PTFs that consume or produce updates") + .build(); + + public static final TableTestProgram NEGATIVE_TTL_FAILS = + invalidArgumentProgram( + "deduplicate-keep-first-negative-ttl-fails", + "a negative state_ttl is rejected at planning time", + "state_ttl => INTERVAL '-1' SECOND", + "The 'state_ttl' argument must not be negative"); + + public static final TableTestProgram NON_LITERAL_TTL_FAILS = + invalidArgumentProgram( + "deduplicate-keep-first-non-literal-ttl-fails", + "a state_ttl that is not a literal is rejected at planning time", + "state_ttl => INTERVAL '1' SECOND * 2", + "The 'state_ttl' argument must be a constant INTERVAL literal."); + + public static final TableTestProgram NON_LITERAL_RESET_FAILS = + invalidArgumentProgram( + "deduplicate-keep-first-non-literal-reset-fails", + "a reset_ttl_on_duplicate that is not a literal is rejected at planning time", + "reset_ttl_on_duplicate => 1 = 1", + "The 'reset_ttl_on_duplicate' argument must be a constant BOOLEAN literal."); + + private static TableTestProgram invalidArgumentProgram( + final String id, final String description, final String argument, final String error) { + return TableTestProgram.of(id, description) + .setupTableSource( + SourceTestStep.newBuilder("user_events") + .addSchema(USER_EVENTS_SCHEMA) + .producedValues(Row.of("Vas", "login")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(USER_EVENTS_SCHEMA) + .consumedValues(new Row[0]) + .build()) + .runFailingSql( + "INSERT INTO sink SELECT * FROM DEDUPLICATE_KEEP_FIRST(" + + "input => TABLE user_events PARTITION BY user_name, " + + argument + + ")", + ValidationException.class, + error) + .build(); + } +} diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/plan/deduplicate-keep-first-event-time-pending-restore.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/plan/deduplicate-keep-first-event-time-pending-restore.json new file mode 100644 index 00000000000000..0f0841b716420f --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/plan/deduplicate-keep-first-event-time-pending-restore.json @@ -0,0 +1,352 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 5, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`user_events`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "user_name", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "ts", + "dataType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "ts", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE" + }, { + "kind" : "LITERAL", + "value" : "1", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE" + }, + "serializableString" : "`ts` - INTERVAL '0.001' SECOND" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$-$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE" + }, { + "kind" : "LITERAL", + "value" : "1", + "type" : "INTERVAL SECOND(6) NOT NULL" + } ], + "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE" + }, + "rowtimeExpr" : { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE" + }, + "idleTimeoutMillis" : -1, + "producedType" : { + "type" : "ROW", + "nullable" : false, + "fields" : [ { + "name" : "user_name", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "ts", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "user_name", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "ts", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "TableSourceScan(table=[[default_catalog, default_database, user_events, watermark=[-(ts, 1:INTERVAL SECOND)], watermarkEmitStrategy=[on-event]]], fields=[user_name, action, ts])" + }, { + "id" : 6, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "user_name", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "ts", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[user_name]])" + }, { + "id" : 7, + "type" : "stream-exec-process-table-function_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "user_name", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "ts", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "nullable" : false, + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "ProcessTableFunction(invocation=[DEDUPLICATE_KEEP_FIRST(TABLE(#0) PARTITION BY($0), DEFAULT(), DEFAULT(), DESCRIPTOR(_UTF-16LE'ts'), DEFAULT())], uid=[DEDUPLICATE_KEEP_FIRST], select=[user_name,action,ts,rowtime], rowType=[RecordType(VARCHAR(2147483647) user_name, VARCHAR(2147483647) action, TIMESTAMP_LTZ(3) *ROWTIME* ts, TIMESTAMP_LTZ(3) *ROWTIME* rowtime)])", + "uid" : "DEDUPLICATE_KEEP_FIRST", + "functionCall" : { + "kind" : "CALL", + "internalName" : "$DEDUPLICATE_KEEP_FIRST$1", + "operands" : [ { + "kind" : "TABLE_ARG_CALL", + "inputIndex" : 0, + "partitionKeys" : [ 0 ], + "type" : { + "type" : "ROW", + "nullable" : false, + "fields" : [ { + "name" : "user_name", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "ts", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + } + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$DEFAULT$1", + "type" : "INTERVAL SECOND(6)" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$DEFAULT$1", + "type" : "BOOLEAN" + }, { + "kind" : "CALL", + "internalName" : "$DESCRIPTOR$1", + "operands" : [ { + "kind" : "LITERAL", + "value" : "ts", + "type" : "CHAR(2) NOT NULL" + } ], + "type" : "DESCRIPTOR NOT NULL" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$DEFAULT$1", + "type" : "VARCHAR(2147483647)" + } ], + "type" : { + "type" : "ROW", + "nullable" : false, + "fields" : [ { + "name" : "user_name", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "ts", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "nullable" : false, + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + } + }, + "inputChangelogModes" : [ [ "INSERT" ] ], + "outputChangelogMode" : [ "INSERT" ], + "inputUpsertKeys" : [ [ ] ] + }, { + "id" : 8, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action` VARCHAR(2147483647)>", + "description" : "Calc(select=[user_name, action])" + }, { + "id" : 9, + "type" : "stream-exec-sink_2", + "configuration" : { + "table.exec.sink.keyed-shuffle" : "AUTO", + "table.exec.sink.not-null-enforcer" : "ERROR", + "table.exec.sink.rowtime-inserter" : "ENABLED", + "table.exec.sink.type-length-enforcer" : "IGNORE", + "table.exec.sink.upsert-materialize" : "AUTO" + }, + "dynamicTableSink" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`sink`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "user_name", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "dataType" : "VARCHAR(2147483647)" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "upsertMaterializeStrategy" : "ADAPTIVE", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action` VARCHAR(2147483647)>", + "description" : "Sink(table=[default_catalog.default_database.sink], fields=[user_name, action])" + } ], + "edges" : [ { + "source" : 5, + "target" : 6, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 6, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 7, + "target" : 8, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 8, + "target" : 9, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + } ] +} \ No newline at end of file diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/savepoint/_metadata new file mode 100644 index 00000000000000..37b4d90141272f Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-event-time-pending-restore/savepoint/_metadata differ diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/plan/deduplicate-keep-first-keyed-restore.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/plan/deduplicate-keep-first-keyed-restore.json new file mode 100644 index 00000000000000..f5add98b8ea558 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/plan/deduplicate-keep-first-keyed-restore.json @@ -0,0 +1,144 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 1, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`user_events`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "user_name", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "dataType" : "VARCHAR(2147483647)" + } ] + } + } + } + }, + "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, user_events]], fields=[user_name, action])" + }, { + "id" : 2, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action` VARCHAR(2147483647)>", + "description" : "Exchange(distribution=[hash[user_name]])" + }, { + "id" : 3, + "type" : "stream-exec-process-table-function_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action` VARCHAR(2147483647)>", + "description" : "ProcessTableFunction(invocation=[DEDUPLICATE_KEEP_FIRST(TABLE(#0) PARTITION BY($0), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], uid=[DEDUPLICATE_KEEP_FIRST], select=[user_name,action], rowType=[RecordType(VARCHAR(2147483647) user_name, VARCHAR(2147483647) action)])", + "uid" : "DEDUPLICATE_KEEP_FIRST", + "functionCall" : { + "kind" : "CALL", + "internalName" : "$DEDUPLICATE_KEEP_FIRST$1", + "operands" : [ { + "kind" : "TABLE_ARG_CALL", + "inputIndex" : 0, + "partitionKeys" : [ 0 ], + "type" : "ROW<`user_name` VARCHAR(2147483647), `action` VARCHAR(2147483647)> NOT NULL" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$DEFAULT$1", + "type" : "INTERVAL SECOND(6)" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$DEFAULT$1", + "type" : "BOOLEAN" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$DEFAULT$1", + "type" : "DESCRIPTOR" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$DEFAULT$1", + "type" : "VARCHAR(2147483647)" + } ], + "type" : "ROW<`user_name` VARCHAR(2147483647), `action` VARCHAR(2147483647)> NOT NULL" + }, + "inputChangelogModes" : [ [ "INSERT" ] ], + "outputChangelogMode" : [ "INSERT" ], + "inputUpsertKeys" : [ [ ] ] + }, { + "id" : 4, + "type" : "stream-exec-sink_2", + "configuration" : { + "table.exec.sink.keyed-shuffle" : "AUTO", + "table.exec.sink.not-null-enforcer" : "ERROR", + "table.exec.sink.rowtime-inserter" : "ENABLED", + "table.exec.sink.type-length-enforcer" : "IGNORE", + "table.exec.sink.upsert-materialize" : "AUTO" + }, + "dynamicTableSink" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`sink`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "user_name", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "action", + "dataType" : "VARCHAR(2147483647)" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "upsertMaterializeStrategy" : "ADAPTIVE", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`user_name` VARCHAR(2147483647), `action` VARCHAR(2147483647)>", + "description" : "Sink(table=[default_catalog.default_database.sink], fields=[user_name, action])" + } ], + "edges" : [ { + "source" : 1, + "target" : 2, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 2, + "target" : 3, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 3, + "target" : 4, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + } ] +} \ No newline at end of file diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/savepoint/_metadata new file mode 100644 index 00000000000000..76e13739bc4aa1 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/deduplicate-keep-first-keyed-restore/savepoint/_metadata differ diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/BuiltInProcessTableFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/BuiltInProcessTableFunction.java index 1ac2e2e03e71d4..2fb174cc37a8f9 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/BuiltInProcessTableFunction.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/BuiltInProcessTableFunction.java @@ -19,16 +19,21 @@ package org.apache.flink.table.runtime.functions.ptf; import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.dataview.ListView; +import org.apache.flink.table.api.dataview.MapView; +import org.apache.flink.table.api.dataview.ValueView; import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.functions.BuiltInFunctionDefinition; import org.apache.flink.table.functions.FunctionRequirement; import org.apache.flink.table.functions.ProcessTableFunction; import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext; +import org.apache.flink.table.runtime.dataview.DataViewUtils; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.inference.CallContext; import org.apache.flink.table.types.inference.StateTypeStrategy; import org.apache.flink.table.types.inference.TypeInference; import org.apache.flink.table.types.inference.TypeStrategies; +import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.utils.DataTypeUtils; import java.util.LinkedHashMap; @@ -109,12 +114,38 @@ private static LinkedHashMap toInternalStateStrategie ctx -> strategy.inferType(ctx) .map( - DataTypeUtils - ::toInternalDataType), + BuiltInProcessTableFunction + ::toInternalStateType), strategy.getTimeToLive(callContext).orElse(null)))); return result; } + /** + * Converts a state type to its internal representation. For {@link ValueView}/{@link + * MapView}/{@link ListView} only the value/key/element types are internalized; the view wrapper + * is preserved, unlike {@link DataTypeUtils#toInternalDataType} which would rewrite the whole + * view to {@code RowData}. + */ + private static DataType toInternalStateType(final DataType stateType) { + final LogicalType logicalType = stateType.getLogicalType(); + if (DataViewUtils.isDataView(logicalType, ValueView.class)) { + final DataType valueType = stateType.getChildren().get(0); + return ValueView.newValueViewDataType(DataTypeUtils.toInternalDataType(valueType)); + } + if (DataViewUtils.isDataView(logicalType, MapView.class)) { + final DataType mapType = stateType.getChildren().get(0); + return MapView.newMapViewDataType( + DataTypeUtils.toInternalDataType(mapType.getChildren().get(0)), + DataTypeUtils.toInternalDataType(mapType.getChildren().get(1))); + } + if (DataViewUtils.isDataView(logicalType, ListView.class)) { + final DataType arrayType = stateType.getChildren().get(0); + return ListView.newListViewDataType( + DataTypeUtils.toInternalDataType(arrayType.getChildren().get(0))); + } + return DataTypeUtils.toInternalDataType(stateType); + } + @Override public Set getRequirements() { if (definition != null) { diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java new file mode 100644 index 00000000000000..dd2ee20104fc89 --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/DeduplicateKeepFirst.java @@ -0,0 +1,198 @@ +/* + * 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.catalog.DataTypeFactory; +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.StateTypeStrategy; +import org.apache.flink.table.types.inference.TypeInference; +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.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; + +import javax.annotation.Nullable; + +import java.util.LinkedHashMap; + +/** Implementation of the {@code DEDUPLICATE_KEEP_FIRST} process table function. */ +@Internal +public class DeduplicateKeepFirst extends BuiltInProcessTableFunction { + + 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 final boolean isEventTime; + + private transient ProjectedRowData projectedOutput; + + private transient RowData.FieldGetter[] payloadGetters; + + private transient TypeSerializer[] payloadSerializers; + + 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(); + this.isEventTime = tableSemantics.timeColumn() != -1; + } + + @Override + public void open(final FunctionContext context) throws Exception { + super.open(context); + projectedOutput = ProjectedRowData.from(outputIndices); + payloadGetters = new RowData.FieldGetter[outputIndices.length]; + payloadSerializers = new TypeSerializer[outputIndices.length]; + for (int i = 0; i < outputIndices.length; i++) { + final LogicalType payloadType = inputRowType.getTypeAt(outputIndices[i]); + payloadGetters[i] = RowData.createFieldGetter(payloadType, outputIndices[i]); + payloadSerializers[i] = InternalSerializers.create(payloadType); + } + } + + @Override + public TypeInference getTypeInference(final DataTypeFactory typeFactory) { + final TypeInference typeInference = super.getTypeInference(typeFactory); + if (isEventTime) { + return typeInference; + } + final LinkedHashMap stateTypeStrategies = + new LinkedHashMap<>(typeInference.getStateTypeStrategies()); + stateTypeStrategies.remove(DeduplicateKeepFirstTypeStrategy.CANDIDATE_STATE_NAME); + return TypeInference.newBuilder() + .staticArguments( + typeInference.getStaticArguments().orElseThrow(IllegalStateException::new)) + .inputTypeStrategy(typeInference.getInputTypeStrategy()) + .stateTypeStrategies(stateTypeStrategies) + .outputTypeStrategy(typeInference.getOutputTypeStrategy()) + .disableSystemArguments(typeInference.disableSystemArguments()) + .build(); + } + + /** Watermarkless mode: emits the first arriving row per key. */ + public void eval( + final Context ctx, + final ValueView seen, + final RowData input, + @Nullable final Long stateTtl, + @Nullable final Boolean resetTtlOnDuplicate) + throws Exception { + // state_ttl and reset_ttl_on_duplicate are 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; + } + + if (seen.getValue() == null) { + projectedOutput.replaceRow(input); + projectedOutput.setRowKind(RowKind.INSERT); + collect(projectedOutput); + seen.setValue(true); + } else if (resetTtlOnDuplicate == null || resetTtlOnDuplicate) { + seen.setValue(true); + } + } + + /** Event-time mode: emits the row with the earliest event time per key. */ + public void eval( + final Context ctx, + final ValueView seen, + final ValueView candidate, + final RowData input, + @Nullable final Long stateTtl, + @Nullable final Boolean resetTtlOnDuplicate) + throws Exception { + final TimeContext context = ctx.timeContext(Long.class); + final Long rowtime = context.time(); + final Long watermark = context.tableWatermark(); + if (watermark != null && rowtime <= watermark) { + return; + } + + if (seen.getValue() == null) { // 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(materializeCandidate(input, rowtime)); + context.registerOnTime(rowtime); + } + } else if (resetTtlOnDuplicate == null || resetTtlOnDuplicate) { + seen.setValue(true); + } + } + + /** + * Never called: watermarkless mode registers no timers, but codegen requires an {@code onTimer} + * matching the state layout once any {@code onTimer} is declared. + */ + public void onTimer(final OnTimerContext ctx, final ValueView seen) {} + + public void onTimer( + final OnTimerContext ctx, + final ValueView seen, + final ValueView candidate) + throws Exception { + final RowData currLowest = candidate.getValue(); + if (currLowest == null) { + return; + } + collect(currLowest.getRow(CANDIDATE_PAYLOAD, outputIndices.length)); + candidate.clear(); + seen.setValue(true); + } + + private GenericRowData materializeCandidate(final RowData input, final long rowtime) { + final GenericRowData payload = new GenericRowData(outputIndices.length); + for (int i = 0; i < outputIndices.length; i++) { + final Object field = payloadGetters[i].getFieldOrNull(input); + payload.setField(i, field == null ? null : payloadSerializers[i].copy(field)); + } + return GenericRowData.of(payload, rowtime); + } +}