code: def _flat_block_with_possible_duplicate_keys(
      block_table, duplicate_allow=False
  ):
      """
      row_value: (encrypt_id, [(block_id, _offset)])
      """
  
      def _mapper(kvs):
          for block_id, id_list in kvs:
              for _i, _id in enumerate(id_list):
                  yield _id, [(block_id, _i)]
  
      def _reducer(v1, v2):
          if not v1:
              return v2
          if not v2:
              return v1
  
          if not duplicate_allow:
              raise ValueError("duplicate match_id detect")
  
          return v1 + v2
  
      return block_table.mapReducePartitions(_mapper, _reducer)
  
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 3 for calling a global function with example usage: _flat_block_with_possible_duplicate_keys(block_table, duplicate_allow) and returns: block_table, unknown, v2, v1