Hello,
I found what looks like a bug in Flink 2.2.1's streaming SQL planner, and I
could not file it myself because my ASF JIRA account request is still pending.
Could a committer open a JIRA issue for it, or tell me if this is already known
or expected behavior?
In short: a Top-N over a ROW_NUMBER() "latest row per key" query keeps a stale
row in streaming mode. The plan shows the outer Rank using AppendFastStrategy
even though its input is updating. Batch mode returns the correct answer.
A four-row reproduction (SQL client and PyFlink) is below.
Thank you,
Sumaiya Azad
----------------------------------------------------------------------
Flink: a Top-N query keeps an outdated row
Title: Streaming Top-N over a ROW_NUMBER "latest row" query keeps a stale row
(outer Rank uses AppendFastStrategy on an updating input)
Version: Flink 2.2.1. Reproduced two ways on 2026-09-28: with the SQL client on
a 2.2.1 session cluster (Java 17), and with PyFlink 2.2.1 (Java 21).
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 A's old row, so the
result keeps the old value. Batch mode gives the right answer.
How to reproduce
Four rows. Key A first has value 100, then a later value 1. Key B has 50. The
last row is a redelivered copy.
```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
```
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.
Expected
`('B', 50.0, 1)`. A's latest value is 1, so B is on top.
Actual
*
Streaming mode: the changelog is a single `+I ('A', 100.0, 1)`. A's old
*
value 100 is never removed.
*
Batch mode: `('B', 50.0, 1)`, which is correct.
Likely cause
The streaming plan runs both ranks with `AppendFastStrategy`:
```
Rank(strategy=[AppendFastStrategy], ... orderBy=[v DESC, k ASC], select=[k, v])
+- Rank(strategy=[AppendFastStrategy], ... partitionBy=[k], orderBy=[$4 DESC,
id DESC])
```
`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.
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 parallelism 2, we saw this in eight
such queries, even with events delivered in time order.
