(self, src_db, dst_db, tick_id)
Worker-specific cleanup on target node.
Worker-specific cleanup on target node.
Worker - specific cleanup on target node .
def finish_remote_batch(self, src_db, dst_db, tick_id):
        """Worker-specific cleanup on target node.
        """

        # merge-leaf on branch should not update tick pos
        st = self._worker_state
        if st.wait_behind:
            dst_db.commit()

            # still need to publish wm info
            if st.local_wm_publish and self.main_worker:
                self.publish_local_wm(src_db, dst_db)

            return

        if self.main_worker:
            dst_curs = dst_db.cursor()

            self.flush_events(dst_curs)

            # send tick event into queue
            if st.send_tick_event:
                q = "select pgq.insert_event(%s, 'pgq.tick-id', %s, %s, null, null, null)"
                dst_curs.execute(q, [st.target_queue, str(tick_id), self.pgq_queue_name])

        CascadedConsumer.finish_remote_batch(self, src_db, dst_db, tick_id)

        if self.main_worker:
            if st.create_tick:
                # create actual tick
                tick_id = self.batch_info['tick_id']
                tick_time = self.batch_info['batch_end']
                self.create_branch_tick(dst_db, tick_id, tick_time)
            if st.local_wm_publish:
                self.publish_local_wm(src_db, dst_db)
def finish_remote_batch ( self , src_db , dst_db , tick_id ) : # merge-leaf on branch should not update tick pos st = self . _worker_state if st . wait_behind : dst_db . commit ( ) # still need to publish wm info if st . local_wm_publish and self . main_worker : self . publish_local_wm ( src_db , dst_db ) return if self . main_worker : dst_curs = dst_db . cursor ( ) self . flush_events ( dst_curs ) # send tick event into queue if st . send_tick_event : q = "select pgq.insert_event(%s, 'pgq.tick-id', %s, %s, null, null, null)" dst_curs . execute ( q , [ st . target_queue , str ( tick_id ) , self . pgq_queue_name ] ) CascadedConsumer . finish_remote_batch ( self , src_db , dst_db , tick_id ) if self . main_worker : if st . create_tick : # create actual tick tick_id = self . batch_info [ 'tick_id' ] tick_time = self . batch_info [ 'batch_end' ] self . create_branch_tick ( dst_db , tick_id , tick_time ) if st . local_wm_publish : self . publish_local_wm ( src_db , dst_db )