This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new e8f946a1b47 [FLINK-41104][python] Return true from MapState.contains
for keys mapped to None
e8f946a1b47 is described below
commit e8f946a1b4767c37b4e2a1d977f90d2514bf49dd
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Sat Oct 24 02:58:18 2026 +0900
[FLINK-41104][python] Return true from MapState.contains for keys mapped to
None
This closes #29599.
---
flink-python/pyflink/fn_execution/state_impl.py | 10 ++-
.../pyflink/fn_execution/tests/test_state_impl.py | 72 +++++++++++++++++++++-
2 files changed, 78 insertions(+), 4 deletions(-)
diff --git a/flink-python/pyflink/fn_execution/state_impl.py
b/flink-python/pyflink/fn_execution/state_impl.py
index 5c7ee243197..42cd672cc7f 100644
--- a/flink-python/pyflink/fn_execution/state_impl.py
+++ b/flink-python/pyflink/fn_execution/state_impl.py
@@ -823,10 +823,14 @@ class InternalSynchronousMapRuntimeState(object):
def contains(self, map_key):
if self._is_empty:
return False
- if self.get(map_key) is None:
+ if map_key in self._write_cache:
+ exists, _ = self._write_cache[map_key]
+ return exists
+ if self._cleared:
return False
- else:
- return True
+ exists, _ = self._map_state_handler.blocking_get(
+ self._state_key, map_key, self._map_key_encoder,
self._map_value_decoder)
+ return exists
def is_empty(self):
if self._is_empty:
diff --git a/flink-python/pyflink/fn_execution/tests/test_state_impl.py
b/flink-python/pyflink/fn_execution/tests/test_state_impl.py
index 73a3f2ada5b..8522b9dde05 100644
--- a/flink-python/pyflink/fn_execution/tests/test_state_impl.py
+++ b/flink-python/pyflink/fn_execution/tests/test_state_impl.py
@@ -21,13 +21,14 @@ import unittest
from unittest import mock
from apache_beam.coders import VarIntCoder, StrUtf8Coder
+from apache_beam.portability.api import beam_fn_api_pb2
from apache_beam.runners.worker.sdk_worker import GlobalCachingStateHandler,
StateHandler
from apache_beam.runners.worker.statecache import StateCache
from pyflink.common import Time, Types
from pyflink.datastream.state import StateTtlConfig, MapStateDescriptor
from pyflink.fn_execution.state_impl import RemoteKeyedStateBackend,
RemoteOperatorStateBackend, \
- CachingMapStateHandler
+ CachingMapStateHandler, InternalSynchronousMapRuntimeState
class StateTtlReadCacheTests(unittest.TestCase):
@@ -432,5 +433,74 @@ class StateTtlReadCacheTests(unittest.TestCase):
self.assertFalse(handler._state_cache.is_cache_enabled())
+class InternalSynchronousMapRuntimeStateTests(unittest.TestCase):
+
+ def setUp(self):
+ self.state = InternalSynchronousMapRuntimeState(
+ InMemoryMapStateHandler(), beam_fn_api_pb2.StateKey(),
+ StrUtf8Coder(), StrUtf8Coder(), 16)
+
+ def test_contains_distinguishes_missing_keys_from_none_values(self):
+ values = {'none': None, 'empty': '', 'value': 'value'}
+ for key, value in values.items():
+ self.state.put(key, value)
+
+ for committed in (False, True):
+ if committed:
+ self.state.commit()
+ with self.subTest(committed=committed):
+ for key, value in values.items():
+ self.assertEqual(self.state.get(key), value)
+ self.assertTrue(self.state.contains(key))
+ self.assertIsNone(self.state.get('missing'))
+ self.assertFalse(self.state.contains('missing'))
+
+ def test_contains_after_remove(self):
+ self.state.put('x', None)
+ self.state.commit()
+ self.state.remove('x')
+
+ self.assertFalse(self.state.contains('x'))
+ self.state.commit()
+ self.assertFalse(self.state.contains('x'))
+
+ def test_contains_after_clear_and_put(self):
+ self.state.put('x', 'old')
+ self.state.commit()
+ self.state.clear()
+ self.assertFalse(self.state.contains('x'))
+
+ self.state.put('y', None)
+
+ self.assertFalse(self.state.contains('x'))
+ self.assertTrue(self.state.contains('y'))
+ self.state.commit()
+ self.assertFalse(self.state.contains('x'))
+ self.assertTrue(self.state.contains('y'))
+
+
+class InMemoryMapStateHandler:
+
+ def __init__(self):
+ self.values = {}
+
+ def blocking_get(self, state_key, map_key, map_key_encoder,
map_value_decoder):
+ return map_key in self.values, self.values.get(map_key)
+
+ def extend(self, state_key, items, map_key_encoder, map_value_encoder):
+ for flag, key, value in items:
+ if flag == CachingMapStateHandler.DELETE:
+ self.values.pop(key, None)
+ elif flag == CachingMapStateHandler.SET_NONE:
+ self.values[key] = None
+ elif flag == CachingMapStateHandler.SET_VALUE:
+ self.values[key] = value
+ else:
+ raise ValueError('Unknown map state update flag: %s' % flag)
+
+ def clear(self, state_key):
+ self.values.clear()
+
+
if __name__ == '__main__':
unittest.main()
