[
https://issues.apache.org/jira/browse/FLINK-40964?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18129688#comment-18129688
]
Mika Naylor commented on FLINK-40964:
-------------------------------------
Absolutely! Thank you for picking this up [~sepurisaikrishna]
> Streaming Top-N over an updating ROW_NUMBER input never retracts a replaced
> row (outer Rank uses AppendFastStrategy)
> --------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40964
> URL: https://issues.apache.org/jira/browse/FLINK-40964
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.2.1
> Environment: Flink 2.2.1 SQL client on a session cluster (Java 17);
> PyFlink 2.2.1 (Java 21)
> Reporter: Sumaiya Azad
> Priority: Major
> Attachments: flink_topn_stale_row.py, flink_topn_stale_row.sql
>
>
> h3. What happens
> We keep the latest value per key, then pick the top key by that value. When a
> key's latest value changes, streaming mode never retracts the key's old row,
> so the result keeps the old value. Batch mode gives the right answer.
> h3. How to reproduce
> Four rows. Key A first has a value of 100, then a later value of 1. Key B has
> 50. The last row is a redelivered copy.
> {code:sql}
> -- t(id STRING, seq INT, k STRING, ts TIMESTAMP(3), v DOUBLE)
> -- ('a1', 1, 'A', '2024-06-01 08:00:00', 100.0)
> -- ('b1', 2, 'B', '2024-06-01 08:05:00', 50.0)
> -- ('a2', 3, 'A', '2024-06-01 08:10:00', 1.0)
> -- ('a2', 4, 'A', '2024-06-01 08:10:00', 1.0)
> WITH latest AS ( -- latest value per key
> SELECT k, v FROM (
> SELECT *, ROW_NUMBER() OVER (PARTITION BY k ORDER BY ts DESC, id DESC) AS
> rk
> FROM t) WHERE rk = 1)
> SELECT k, v, r FROM ( -- top 1 key by that value
> SELECT k, v, ROW_NUMBER() OVER (ORDER BY v DESC, k ASC) AS r FROM latest)
> WHERE r <= 1
> {code}
> Two attached files run this query in streaming mode and in batch mode:
> * {{{}flink_topn_stale_row.sql{}}}, for the Flink SQL client:
> {{./bin/sql-client.sh -f flink_topn_stale_row.sql}}
> * {{{}flink_topn_stale_row.py{}}}, for PyFlink: {{{}python
> flink_topn_stale_row.py{}}}. It also prints the plan and exits 0 when the bug
> reproduces.
> Reproduced two ways on 2026-09-28: with the SQL client on a Flink 2.2.1
> session cluster (Java 17), and with PyFlink 2.2.1 (Java 21).
> h3. Expected
> {{{}('B', 50.0, 1){}}}. A's latest value is 1, so B is on top.
> h3. Actual
> * Streaming mode: the changelog is a single {{{}+I ('A', 100.0, 1){}}}. A's
> old value of 100 is never removed.
> * Batch mode: {{{}('B', 50.0, 1){}}}, which is correct.
> h3. Likely cause
> The streaming plan runs both ranks with {{{}AppendFastStrategy{}}}:
> {code:java}
> Rank(strategy=[AppendFastStrategy], ... orderBy=[v DESC, k ASC], select=[k,
> v])
> +- Rank(strategy=[AppendFastStrategy], ... partitionBy=[k], orderBy=[$4 DESC,
> id DESC])
> {code}
> {{AppendFastStrategy}} assumes its input only ever inserts rows. The inner
> rank does not: it replaces A's row when a later row for A arrives. The outer
> rank, therefore, never retracts A's old row.
> h3. Why it matters
> Any Top-N or filter built on a "latest row" query can publish a wrong row,
> and nothing reports an error. On real data, at a parallelism of 2, we saw
> this in 8 such queries, even with events delivered in time order.
> First reported on [email protected]:
> [https://lists.apache.org/thread/1zw7fd76ypks7qgryn681qxj66z075ls]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
