code: def _to_blocks(
      kvs,
      data_manager=None,
      retrieval_index_dict=None,
      partition_order_mappings=None,
      na_values=None,
  ):
      """
      sample_id/match_id,label(maybe missing),weight(maybe missing),X
      """
      block_id = None
  
      schema = data_manager.schema
  
      splits = [[] for _ in range(data_manager.block_num)]
      sample_id_block = (
          data_manager.loc_block(
              schema.sample_id_name, with_offset=False
          )
          if schema.sample_id_name
          else None
      )
  
      match_id_block = (
          data_manager.loc_block(
              schema.match_id_name, with_offset=False
          )
          if schema.match_id_name
          else None
      )
      match_id_column_index = retrieval_index_dict["match_id_index"]
  
      label_block = (
          data_manager.loc_block(schema.label_name, with_offset=False)
          if schema.label_name
          else None
      )
      label_column_index = retrieval_index_dict["label_index"]
  
      weight_block = (
          data_manager.loc_block(schema.weight_name, with_offset=False)
          if schema.weight_name
          else None
      )
      weight_column_index = retrieval_index_dict["weight_index"]
  
      column_indexes = retrieval_index_dict["column_indexes"]
  
      columns = schema.columns
      column_blocks_mapping = dict()
      for col_id, col_name in zip(column_indexes, columns):
          bid = data_manager.loc_block(col_name, with_offset=False)
          if bid not in column_blocks_mapping:
              column_blocks_mapping[bid] = []
  
          column_blocks_mapping[bid].append(col_id)
  
      block_row_size = data_manager.block_row_size
  
      lid = 0
      for key, value in kvs:
          if block_id is None:
              block_id = partition_order_mappings[key]["start_block_id"]
          lid += 1
  
          # columns = value.split(",", -1)
          splits[sample_id_block].append(key)
          if match_id_block:
              splits[match_id_block].append(
                  value[match_id_column_index]
              )
          if label_block:
              splits[label_block].append([value[label_column_index]])
          if weight_block:
              splits[weight_block].append([value[weight_column_index]])
  
          for bid, col_id_list in column_blocks_mapping.items():
              splits[bid].append(
                  [value[col_id] for col_id in col_id_list]
              )
  
          if lid % block_row_size == 0:
              converted_blocks = data_manager.convert_to_blocks(splits)
              yield block_id, converted_blocks
              block_id += 1
              splits = [[] for _ in range(data_manager.block_num)]
  
      if lid % block_row_size:
          converted_blocks = data_manager.convert_to_blocks(splits)
          yield block_id, converted_blocks
  
imports: import functools
  import pandas
  from typing import Union
  from .conf.default_config import DATAFRAME_BLOCK_ROW_SIZE
  from .entity import types
  from ._dataframe import DataFrame
  from .manager import DataManager
  from fate.arch.trace import auto_trace
label_desc: Use image node_id 1 for calling a global function with example usage: _to_blocks(kvs, data_manager, retrieval_index_dict, partition_order_mappings, na_values) without return types