[GitHub] [flink] HuangXingBo commented on a diff in pull request #20375: [FLINK-28559][python] Support DataStream PythonKeyedProcessOperator in Thread Mode

[GitBox <gi...@apache.org>]

HuangXingBo commented on code in PR #20375:
URL: https://github.com/apache/flink/pull/20375#discussion_r932955318


##########
flink-python/pyflink/fn_execution/embedded/converters.py:
##########
@@ -214,9 +228,23 @@ def from_field_type_proto(field_type):
             [from_field_type_proto(f.type) for f in field_type.row_schema.fields],
             [f.name for f in field_type.row_schema.fields])
     elif type_name == schema_type_name.BASIC_ARRAY:
-        return ListDataConverter(from_field_type_proto(field_type.collection_element_type))
+        return ArrayDataConverter(from_field_type_proto(field_type.collection_element_type))
     elif type_name == schema_type_name.MAP:
         return DictDataConverter(from_field_type_proto(field_type.map_info.key_type),
                                  from_field_type_proto(field_type.map_info.value_type))
 
     return IdentityDataConverter()
+
+
+def from_type_info(type_info: TypeInformation):
+    if isinstance(type_info, (PickledBytesTypeInfo, RowTypeInfo, TupleTypeInfo)):

Review Comment:
   Because pemja doesn't recognize these two types, which means that we need data converters to transform these state data. I think we can use this as an optimization later. WDYT
   



##########
flink-python/pyflink/datastream/data_stream.py:
##########
@@ -1205,40 +1211,47 @@ def __init__(self, reduce_function):
                     self._open_func = None
                     self._close_func = None
                     self._reduce_function = reduce_function
-                self._reduce_value_state = None  # type: ValueState
+                self._reduce_state = None  # type: ReducingState
+                self._in_batch_execution_mode = True
+                self._has_started_key_set = set()
 
             def open(self, runtime_context: RuntimeContext):
                 if self._open_func:
                     self._open_func(runtime_context)
 
-                self._reduce_value_state = runtime_context.get_state(
-                    ValueStateDescriptor("_reduce_state" + str(uuid.uuid4()), output_type))
-                from pyflink.fn_execution.datastream.process.runtime_context import (
-                    StreamingRuntimeContext)
-                self._in_batch_execution_mode = \
-                    cast(StreamingRuntimeContext, runtime_context)._in_batch_execution_mode
+                self._reduce_state = runtime_context.get_reducing_state(
+                    ReducingStateDescriptor(
+                        "_reduce_state" + str(uuid.uuid4()),
+                        self._reduce_function,
+                        output_type))
+
+                if python_execution_mode == "process":
+                    from pyflink.fn_execution.datastream.process.runtime_context import (
+                        StreamingRuntimeContext)
+                    self._in_batch_execution_mode = (
+                        cast(StreamingRuntimeContext, runtime_context)._in_batch_execution_mode)
+                else:
+                    self._in_batch_execution_mode = runtime_context.get_job_parameter(
+                        "inBatchExecutionMode", "false") == "true"
 
             def close(self):
                 if self._close_func:
                     self._close_func()
 
             def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
-                reduce_value = self._reduce_value_state.value()
-                if reduce_value is not None:
-                    reduce_value = self._reduce_function(reduce_value, value)
-                else:
-                    # register a timer for emitting the result at the end when this is the
-                    # first input for this key
-                    if self._in_batch_execution_mode:
+                self._reduce_state.add(value)
+                if self._in_batch_execution_mode:
+                    key = ctx.get_current_key()
+                    if isinstance(key, list):
+                        key = tuple(key)
+                    if key not in self._has_started_key_set:
                         ctx.timer_service().register_event_time_timer(0x7fffffffffffffff)
-                    reduce_value = value
-                self._reduce_value_state.update(reduce_value)
-                if not self._in_batch_execution_mode:
-                    # only emitting the result when all the data for a key is received
-                    yield reduce_value
+                        self._has_started_key_set.add(key)

Review Comment:
   Yes. Good catch



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org