[
https://issues.apache.org/jira/browse/FLINK-41216?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18137630#comment-18137630
]
Rui Fan commented on FLINK-41216:
---------------------------------
[~gustavodemorais] Could you please confirm which versions this affects? We
plan to cut the 1.20.6/2.2.2/2.3.1 RCs around Nov 10. Do you plan to fix it for
them?
> Retracting ROW_NUMBER Top-N emits +U before -U for rows moving up, so
> operators keyed by the input upsert key lose rows
> -----------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41216
> URL: https://issues.apache.org/jira/browse/FLINK-41216
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner, Table SQL / Runtime
> Reporter: Gustavo de Morais
> Priority: Major
>
> When a row of a ROW_NUMBER Top-N with the row number in its output moves up
> in the ranking, {{RetractableTopNFunction}} emits it at its new rank before
> retracting it at its old rank. Deleting the top row of [1, 2, 3] emits:
> {code}
> -U[1,1] +U[2,1] -U[2,2] +U[3,2] -D[3,3]
> {code}
> The order is correct per {{rn}}, but not per the input key {{id}}, which the
> planner also exposes as an upsert key of the Top-N. Operators and sinks keyed
> by {{id}} end up with wrong results. FLINK-23142 fixed the same ordering
> issue in {{UpdatableTopNFunction}}.
> {code:sql}
> CREATE TABLE items (id BIGINT, score INT, PRIMARY KEY (id) NOT ENFORCED)
> WITH ('connector' = 'values', 'changelog-mode' = 'I,UB,UA,D');
> CREATE TABLE snk (id BIGINT, rn BIGINT, PRIMARY KEY (id) NOT ENFORCED)
> WITH ('connector' = 'values', 'sink-insert-only' = 'false');
> INSERT INTO snk SELECT id, rn FROM
> (SELECT *, ROW_NUMBER() OVER (ORDER BY score DESC) AS rn FROM items) WHERE
> rn <= 3;
> {code}
> ||Input||Sink PK||Result||Correct||
> |+I[1, 100], +I[2, 90], +I[3, 80], -D[1, 100]|(id)|[1, 1], [2, 1]|no, id 1
> stays after its delete and id 3 is missing|
> |same|(rn)|[2, 1], [3, 2]|yes|
> |+I[1, 100], +I[2, 90], +I[3, 80], -U[1, 100], +U[1, 50]|(id)|[1, 3], [2,
> 1]|no, id 3 is missing|
> |same|(rn)|[1, 3], [2, 1], [3, 2]|yes|
> Because {{id}} is an upsert key, the planner adds no upsert materializer in
> front of the sink. The upsert sink drops the -U messages and receives
> {{+U[2,1] +U[3,2] -D[3,3]}}, so it deletes id 3 and never deletes id 1. A
> regular join keyed by {{id}} has the same problem: it stores id 2 on
> {{+U[2,1]}} and removes it again on {{-U[2,2]}}. Since FLINK-40997, such a
> join always picks {{id}} when the row number is the last column.
> Possible solutions:
> * Emit the retraction of each moved row before its insert at the new rank,
> walking the ranks top-down. This keeps both {{id}} and {{rn}} valid upsert
> keys and changes no plan or state.
> * Stop exposing the input key as an upsert key of a Top-N that outputs the
> row number. This is a smaller change, but it changes plans, so jobs that
> re-submit their SQL switch state keys.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
