def join(self, follower_window, leader_window, delay):
        """
        Assume leader stream is exponentially larger than follower stream,
        we cache the all the events from now back to watermark. example id
        maybe duplicated in follower stream
        Return:
            show_matches: an array with tuples(follower, leader), in same order
                with leader event
        """
        leader_matches = []
        leader_mismatches = {}
        # keys example: [(req_id, cid), click_id]
        keys = self._expr.keys()
        follower_dict = follower_window.as_dict(keys)
        idx = 0
        while idx < leader_window.size():
            elem = leader_window[idx]
            #1. find the first matching key
            key_idx = []
            leader_window.key_map_fn(elem)
            leader = elem.item
            key_str_arr = make_index_by_attr(
                keys, leader, key_idx)
            found = False
            for ki, k in enumerate(key_str_arr):
                if k not in follower_dict:
                    continue
                cd = follower_dict[k]
                key_idx = key_idx[ki]
                for i in reversed(range(len(cd))):
                    #A leader can match multiple conversion event, add
                    # all the matched conversion-leader pair to result
                    follower = follower_window[cd[i]].item
                    #2. select all the matching items from the specific key
                    # in follower side.
                    if self._expr.run_func(key_idx)(leader, follower):
                        ## delay range check
                        time_df = fcc.time_diff(leader.event_time,
                                                follower.event_time)
                        if abs(time_df) <= delay:
                            leader_matches.append((cd[i], idx))
                found = True
                break
            if not found:
                leader_mismatches[leader_window[idx].index] = leader
            idx += 1
        return leader_matches, leader_mismatches

        Assume leader stream is exponentially larger than follower stream,
        we cache the all the events from now back to watermark. example id
        maybe duplicated in follower stream
        Return:
            show_matches: an array with tuples(follower, leader), in same order
                with leader event
        
fedlearner/data_join/joiner_impl/universal_joiner.py