This is an automated email from the ASF dual-hosted git repository. dianfu pushed a commit to branch release-2.2 in repository https://gitbox.apache.org/repos/asf/flink.git
commit bc2c5472b2cee7205eeca3e16e72d092b9988baf 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) --- .../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 224e57110fe..6a005856a17 100644 --- a/flink-python/pyflink/datastream/tests/test_data_stream.py +++ b/flink-python/pyflink/datastream/tests/test_data_stream.py @@ -1119,6 +1119,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 ad9e6c279f8..42cd672cc7f 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
