code: def get_partition_order_by_raw_table(
      table, block_row_size, key_type="sample_id"
  ):
      def _get_block_summary(kvs):
          try:
              if key_type == "sample_id":
                  key = next(kvs)[0]
              else:
                  key = next(kvs)[1][0]
  
              block_size = 1 + sum(1 for kv in kvs)
          except StopIteration:
              key, block_size = 0, 0
  
          return {key: block_size}
  
      block_summary = table.applyPartitions(_get_block_summary).reduce(
          lambda blk1, blk2: {**blk1, **blk2}
      )
  
      start_index, acc_block_num = 0, 0
      block_order_mappings = dict()
  
      if not block_summary:
          return block_order_mappings
  
      for blk_key, blk_size in sorted(block_summary.items()):
          block_num = (blk_size + block_row_size - 1) // block_row_size
          block_order_mappings[blk_key] = dict(
              start_index=start_index,
              end_index=start_index + blk_size - 1,
              start_block_id=acc_block_num,
              end_block_id=acc_block_num + block_num - 1,
          )
  
          start_index += blk_size
          acc_block_num += block_num
  
      return block_order_mappings
  
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 3 for calling a global function with example usage: get_partition_order_by_raw_table(table, block_row_size, key_type) and returns: block_order_mappings, dict, block_order_mappings