code: def bucketize(df: DataFrame, boundaries: Union[pd.DataFrame, dict]):
      if isinstance(boundaries, pd.DataFrame):
          boundaries = dict(
              [
                  (_name, boundaries[_name].tolist())
                  for _name in boundaries
              ]
          )
      elif not isinstance(boundaries, dict):
          raise ValueError("boundaries should be pd.DataFrame or dict")
  
      data_manager = df.data_manager.duplicate()
      field_names = list(
          filter(
              lambda field_name: field_name in boundaries,
              data_manager.infer_operable_field_names(),
          )
      )
      blocks_loc = data_manager.loc_block(field_names)
  
      _boundaries_list = []
      for name, (_bid, _) in zip(field_names, blocks_loc):
          if BlockType.is_tensor(data_manager.blocks[_bid].block_type):
              _boundary = torch.tensor(
                  boundaries[name], dtype=torch.float64
              )
              _boundary[-1] = torch.inf
          else:
              _boundary = np.array(boundaries[name], dtype=np.float64)
              _boundary[-1] = np.inf
  
          _boundaries_list.append((_bid, _, _boundary))
  
      narrow_blocks, dst_blocks = data_manager.split_columns(
          field_names, BlockType.get_block_type(BUCKETIZE_RESULT_TYPE)
      )
  
      def _mapper(
          blocks,
          boundaries_list: list = None,
          narrow_loc: list = None,
          dst_bids: list = None,
          dm: DataManager = None,
      ):
          ret_blocks = [block for block in blocks]
  
          for i in range(len(ret_blocks), dm.block_num):
              ret_blocks.append([])
  
          for bid, offsets in narrow_loc:
              ret_blocks[bid] = ret_blocks[bid][:, offsets]
  
          for dst_bid, (src_bid, src_offset, boundary) in zip(
              dst_bids, boundaries_list
          ):
              if isinstance(blocks[src_bid], torch.Tensor):
                  ret = torch.bucketize(
                      blocks[src_bid][:, [src_offset]],
                      boundary,
                      out_int32=False,
                  )
              else:
                  ret = np.digitize(
                      blocks[src_bid][:, [src_offset]], boundary
                  )
  
              ret_blocks[dst_bid] = dm.blocks[dst_bid].convert_block(
                  ret
              )
  
          return ret_blocks
  
      bucketize_mapper = functools.partial(
          _mapper,
          boundaries_list=_boundaries_list,
          narrow_loc=narrow_blocks,
          dst_bids=dst_blocks,
          dm=data_manager,
      )
  
      block_table = df.block_table.mapValues(bucketize_mapper)
  
      block_indexes = data_manager.infer_operable_blocks()
      if len(block_indexes) > 1:
          to_promote_types = []
          for _bid in block_indexes:
              to_promote_types.append(
                  (_bid, data_manager.get_block(_bid).block_type)
              )
  
          data_manager.promote_types(to_promote_types)
          block_table, data_manager = compress_blocks(
              block_table, data_manager
          )
  
      return DataFrame(
          df._ctx,
          block_table,
          partition_order_mappings=df.partition_order_mappings,
          data_manager=data_manager,
      )
  
file_path: FATE/python/fate/arch/dataframe/ops/_encoder.py
imports: import functools
  import pandas
  import numpy
  import torch
  from sklearn.preprocessing import OneHotEncoder
  from typing import Union
  from ._compress_block import compress_blocks
  from .._dataframe import DataFrame
  from ..manager import BlockType, DataManager
label_desc: Use image node_id 4 for calling a global function with example usage: bucketize(df, boundaries) and returns: DataFrame, ret_blocks