Gustavo de Morais created FLINK-41216:
-----------------------------------------
Summary: 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
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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
