[
https://issues.apache.org/jira/browse/FLINK-40964?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18130813#comment-18130813
]
sepuri sai krishna commented on FLINK-40964:
--------------------------------------------
Does not reproduce on master. FLINK-40368 fixed it, Resolved for 2.2.2, 2.3.1
and 2.4.0 and
backported to release-2.2 and release-2.3. This report is against 2.2.1.
On master the outer Rank gets RetractStrategy and the inner Rank emits UB, so
the stale row is
retracted. The two Rank nodes from the physical plan, b the key and a the
ranked value:
{noformat}
Rank(strategy=[RetractStrategy], rankType=[ROW_NUMBER], rankRange=[rankStart=1,
rankEnd=1], partitionBy=[], orderBy=[a DESC, b ASC], select=[a, b, w0$o0],
changelogMode=[I,UA,D])
...
+- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
rankRange=[rankStart=1, rankEnd=1], partitionBy=[b], orderBy=[ROWTIME rowtime
DESC, a DESC], select=[a, b, rowtime], changelogMode=[I,UB,UA,D])
{noformat}
An ITCase with your four rows, including the redelivered copy, gives the
expected single row B in all
five parameter combinations. Reverting FLINK-40368's guard to
RankUtil.isDeduplication makes it keep
A's stale row, List(B,50,1, A,100,1), in the three mini batch disabled
combinations only.
The tests from FLINK-40368 all order ascending and feed an aggregate, so keep
last plus a downstream
Top-N is uncovered. I have a regression test ready if you want it as a follow
up, otherwise this can
be closed as fixed by FLINK-40368.
> 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
> Assignee: sepuri sai krishna
> 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
(v10.3.26#10030026)
