dianfu commented on code in PR #29576:
URL: https://github.com/apache/flink/pull/29576#discussion_r4358338837
##########
flink-python/pyflink/fn_execution/state_impl.py:
##########
@@ -1109,8 +1117,11 @@ def _create_internal_map_state(
write_cache_size = 0
else:
write_cache_size = self._map_state_write_cache_size
+ map_state_handler = self._map_state_handler
+ if cache_type !=
SynchronousKvRuntimeState.CacheType.ENABLE_READ_WRITE_CACHE:
+ map_state_handler = self._uncached_map_state_handler
Review Comment:
Could we also handle the local `_is_empty` cache in
`InternalSynchronousMapRuntimeState`? Using an uncached state handler does not
disable this cache: `put()` sets `_is_empty = False`, `commit()` leaves it
unchanged, and subsequent `is_empty()` calls return it without consulting the
backend.
For example, with NeverReturnExpired, write and flush an entry, advance time
past its TTL, and call get(key) to trigger removal of the expired backend
entry. is_empty() can still return False even though the backend is now empty.
Could we propagate the effective read-cache policy to the map runtime state
and avoid reusing a cached non-empty result when reads must be uncached,
flushing pending mutations before checking the backend and preserving local
clear/write semantics? Resetting `_is_empty` only at commit would not fully
address this, since is_empty() can cache False again and reuse it after
expiration within the same bundle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
