Bettle created FLINK-41085:
------------------------------
Summary: AppendOnlyTopNFunction emits an inconsistent changelog
after cache eviction when sorting on NaN
Key: FLINK-41085
URL: https://issues.apache.org/jira/browse/FLINK-41085
Project: Flink
Issue Type: Bug
Reporter: Bettle
Related issues: FLINK-40927 and FLINK-27627.
Changing only the TopN cache capacity changes the emitted changelog when the
sort field contains NaN. After cache eviction, rebuilding the buffer from keyed
state can silently lose a record and subsequently emit an additional INSERT
without a corresponding DELETE.
*Environment*
||Setting||Value||
|Flink version|1.20.3 release dependencies|
|Java|Temurin 8u492|
|Operator|AppendOnlyTopNFunction|
|State backend|HashMapStateBackend|
|State TTL|Disabled|
|Input|Append-only|
|Partition key|city|
|Sort order|amount DOUBLE DESC|
|Rank range|ROW_NUMBER, [1, 2]|
|Output rank column|Disabled|
|Offset|None|
The reproduction uses a Java operator test harness with released operator
bytecode, not copied or modified operator sources. Partition and sort keys are
generated through KeySelectorUtil.getRowDataSelector(), as used by
StreamExecRank, producing BinaryRowData keys.
*Reproduction*
Process the following records in order:
||Step||City||Record||Amount||
|1|Beijing|A|NaN|
|2|Beijing|B|20.0|
|3|Shanghai|X|77.0|
|4|Beijing|C|19.0|
Run the same input twice, changing only the cacheSize constructor argument:
||Run||cacheSize||Cache behavior||
|Without eviction|10000|Beijing's buffer remains cached|
|With eviction|2|For Top2, only one partition is cached; Shanghai evicts
Beijing|
When C19 arrives in the second run, Beijing's buffer must be reconstructed from
state. TTL is disabled, so no state expiration is involved.
*Actual output*
The following table shows only the Beijing changelog:
||Output order||Without cache eviction||With cache eviction||
|1|INSERT A(NaN)|INSERT A(NaN)|
|2|INSERT B(20.0)|INSERT B(20.0)|
|3|No further output|INSERT C(19.0)|
|DELETE emitted|None|None|
|Records retained downstream|2|3|
No DELETE accompanies the additional INSERT. A downstream consumer applying
this changelog retains three Beijing records, although the requested result is
Top2.
The assertion that both runs should produce identical changelogs failed with
the same difference in three separate Maven executions.
*Observed state contents*
The actual MapState contents for Beijing were inspected during execution:
||After processing||State key||Stored records||
|A(NaN)|NaN|[A(NaN)]|
|B(20.0)|NaN|[A(NaN)]|
|B(20.0)|20.0|[A(NaN), B(20.0)]|
After B20, there are two distinct state entries. The original NaN entry remains
unchanged, while the entry under 20.0 contains both records.
*Observed reconstruction sequence*
On Beijing's cache miss, a read-only tracer recorded the following actual
loading sequence:
||Loading order||State key||Loaded records||Buffer after loading||Record count||
|First|20.0|[A(NaN), B(20.0)]|20.0 → [A(NaN), B(20.0)]|2|
|Second|NaN|[A(NaN)]|20.0 → [A(NaN)]|1|
The second load replaces the first list. The reconstructed buffer contains only
one record, so C19 is subsequently admitted and emitted as an INSERT.
The same output difference was also verified without the tracing proxy.
*Cause*
The generated FLOAT/DOUBLE comparator uses the following expression:
a > b ? 1 : a < b ? -1 : 0
It returns 0 when comparing NaN with 20.0, although the generated binary state
keys are not equal.
This creates inconsistent key semantics between the in-memory TopN buffer and
MapState:
||Operation||Treatment of NaN and 20.0||
|MapState key equality|Different keys; both entries are retained|
|TopNBuffer TreeMap comparison|Comparator-equivalent keys; one entry is used|
AppendOnlyTopNFunction writes the complete comparator-equivalent group using
the current input's sort key:
dataState.put(sortKey, new ArrayList<>(inputs));
During cache reconstruction, initHeapStates() iterates the state entries and
calls:
buffer.putAll(sortKey, values);
TopNBuffer.putAll() calls TreeMap.put(). Because the comparator considers the
keys equal, the later list replaces the earlier list. Consequently, the
state-entry loading order can affect the reconstructed buffer and subsequent
output.
*Expected behavior*
Changing the capacity of an optimization cache must not change the logical TopN
contents or the emitted changelog.
Rebuilding the TopN buffer should preserve its logical contents and must not
silently lose a previously emitted record.
The broader NaN ordering semantics may require discussion under FLINK-40927,
but cache eviction should not change query results.
*Scope*
This is an operator-level reproduction using Flink 1.20.3 release dependencies,
not an end-to-end SQL job. It does not involve checkpoint recovery.
If this consequence should be tracked directly under FLINK-40927, please feel
free to mark this report as a duplicate.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
