code: def apply_row(
      df: "DataFrame",
      func,
      columns: list = None,
      with_label=False,
      with_weight=False,
      enable_type_align_checking=True,
  ) -> "DataFrame":
      """
      In current version, assume that the apply_row results' lengths are equal
      """
      data_manager = df.data_manager
      dst_data_manager, _ = data_manager.derive_new_data_manager(
          with_sample_id=True,
          with_match_id=True,
          with_label=not with_label,
          with_weight=not with_weight,
          columns=None,
      )
  
      non_operable_field_names = dst_data_manager.get_field_name_list()
      non_operable_blocks = [
          data_manager.loc_block(field_name, with_offset=False)
          for field_name in non_operable_field_names
      ]
      fields_loc = data_manager.get_fields_loc(
          with_sample_id=False,
          with_match_id=False,
          with_label=with_label,
          with_weight=with_weight,
      )
  
      fields_name = data_manager.get_field_name_list(
          with_sample_id=False,
          with_match_id=False,
          with_label=with_label,
          with_weight=with_weight,
      )
  
      operable_blocks = sorted(
          list(
              set(
                  data_manager.loc_block(fields_name, with_offset=False)
              )
          )
      )
      is_numeric = True
      for bid in operable_blocks:
          if not data_manager.get_block(bid).is_numeric():
              is_numeric = False
              break
      block_column_in_orders = list()
      if is_numeric:
          for bid in operable_blocks:
              field_indexes = data_manager.get_block(bid).field_indexes
              block_column_in_orders.extend(
                  [
                      data_manager.get_field_name(field_index)
                      for field_index in field_indexes
                  ]
              )
  
      _apply_func = functools.partial(
          _apply,
          func=func,
          src_operable_blocks=operable_blocks,
          src_field_names=fields_name,
          src_fields_loc=fields_loc,
          src_non_operable_blocks=non_operable_blocks,
          ret_columns=columns,
          dst_dm=dst_data_manager,
          is_numeric=is_numeric,
          need_shuffle=True
          if block_column_in_orders == fields_name
          else False,
          block_column_in_orders=block_column_in_orders,
          enable_type_align_checking=enable_type_align_checking,
      )
  
      dst_block_table_with_dm = df.block_table.mapValues(_apply_func)
  
      dst_data_manager = dst_block_table_with_dm.first()[1][1]
      dst_block_table = dst_block_table_with_dm.mapValues(
          lambda blocks_with_dm: blocks_with_dm[0]
      )
  
      return DataFrame(
          df._ctx,
          dst_block_table,
          df.partition_order_mappings,
          dst_data_manager,
      )
  
file_path: FATE/python/fate/arch/dataframe/ops/_apply_row.py
imports: import functools
  import pandas
  import torch
  from collections.abc import Iterable
  from .._dataframe import DataFrame
  from ..manager.block_manager import Block, BlockType
  from ..manager.data_manager import DataManager
  from ..utils._auto_column_name_generated import generated_default_column_names
label_desc: Use image node_id 1 for calling a global function with example usage: apply_row(df, func, columns, with_label, with_weight, enable_type_align_checking) and returns: DataFrame