Aaron He created FLINK-41095:
--------------------------------
Summary: MapStateAdaptor does not remove keys mapped to null in
asyncPut and asyncPutAll
Key: FLINK-41095
URL: https://issues.apache.org/jira/browse/FLINK-41095
Project: Flink
Issue Type: Bug
Components: Runtime / State Backends
Reporter: Aaron He
h2. Problem
The State V2 MapState contract specifies that asyncPut(key, null) and
null-valued entries in asyncPutAll(map) remove the corresponding keys, an empty
map is a no-op, and a null map reference is forbidden.
MapStateAdaptor (which backs State V2 MapState on V1 state backends such as
HashMapStateBackend and EmbeddedRocksDBStateBackend via
AsyncKeyedStateBackendAdaptor) delegates asyncPut and asyncPutAll directly to
the V1 InternalMapState#put and #putAll methods:
* On HeapMapState, put(k, null) and putAll with null entry values store a null
value in the map instead of removing the key, so asyncContains(k) remains true
and asyncKeys()/asyncEntries() still emit the key.
* On RocksDBMapState, putAll serializes null values as null markers instead of
deleting the entries, so asyncContains(k) remains true.
h2. Proposal
In MapStateAdaptor:
* Route null values in asyncPut to delegatedState.remove(key).
* In asyncPutAll, validate checkNotNull(map, "Map must not be null"), return a
completed future on an empty map without calling the delegate, and when the map
contains null values, remove those keys via delegatedState.remove(key) while
forwarding any non-null entries to delegatedState.putAll.
h2. Public source
[MapState.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-core-api/src/main/java/org/apache/flink/api/common/state/v2/MapState.java#L60-L90]
[MapStateAdaptor.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/MapStateAdaptor.java#L61-L90]
h2. Tests and acceptance
{noformat}
./mvnw -pl flink-runtime -Dtest=AsyncKeyedStateBackendAdaptorTest test
./mvnw -pl flink-state-backends/flink-statebackend-rocksdb
-Dtest=EmbeddedRocksDBStateBackendTest test
{noformat}
h2. Proposed release note
MapStateAdaptor now removes keys when null values are passed to asyncPut or
asyncPutAll and rejects a null map argument, matching the State V2 MapState
contract.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
