code: def transform_to_tensor(
      block_table,
      data_manager: "DataManager",
      dtype=None,
      partition_order_mappings=None,
  ):
      def _merge_blocks(src_blocks, bids=None, fields=None):
          if len(bids) == 1:
              bid = bids[0]
              t = src_blocks[bid]
          else:
              i = 0
              tensors = []
              while i < len(fields):
                  bid = fields[i][0]
                  indexes = [fields[i][1]]
                  j = i + 1
                  while (
                      j < len(fields)
                      and bid == fields[j][0]
                      and indexes[-1] + 1 == fields[j][1]
                  ):
                      indexes.append(fields[j][1])
                      j += 1
  
                  tensors.append(src_blocks[bid][:, indexes])
                  i = j
  
              t = torch.hstack(tensors)
  
          if dtype:
              t = t.type(getattr(torch, t))
  
          return t
  
      def _convert_to_phe_tensor(
          blocks, bid: int = None, dm: DataManager = None
      ):
          phe_block = dm.get_block(bid)
          return phe_block.convert_to_phe_tensor(
              blocks[bid], shape=(len(blocks[0]), 1)
          )
  
      block_indexes = data_manager.infer_operable_blocks()
      is_phe_tensor = False
      for block_id in block_indexes:
          if data_manager.get_block(block_id).is_phe_tensor():
              is_phe_tensor = True
              break
          if not data_manager.get_block(block_id).is_numeric:
              raise ValueError(
                  "Transform to distributed tensor should ensure every field is numeric"
              )
  
      if is_phe_tensor:
          if len(block_indexes) > 1:
              raise ValueError(
                  "To use as_tensor of phe_tensor type, it should be only single column"
              )
          block_id = block_indexes[0]
          _convert_to_phe_tensor_func = functools.partial(
              _convert_to_phe_tensor, bid=block_id, dm=data_manager
          )
          phe_table = block_table.mapValues(_convert_to_phe_tensor_func)
  
          shape_table = block_table.mapValues(
              lambda blocks: (len(blocks[0]), 1)
          )
          shapes = [
              shape_obj
              for k, shape_obj in sorted(shape_table.collect())
          ]
  
          return tensor.DTensor.from_sharding_table(
              phe_table,
              shapes=shapes,
              dtype=data_manager.get_block(block_id).dtype,
              device=data_manager.get_block(block_id).device,
          )
      else:
          field_names = data_manager.infer_operable_field_names()
          fields_loc = data_manager.loc_block(field_names)
  
          _merged_func = functools.partial(
              _merge_blocks, bids=block_indexes, fields=fields_loc
          )
          merged_table = block_table.mapValues(_merged_func)
  
          shape_table = merged_table.mapValues(lambda v: v.shape)
          shapes = [
              shape_obj
              for k, shape_obj in sorted(shape_table.collect())
          ]
  
          return tensor.DTensor.from_sharding_table(
              merged_table, shapes=shapes
          )
  
file_path: FATE/python/fate/arch/dataframe/ops/_transformer.py
imports: import functools
  import pandas
  from typing import List, Tuple
  import torch
  from fate.arch import tensor
  import numpy
  from ..manager.data_manager import DataManager
label_desc: Use image node_id 1 for calling a global function with example usage: transform_to_tensor(block_table, data_manager, dtype, partition_order_mappings) and returns: t, phe_block, tensor, tensor