This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 69285481f8d [FLINK-41106][python] Iterate key-value pairs in
MapState.put_all (#29597)
69285481f8d is described below
commit 69285481f8d9fe3dc2e86456140582553e9ecce2
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Fri Oct 30 16:05:28 2026 +0900
[FLINK-41106][python] Iterate key-value pairs in MapState.put_all (#29597)
Generated-by: Claude Code 2.1.308 (Claude Opus 5.5)
---
.../pyflink/datastream/tests/test_data_stream.py | 31 ++++++++++++++++++++++
flink-python/pyflink/fn_execution/state_impl.py | 2 +-
2 files changed, 32 insertions(+), 1 deletion(-)
diff --git a/flink-python/pyflink/datastream/tests/test_data_stream.py
b/flink-python/pyflink/datastream/tests/test_data_stream.py
index dbf339e70e2..21e6b0b2bf0 100644
--- a/flink-python/pyflink/datastream/tests/test_data_stream.py
+++ b/flink-python/pyflink/datastream/tests/test_data_stream.py
@@ -1120,6 +1120,37 @@ class ProcessDataStreamTests(DataStreamTests):
"current_value: Row(f0=4, f1='1603708289000')"]
self.assert_equals_sorted(expected, results)
+ def test_map_state_put_all(self):
+ ds = self.env.from_collection([(1, 'a'), (2, 'a')],
+ type_info=Types.TUPLE([Types.INT(),
Types.STRING()]))
+
+ class MyProcessFunction(KeyedProcessFunction):
+
+ def __init__(self):
+ self.int_key_state = None
+ self.str_key_state = None
+
+ def open(self, runtime_context: RuntimeContext):
+ self.int_key_state = runtime_context.get_map_state(
+ MapStateDescriptor('int_key_state', Types.INT(),
Types.STRING()))
+ self.str_key_state = runtime_context.get_map_state(
+ MapStateDescriptor('str_key_state', Types.STRING(),
Types.STRING()))
+
+ def process_element(self, value, ctx):
+ self.int_key_state.put_all({value[0]: value[1]})
+ self.str_key_state.put_all({'ab': 'v%d' % value[0]})
+ yield str(sorted(self.int_key_state.items())) + ' ' + \
+ str(sorted(self.str_key_state.items()))
+
+ ds.key_by(lambda x: x[1], key_type=Types.STRING()) \
+ .process(MyProcessFunction(), output_type=Types.STRING()) \
+ .add_sink(self.test_sink)
+ self.env.execute('test_map_state_put_all')
+ results = self.test_sink.get_results()
+ expected = ["[(1, 'a')] [('ab', 'v1')]",
+ "[(1, 'a'), (2, 'a')] [('ab', 'v2')]"]
+ self.assert_equals_sorted(expected, results)
+
class ProcessDataStreamStreamingTests(DataStreamStreamingTests,
ProcessDataStreamTests,
PyFlinkStreamingTestCase):
diff --git a/flink-python/pyflink/fn_execution/state_impl.py
b/flink-python/pyflink/fn_execution/state_impl.py
index 5e79f50f97b..5c7ee243197 100644
--- a/flink-python/pyflink/fn_execution/state_impl.py
+++ b/flink-python/pyflink/fn_execution/state_impl.py
@@ -799,7 +799,7 @@ class InternalSynchronousMapRuntimeState(object):
self.commit()
def put_all(self, dict_value):
- for map_key, map_value in dict_value:
+ for map_key, map_value in dict_value.items():
self._write_cache[map_key] = (True, map_value)
self._is_empty = False
self._mod_count += 1
