code: def _cmp_dfs(
      lhs_block_table,
      rhs_block_table,
      op,
      lhs_block_loc,
      rhs_block_loc,
      block_indexes,
      indexers,
  ):
      block_index_set = set(block_indexes)
  
      def _cmp_partition(l_blocks, r_blocks):
          ret_blocks = [[] for i in range(l_blocks)]
          for bid in range(len(l_blocks)):
              if bid not in block_index_set:
                  ret_blocks[bid] = l_blocks[bid]
  
          for bid, indexer in zip(block_indexes, indexers):
              cmp_ret = torch.empty(l_blocks[bid].shape)
              for idx in indexer:
                  _, l_offset = lhs_block_loc[idx]
                  r_bid, r_offset = rhs_block_loc[idx]
                  cmp_ret[:, l_offset] = op(
                      l_blocks[bid][l_offset], r_blocks[r_bid][r_offset]
                  )
  
              ret_blocks[bid] = cmp_ret
  
          return ret_blocks
  
      block_table = lhs_block_table.join(
          rhs_block_table, _cmp_partition
      )
  
      return block_table
  
file_path: FATE/python/fate/arch/dataframe/ops/_cmp.py
imports: import numpy
  import pandas
  import torch
  from .._dataframe import DataFrame
  from .._dataframe import DataManager
  from ..manager import BlockType
  from .utils.operators import binary_operate
  from .utils.series_align import series_to_ndarray
  from ._compress_block import compress_blocks
label_desc: Use image node_id 3 for calling a global function with example usage: _cmp_dfs(lhs_block_table, rhs_block_table, op, lhs_block_loc, rhs_block_loc, block_indexes, indexers) and returns: block_table, ret_blocks