code: def transform_flatten_data_to_df(
      ctx,
      flatten_table,
      data_manager: DataManager,
      key_type,
      value_with_sample_id=True,
  ):
      if flatten_table.count() == 0:
          return DataFrame(
              ctx=ctx,
              block_table=flatten_table,
              partition_order_mappings=dict(),
              data_manager=data_manager.duplicate(),
          )
  
      partition_order_mappings = get_partition_order_by_raw_table(
          flatten_table, data_manager.block_row_size, key_type=key_type
      )
      block_num = data_manager.block_num
  
      def _convert_to_blocks(kvs):
          bid = None
          ret_blocks = [[] for _ in range(block_num)]
  
          lid = 0
          for _, value in kvs:
              if value_with_sample_id:
                  data = value[1]
              else:
                  data = value
              lid += 1
              if bid is None:
                  sample_id = data[0]
                  bid = partition_order_mappings[sample_id][
                      "start_block_id"
                  ]
  
              for i in range(block_num):
                  ret_blocks[i].append(data[i])
  
              if lid % data_manager.block_row_size == 0:
                  ret_blocks = [
                      data_manager.blocks[i].convert_block(block)
                      for i, block in enumerate(ret_blocks)
                  ]
                  yield bid, ret_blocks
                  bid += 1
                  ret_blocks = [[] for _ in range(block_num)]
  
          if lid % data_manager.block_row_size:
              ret_blocks = [
                  data_manager.blocks[i].convert_block(block)
                  for i, block in enumerate(ret_blocks)
              ]
              yield bid, ret_blocks
  
      block_table = flatten_table.mapPartitions(
          _convert_to_blocks, use_previous_behavior=False
      )
  
      return DataFrame(
          ctx=ctx,
          block_table=block_table,
          partition_order_mappings=partition_order_mappings,
          data_manager=data_manager.duplicate(),
      )
  
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 7 for calling a global function with example usage: transform_flatten_data_to_df(ctx, flatten_table, data_manager, key_type, value_with_sample_id) and returns: DataFrame, DataFrame