code: def loc_with_sample_id_replacement(df: DataFrame, indexer):
      """
      indexer: table,
              row: (key=random_key,
              value=(sample_id, (src_block_id, src_offset))
      """
      if indexer.count() == 0:
          return df.empty_frame()
  
      data_manager = df.data_manager
      partition_order_mappings = get_partition_order_by_raw_table(
          indexer, data_manager.block_row_size, key_type="block_id"
      )
  
      def _aggregate(kvs):
          bid, offset = None, 0
          flat_ret = []
          for k, values in kvs:
              sample_id, (src_block_id, src_offset) = values
              if bid is None:
                  bid = partition_order_mappings[sample_id][
                      "start_block_id"
                  ]
  
              flat_ret.append(
                  (src_block_id, src_offset, sample_id, bid, offset)
              )
  
              offset += 1
              if offset == data_manager.block_row_size:
                  offset = 0
                  bid += 1
  
          flat_ret.sort()
          i = 0
          l = len(flat_ret)
          while i < l:
              j = i
              while j < l and flat_ret[i][0] == flat_ret[j][0]:
                  j += 1
  
              agg_ret = [flat_ret[k][1:] for k in range(i, j)]
              yield flat_ret[i][0], agg_ret
  
              i = j
  
      sample_id_index = data_manager.loc_block(
          data_manager.schema.sample_id_name, with_offset=False
      )
      block_num = data_manager.block_num
  
      def _convert_to_row(kvs):
          ret_dict = {}
          for block_id, (blocks, block_indexer) in kvs:
              flat_blocks = [
                  Block.transform_block_to_list(block)
                  for block in blocks
              ]
              for (
                  src_row_id,
                  sample_id,
                  dst_block_id,
                  dst_row_id,
              ) in block_indexer:
                  if dst_block_id not in ret_dict:
                      ret_dict[dst_block_id] = []
  
                  row_data = [
                      flat_blocks[i][src_row_id]
                      for i in range(block_num)
                  ]
                  row_data[sample_id_index] = sample_id
  
                  ret_dict[dst_block_id].append((dst_row_id, row_data))
  
          for dst_block_id, value_list in ret_dict.items():
              yield dst_block_id, sorted(value_list)
  
      agg_indexer = indexer.mapReducePartitions(
          _aggregate, lambda l1, l2: l1 + l2
      )
      block_table = df.block_table.join(
          agg_indexer, lambda v1, v2: (v1, v2)
      )
      block_table = block_table.mapReducePartitions(
          _convert_to_row, _merge_list
      )
  
      def _convert_to_frame_block(blocks, data_manager):
          convert_blocks = []
          for idx, block_schema in enumerate(data_manager.blocks):
              block_content = [row_data[1][idx] for row_data in blocks]
              convert_blocks.append(
                  block_schema.convert_block(block_content)
              )
  
          return convert_blocks
  
      _convert_to_frame_block_func = functools.partial(
          _convert_to_frame_block, data_manager=data_manager
      )
      block_table = block_table.mapValues(_convert_to_frame_block_func)
  
      return DataFrame(
          ctx=df._ctx,
          block_table=block_table,
          partition_order_mappings=partition_order_mappings,
          data_manager=data_manager,
      )
  
file_path: FATE/python/fate/arch/dataframe/ops/_indexer.py
imports: import functools
  import uuid
  from ..manager import Block, DataManager
  from .._dataframe import DataFrame
label_desc: Use image node_id 9 for calling a global function with example usage: loc_with_sample_id_replacement(df, indexer) and returns: DataFrame, df, convert_blocks