This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new 43f0b465d50 [FLINK-41104][python] Return true from MapState.contains
for keys mapped to None
43f0b465d50 is described below
commit 43f0b465d5033962ae70e96e0d9fa65b7897a6fa
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 | 98 ++++++++++++++++++++++
2 files changed, 105 insertions(+), 3 deletions(-)
diff --git a/flink-python/pyflink/fn_execution/state_impl.py
b/flink-python/pyflink/fn_execution/state_impl.py
index e06da081713..7b3694727b8 100644
--- a/flink-python/pyflink/fn_execution/state_impl.py
+++ b/flink-python/pyflink/fn_execution/state_impl.py
@@ -809,10 +809,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 is None:
diff --git a/flink-python/pyflink/fn_execution/tests/test_state_impl.py
b/flink-python/pyflink/fn_execution/tests/test_state_impl.py
new file mode 100644
index 00000000000..30f42e73ff3
--- /dev/null
+++ b/flink-python/pyflink/fn_execution/tests/test_state_impl.py
@@ -0,0 +1,98 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import unittest
+
+from apache_beam.coders import StrUtf8Coder
+from apache_beam.portability.api import beam_fn_api_pb2
+
+from pyflink.fn_execution.state_impl import CachingMapStateHandler, \
+ InternalSynchronousMapRuntimeState
+
+
+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()
