code: def extract_user_defined_aggregate_function(
      current_index,
      user_defined_function_proto,
      distinct_info_dict: Dict[
          Tuple[List[str]], Tuple[List[int], List[int]]
      ],
  ):
      user_defined_agg = load_aggregate_function(
          user_defined_function_proto.payload
      )
      assert isinstance(user_defined_agg, ImperativeAggregateFunction)
      args_str = []
      local_variable_dict = {}
      for arg in user_defined_function_proto.inputs:
          if arg.HasField("inputOffset"):
              # the input argument is a column of the input row
              args_str.append("value[%s]" % arg.inputOffset)
          else:
              # the input argument is a constant value
              (
                  constant_value_name,
                  parsed_constant_value,
              ) = _parse_constant_value(arg.inputConstant)
              for key, value in local_variable_dict.items():
                  if value == parsed_constant_value:
                      constant_value_name = key
                      break
              if constant_value_name not in local_variable_dict:
                  local_variable_dict[
                      constant_value_name
                  ] = parsed_constant_value
              args_str.append(constant_value_name)
  
      if user_defined_function_proto.distinct:
          if tuple(args_str) in distinct_info_dict:
              distinct_info_dict[tuple(args_str)][0].append(
                  current_index
              )
              distinct_info_dict[tuple(args_str)][1].append(
                  user_defined_function_proto.filter_arg
              )
              distinct_index = distinct_info_dict[tuple(args_str)][0][0]
          else:
              distinct_info_dict[tuple(args_str)] = (
                  [current_index],
                  [user_defined_function_proto.filter_arg],
              )
              distinct_index = current_index
      else:
          distinct_index = -1
      if (
          user_defined_function_proto.takes_row_as_input
          and not local_variable_dict
      ):
          # directly use `value` as input argument
          # e.g.
          # lambda value: Row(value[0], value[1])
          #   can be optimized to
          # lambda value: value
          func_str = "lambda value : [value]"
      else:
          func_str = "lambda value : (%s,)" % ",".join(args_str)
      return (
          user_defined_agg,
          eval(func_str, local_variable_dict)
          if args_str
          else lambda v: tuple(),
          user_defined_function_proto.filter_arg,
          distinct_index,
      )
  
imports: import datetime
  import threading
  import time
  from collections.abc import Generator
  from functools import partial
  from typing import Any, Tuple, Dict, List
  from pyflink.common import Row
  from pyflink.fn_execution import pickle
  from pyflink.serializers import PickleSerializer
  from pyflink.table import functions
  from pyflink.table.udf import DelegationTableFunction, DelegatingScalarFunction, ImperativeAggregateFunction, PandasAggregateFunctionWrapper
label_desc: Use image node_id 8 for calling a global function with example usage: extract_user_defined_aggregate_function(current_index, user_defined_function_proto, distinct_info_dict) and returns: user_defined_agg,  unknown,  user_defined_function_proto,  distinct_index