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. A regression test with a controllable clock
should cover both a
False value established by put() and one populated by is_empty(), followed by
expiration, backend cleanup, and another is_empty() call.
--
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]
