code: def create_table_operation_from_proto(
      proto, input_coder_info, output_coder_into
  ):
      from pyflink.fn_execution.table.operations import (
          TableFunctionOperation,
      )
  
      serialized_fn = parse_function_proto(proto)
  
      input_data_converter = from_schema_proto(
          parse_coder_proto(input_coder_info).flatten_row_type.schema
      )
      output_data_converter = from_schema_proto(
          parse_coder_proto(output_coder_into).flatten_row_type.schema
      )
  
      table_operation = TableFunctionOperation(serialized_fn)
  
      process_element_func = table_operation.process_element
  
      def process_element(value):
          actual_value = input_data_converter.to_internal(value)
          results = process_element_func(actual_value)
          for result in results:
              yield output_data_converter.to_external(result)
  
      table_operation.process_element = process_element
  
      return table_operation
  
file_path: flink/flink-python/pyflink/fn_execution/embedded/operation_utils.py
imports: from pyflink.fn_execution.embedded.operations import OneInputFunctionOperation, TwoInputFunctionOperation
  from pyflink.fn_execution.embedded.converters import from_type_info_proto, from_schema_proto
label_desc: Use image node_id 5 for calling a global function with example usage: create_table_operation_from_proto(proto, input_coder_info, output_coder_into) and returns: table_operation