[
https://issues.apache.org/jira/browse/FLINK-41216?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Gustavo de Morais updated FLINK-41216:
--------------------------------------
Description:
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.
was:
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,x,1] +U[2,x,1] -U[2,x,2] +U[3,x,2] -D[3,x,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. A regular join keyed by
{{id}} stores id 2 on {{+U[2,x,1]}} and removes it again on {{-U[2,x,2]}}, so
the row is gone from the join state. Upsert sinks keyed by {{id}} have the same
problem. FLINK-23142 fixed the same ordering issue in {{UpdatableTopNFunction}}.
Since FLINK-40997, the join always picks {{id}} for this query, so it is wrong
on every compilation instead of only some.
{code:sql}
CREATE TABLE items (id BIGINT, score INT, brand STRING, PRIMARY KEY (id) NOT
ENFORCED)
WITH ('connector' = 'values', 'changelog-mode' = 'I,UB,UA,D');
CREATE TABLE z (k STRING, z STRING) WITH ('connector' = 'values');
INSERT INTO snk SELECT t.id, t.rn, z.z
FROM (SELECT * FROM (SELECT *, ROW_NUMBER() OVER (ORDER BY score DESC) AS rn
FROM items)
WHERE rn <= 3) t
JOIN z ON t.brand = z.k;
{code}
items: +I[1, 100, x], +I[2, 90, x], +I[3, 80, x], -D[1, 100, x], then z: +I[x,
Z], +I[x, W].
Expected: [2, 1, Z], [3, 2, Z], [2, 1, W], [3, 2, W]. Actual: no rows.
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.
> 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)
