code: def loc(
      df: DataFrame, indexer, target="sample_id", preserve_order=False
  ):
      """
      indexer: table, key=sample_id, value=(block_id, block_offset)
      """
      if target != "sample_id":
          raise ValueError(
              f"Only target=sample_id is supported, but target={target} is found"
          )
      flatten_table = flatten_data(df, key_type="sample_id")
      if not preserve_order:
          flatten_table = flatten_table.join(indexer, lambda v1, v2: v1)
          if not flatten_table.count():
              return df.empty_frame()
          return transform_flatten_data_to_df(
              df._ctx,
              flatten_table,
              df.data_manager,
              key_type="sample_id",
              value_with_sample_id=False,
          )
      else:
          flatten_table_with_dst_indexer = flatten_table.join(
              indexer, lambda v1, v2: (v2[0], (v2[1], v1))
          )
          if not flatten_table_with_dst_indexer.count():
              return df.empty_frame()
  
          def _aggregate(kvs):
              values = [value for key, value in kvs]
              values.sort()
              i = 0
              l = len(values)
              while i < l:
                  j = i + 1
                  while j < l and values[j][0] == values[i][0]:
                      j += 1
  
                  yield values[i][0], [
                      values[k][1] for k in range(i, j)
                  ]
  
                  i = j
  
          data_manager = df.data_manager.duplicate()
          block_num = data_manager.block_num
  
          def _to_blocks(values):
              block_size = len(values)
              ret_blocks = [
                  [None] * block_size for _ in range(block_num)
              ]
  
              for row_id, row_data in values:
                  for j in range(block_num):
                      ret_blocks[j][row_id] = row_data[j]
  
              for idx, block_schema in enumerate(data_manager.blocks):
                  ret_blocks[idx] = block_schema.convert_block(
                      ret_blocks[idx]
                  )
  
              return ret_blocks
  
          agg_data = flatten_table_with_dst_indexer.mapReducePartitions(
              _aggregate, lambda v1, v2: v1 + v2
          )
          block_table = agg_data.mapValues(_to_blocks)
  
          partition_order_mappings = (
              get_partition_order_mappings_by_block_table(
                  block_table,
                  block_row_size=data_manager.block_row_size,
              )
          )
  
          return DataFrame(
              df._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 8 for calling a global function with example usage: loc(df, indexer, target, preserve_order) and returns: transform_flatten_data_to_df, DataFrame, df, df, ret_blocks