code: def extract_user_defined_function(
      user_defined_function_proto,
      pandas_udaf=False,
      one_arg_optimization=False,
  ) -> Tuple[str, Dict, List]:
      """
      Extracts user-defined-function from the proto representation of a
      :class:`UserDefinedFunction`.
  
      :param user_defined_function_proto: the proto representation of the Python
      :param pandas_udaf: whether the user_defined_function_proto is pandas udaf
      :param one_arg_optimization: whether the optimization enabled
      :class:`UserDefinedFunction`
      """
  
      def _next_func_num():
          global _func_num
          _func_num = _func_num + 1
          return _func_num
  
      def _extract_input(args) -> Tuple[str, Dict, List]:
          local_variable_dict = {}
          local_funcs = []
          args_str = []
          for arg in args:
              if arg.HasField("udf"):
                  # for chaining Python UDF input: the input argument is a Python ScalarFunction
                  (
                      udf_arg,
                      udf_variable_dict,
                      udf_funcs,
                  ) = extract_user_defined_function(
                      arg.udf, one_arg_optimization=one_arg_optimization
                  )
                  args_str.append(udf_arg)
                  local_variable_dict.update(udf_variable_dict)
                  local_funcs.extend(udf_funcs)
              elif arg.HasField("inputOffset"):
                  if one_arg_optimization:
                      args_str.append("value")
                  else:
                      # 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)
                  args_str.append(constant_value_name)
                  local_variable_dict[
                      constant_value_name
                  ] = parsed_constant_value
          return ",".join(args_str), local_variable_dict, local_funcs
  
      variable_dict = {}
      user_defined_funcs = []
  
      user_defined_func = pickle.loads(
          user_defined_function_proto.payload
      )
      if pandas_udaf:
          user_defined_func = PandasAggregateFunctionWrapper(
              user_defined_func
          )
      func_name = "f%s" % _next_func_num()
      if isinstance(
          user_defined_func, DelegatingScalarFunction
      ) or isinstance(user_defined_func, DelegationTableFunction):
          if user_defined_function_proto.is_pandas_udf:
              variable_dict[func_name] = partial(
                  check_pandas_udf_result, user_defined_func.func
              )
          else:
              variable_dict[func_name] = user_defined_func.func
      else:
          variable_dict[func_name] = user_defined_func.eval
      user_defined_funcs.append(user_defined_func)
  
      func_args, input_variable_dict, input_funcs = _extract_input(
          user_defined_function_proto.inputs
      )
      variable_dict.update(input_variable_dict)
      user_defined_funcs.extend(input_funcs)
      if user_defined_function_proto.takes_row_as_input:
          if input_variable_dict:
              # for constant or other udfs as input arguments.
              func_str = "%s(%s)" % (func_name, func_args)
          elif user_defined_function_proto.is_pandas_udf or pandas_udaf:
              # for pandas udf/udaf, the input data structure is a List of Pandas.Series
              # we need to merge these Pandas.Series into a Pandas.DataFrame
              variable_dict[
                  "wrap_input_series_as_dataframe"
              ] = wrap_input_series_as_dataframe
              func_str = "%s(wrap_input_series_as_dataframe(%s))" % (
                  func_name,
                  func_args,
              )
          else:
              # directly use `value` as input argument
              # e.g.
              # lambda value: Row(value[0], value[1])
              #   can be optimized to
              # lambda value: value
              func_str = "%s(value)" % func_name
      else:
          func_str = "%s(%s)" % (func_name, func_args)
      return func_str, variable_dict, user_defined_funcs
  
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 6 for calling a global function with example usage: extract_user_defined_function(user_defined_function_proto, pandas_udaf, one_arg_optimization) and returns: func_str,  variable_dict,  user_defined_funcs, _func_num, str,  local_variable_dict,  local_funcs