code: def arith_operate(lhs: DataFrame, rhs, op) -> "DataFrame":
      data_manager = lhs.data_manager.duplicate()
      block_indexes = data_manager.infer_operable_blocks()
      column_names = data_manager.infer_operable_field_names()
  
      if isinstance(rhs, DataFrame):
          rhs_column_names = (
              rhs.data_manager.infer_operable_field_names()
          )
          if (
              len(column_names) != len(rhs_column_names)
              or len(column_names) > 1
          ):
              raise ValueError(
                  f"Operation={op} of two dataframe should have same column length=1"
              )
  
          rhs_block_id = rhs.data_manager.infer_operable_blocks()[0]
          block_table = _operate(
              lhs.block_table,
              rhs.block_table,
              op,
              block_indexes,
              rhs_block_id,
          )
          to_promote_blocks = data_manager.try_to_promote_types(
              block_indexes,
              rhs.data_manager.get_block(rhs_block_id).block_type,
          )
      elif isinstance(rhs, (np.ndarray, list, pd.Series)):
          if isinstance(rhs, pd.Series):
              rhs = series_to_ndarray(rhs, column_names)
          if isinstance(rhs, list):
              rhs = np.array(rhs)
          if len(rhs.shape) > 2:
              raise ValueError("NdArray's Dimension should <= 2")
          if len(column_names) != rhs.size:
              raise ValueError(
                  f"Size of List/NDArray should = {len(lhs.schema.columns)}"
              )
          rhs = rhs.reshape(-1)
          field_indexes = [
              data_manager.get_field_offset(name)
              for name in column_names
          ]
          field_indexes_mappings = dict(
              zip(field_indexes, range(len(field_indexes)))
          )
          rhs_blocks = [
              np.array([]) for i in range(data_manager.block_num)
          ]
          rhs_types = []
          for bid in block_indexes:
              indexer = [
                  field_indexes_mappings[field]
                  for field in data_manager.get_block(bid).field_indexes
              ]
              rhs_blocks[bid] = rhs[indexer]
              rhs_types.append(rhs_blocks[bid].dtype)
  
          block_table = binary_operate(
              lhs.block_table, rhs_blocks, op, block_indexes
          )
          to_promote_blocks = data_manager.try_to_promote_types(
              block_indexes, rhs_types
          )
  
      elif isinstance(
          rhs,
          (
              bool,
              int,
              float,
              np.int32,
              np.float32,
              np.int64,
              np.float64,
              np.bool_,
          ),
      ):
          block_table = binary_operate(
              lhs.block_table, rhs, op, block_indexes
          )
          to_promote_blocks = data_manager.try_to_promote_types(
              block_indexes, rhs
          )
      else:
          raise ValueError(
              f"Operation={op} between dataframe and {type(rhs)} is not implemented"
          )
  
      if to_promote_blocks:
          block_table, data_manager = promote_types(
              block_table, data_manager, to_promote_blocks
          )
  
      return type(lhs)(
          lhs._ctx,
          block_table,
          lhs.partition_order_mappings,
          data_manager,
      )
  
file_path: FATE/python/fate/arch/dataframe/ops/_arithmetic.py
imports: import numpy
  import pandas
  from fate.arch.computing.api import is_table
  from .._dataframe import DataFrame
  from ._promote_types import promote_types
  from .utils.series_align import series_to_ndarray
  from .utils.operators import binary_operate
label_desc: Use image node_id 1 for calling a global function with example usage: arith_operate(lhs, rhs, op) and returns: type