code: def guest_run(ctx, df: DataFrame, curve_type="curve25519", **kwargs):
      curve = Curve25519()
      match_id = df.match_id.block_table
  
      encrypt_func = functools.partial(_encrypt_bytes, curve=curve)
      guest_first_sign_match_id = match_id.mapValues(encrypt_func)
      ctx.hosts.put(GUEST_FIRST_SIGN, guest_first_sign_match_id)
  
      host_first_sign_match_ids = ctx.hosts.get(HOST_FIRST_SIGN)
      host_second_sign_match_ids = []
  
      dh_func = functools.partial(_diffie_hellman, curve=curve)
  
      for i, host_first_sign_match_id in enumerate(
          host_first_sign_match_ids
      ):
          host_second_sign_match_ids.append(
              _flat_block_with_possible_duplicate_keys(
                  host_first_sign_match_ids[i].mapValues(dh_func),
                  duplicate_allow=False,
              )
          )
  
      guest_second_sign_match_ids = ctx.hosts.get(GUEST_SECOND_SIGN)
  
      flat_intersect_id = None
      for guest_second_sign_id, host_second_sign_match_id in zip(
          guest_second_sign_match_ids, host_second_sign_match_ids
      ):
          intersect_eid = guest_second_sign_id.join(
              host_second_sign_match_id,
              lambda id_list_l, id_list_r: (id_list_l, id_list_r),
          )
          intersect_single = _flat_block_key(intersect_eid)
          if not flat_intersect_id:
              flat_intersect_id = intersect_single
          else:
              flat_intersect_id = flat_intersect_id.join(
                  intersect_single,
                  lambda id_list_l, id_list_r: id_list_l + id_list_r,
              )
  
      """
      a.  flatmap=>
          key=(bid, offset), value=[(host0_bid, host0_offset)...]
      b. df => flatMap
          key=(bid, offset), value=(sample_id, data)
      c. (bid, offset), (host
      """
      flatten_df = df.flatten(key_type="block_id", with_sample_id=True)
      intersect_with_offset_ids = flatten_df.join(
          flat_intersect_id, lambda vl, vr: (vl, vr)
      )
      """
      host_indexer: key=(block_id, offset), value=(sample_id, (bid, offset))
      """
      for host_id in range(len(guest_second_sign_match_ids)):
          host_indexer = intersect_with_offset_ids.mapValues(
              lambda v: (v[0][0], v[1][host_id])
          )
          ctx.hosts[host_id].put(HOST_INDEXER, host_indexer)
  
      intersect_guest_data = intersect_with_offset_ids.mapValues(
          lambda v: v[0]
      )
  
      guest_df = DataFrame.from_flatten_data(
          ctx,
          intersect_guest_data,
          df.data_manager,
          key_type="block_id",
      )
      ctx.metrics.log_metrics(
          {"intersect_count": guest_df.shape[0]},
          name="intersect_id_count",
          type="custom",
      )
  
      """
      the following just for debug, need to be delete 
      """
      # ids = [v[0] for k, v in sorted(guest_df.block_table.collect())]
      # logger.debug(f"intersect ids is: {ids}")
  
      return guest_df
  
file_path: FATE/python/fate/arch/protocol/psi/ecdh/_run.py
imports: import functools
  import logging
  from fate.arch.dataframe import DataFrame
  from fate_utils.psi import Curve25519
label_desc: Use image node_id 6 for calling a global function with example usage: guest_run(ctx, df, curve_type) and returns: guest_df