[
https://issues.apache.org/jira/browse/FLINK-41095?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41095:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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)
