[
https://issues.apache.org/jira/browse/FLINK-41419?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Vas Shabu updated FLINK-41419:
------------------------------
Description:
Deduplication is a common Flink SQL need. Today users express it with a
ROW_NUMBER() over-window, which the planner recognizes as deduplication only
when it is ordered by a time attribute. Ordering by a version column instead
turns it into a general Top-N. The alternative, a LAST_VALUE aggregate per
column, can combine columns from different records.
This ticket adds two built-in PTFs, each with keep => 'first' | 'last':
* DEDUPLICATE: keeps one row per key, ordered by arrival or by sequence_by. It
accepts insert-only and updating input, and its output is updating.
- DEDUPLICATE_WINDOW: emits one final row per key and per-key event-time
window. The window closes after a fixed duration, after a period of inactivity,
or whichever comes first. It accepts insert-only input, and its output is
insert-only.
{code:java}
DEDUPLICATE(
input => TABLE(t) PARTITION BY <cols>,
keep => 'first' | 'last',
sequence_by => DESCRIPTOR(<cols>),
state_ttl => <interval>,
reset_ttl_on_duplicate => <boolean>,
uid => <string>
){code}
{code:java}
DEDUPLICATE_WINDOW(
input => TABLE(t) PARTITION BY <cols>,
keep => 'first' | 'last',
duration => <interval>,
inactivity => <interval>,
sequence_by => DESCRIPTOR(<cols>),
state_ttl => <interval>,
reset_ttl_on_duplicate => <boolean>,
on_time => DESCRIPTOR(<rowtime>),
uid => <string>
)
{code}
PARTITION BY is required for both functions. For updating input, DEDUPLICATE
requires the PARTITION BY columns to contain an upsert key. DEDUPLICATE_WINDOW
requires on_time and at least one of duration or inactivity; updating input is
rejected at planning time.
See
[FLIP-622|https://cwiki.apache.org/confluence/spaces/FLINK/pages/462816464/FLIP-622+Add+built-in+DEDUPLICATE+and+DEDUPLICATE_WINDOW+PTFs]
for full motivation, rejected alternatives, and compatibility plan. It
supersedes FLIP-614 / DEDUPLICATE_KEEP_FIRST, which is unreleased.
was:
Deduplication is a common Flink SQL need. Today users express it with a
ROW_NUMBER() over-window, which the planner recognizes as deduplication only
when it is ordered by a time attribute. Ordering by a version column instead
turns it into a general Top-N. The alternative, a LAST_VALUE aggregate per
column, can combine columns from different records.
This ticket adds two built-in PTFs, each with keep => 'first' | 'last':
* DEDUPLICATE: keeps one row per key, ordered by arrival or by sequence_by. It
accepts insert-only and updating input, and its output is updating.
- DEDUPLICATE_WINDOW: emits one final row per key and per-key event-time
window. The window closes after a fixed duration, after a period of inactivity,
or whichever comes first. It accepts insert-only input, and its output is
insert-only.
{code:java}
DEDUPLICATE(
input => TABLE(t) PARTITION BY <cols>,
keep => 'first' | 'last',
sequence_by => DESCRIPTOR(<cols>),
state_ttl => <interval>,
reset_ttl_on_duplicate => <boolean>,
uid => <string>
){code}
{code:java}
DEDUPLICATE_WINDOW(
input => TABLE(t) PARTITION BY <cols>,
keep => 'first' | 'last',
duration => <interval>,
inactivity => <interval>,
sequence_by => DESCRIPTOR(<cols>),
state_ttl => <interval>,
reset_ttl_on_duplicate => <boolean>,
on_time => DESCRIPTOR(<rowtime>),
uid => <string>
)
{code}
PARTITION BY is required for both functions. For updating input, DEDUPLICATE
requires the PARTITION BY columns to contain an upsert key. DEDUPLICATE_WINDOW
requires on_time and at least one of duration or inactivity; updating input is
rejected at planning time.
See
[FLIP-622|https://cwiki.apache.org/confluence/spaces/FLINK/pages/462816464/FLIP-622+Add+built-in+DEDUPLICATE+and+DEDUPLICATE_WINDOW+PTFs]
for full motivation, rejected alternatives, and compatibility plan. It
supersedes FLIP-614 / DEDUPLICATE_KEEP_FIRST, which is unreleased.
> FLIP-622: Add built-in DEDUPLICATE, DEDUPLICATE_WINDOW PTF
> ----------------------------------------------------------
>
> Key: FLINK-41419
> URL: https://issues.apache.org/jira/browse/FLINK-41419
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / API, Table SQL / Client
> Reporter: Vas Shabu
> Priority: Major
>
> Deduplication is a common Flink SQL need. Today users express it with a
> ROW_NUMBER() over-window, which the planner recognizes as deduplication only
> when it is ordered by a time attribute. Ordering by a version column instead
> turns it into a general Top-N. The alternative, a LAST_VALUE aggregate per
> column, can combine columns from different records.
> This ticket adds two built-in PTFs, each with keep => 'first' | 'last':
> * DEDUPLICATE: keeps one row per key, ordered by arrival or by sequence_by.
> It accepts insert-only and updating input, and its output is updating.
> - DEDUPLICATE_WINDOW: emits one final row per key and per-key event-time
> window. The window closes after a fixed duration, after a period of
> inactivity, or whichever comes first. It accepts insert-only input, and its
> output is insert-only.
> {code:java}
> DEDUPLICATE(
> input => TABLE(t) PARTITION BY <cols>,
> keep => 'first' | 'last',
> sequence_by => DESCRIPTOR(<cols>),
> state_ttl => <interval>,
> reset_ttl_on_duplicate => <boolean>,
> uid => <string>
> ){code}
> {code:java}
> DEDUPLICATE_WINDOW(
> input => TABLE(t) PARTITION BY <cols>,
> keep => 'first' | 'last',
> duration => <interval>,
> inactivity => <interval>,
> sequence_by => DESCRIPTOR(<cols>),
> state_ttl => <interval>,
> reset_ttl_on_duplicate => <boolean>,
> on_time => DESCRIPTOR(<rowtime>),
> uid => <string>
> )
> {code}
>
> PARTITION BY is required for both functions. For updating input, DEDUPLICATE
> requires the PARTITION BY columns to contain an upsert key.
> DEDUPLICATE_WINDOW requires on_time and at least one of duration or
> inactivity; updating input is rejected at planning time.
> See
> [FLIP-622|https://cwiki.apache.org/confluence/spaces/FLINK/pages/462816464/FLIP-622+Add+built-in+DEDUPLICATE+and+DEDUPLICATE_WINDOW+PTFs]
> for full motivation, rejected alternatives, and compatibility plan. It
> supersedes FLIP-614 / DEDUPLICATE_KEEP_FIRST, which is unreleased.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
