(self, ev, dst_curs)
handle one EXECUTE event
handle one EXECUTE event
handle one EXECUTE event
def handle_execute_event(self, ev, dst_curs):
        """handle one EXECUTE event"""

        if self.copy_thread:
            return

        # parse event
        fname = ev.extra1
        s_attrs = ev.extra2
        exec_attrs = ExecAttrs(urlenc = s_attrs)
        sql = ev.data

        # fixme: curs?
        pgver = dst_curs.connection.server_version
        if pgver >= 80300:
            dst_curs.execute("set local session_replication_role = 'local'")

        seq_map = {}
        q = "select seq_name, local from londiste.get_seq_list(%s) where local"
        dst_curs.execute(q, [self.queue_name])
        for row in dst_curs.fetchall():
            seq_map[row['seq_name']] = row['seq_name']

        tbl_map = {}
        for tbl, t in self.table_map.items():
            tbl_map[t.name] = t.dest_table

        q = "select * from londiste.execute_start(%s, %s, %s, false, %s)"
        res = self.exec_cmd(dst_curs, q, [self.queue_name, fname, sql, s_attrs], commit = False)
        ret = res[0]['ret_code']
        if ret > 200:
            self.log.warning("Skipping execution of '%s'", fname)
            if pgver >= 80300:
                dst_curs.execute("set local session_replication_role = 'replica'")
            return

        if exec_attrs.need_execute(dst_curs, tbl_map, seq_map):
            self.log.info("%s: executing sql")
            xsql = exec_attrs.process_sql(sql, tbl_map, seq_map)
            for stmt in skytools.parse_statements(xsql):
                dst_curs.execute(stmt)
        else:
            self.log.info("%s: execution not needed on this node")

        q = "select * from londiste.execute_finish(%s, %s)"
        self.exec_cmd(dst_curs, q, [self.queue_name, fname], commit = False)
        if pgver >= 80300:
            dst_curs.execute("set local session_replication_role = 'replica'")
def handle_execute_event ( self , ev , dst_curs ) : if self . copy_thread : return # parse event fname = ev . extra1 s_attrs = ev . extra2 exec_attrs = ExecAttrs ( urlenc = s_attrs ) sql = ev . data # fixme: curs? pgver = dst_curs . connection . server_version if pgver >= 80300 : dst_curs . execute ( "set local session_replication_role = 'local'" ) seq_map = { } q = "select seq_name, local from londiste.get_seq_list(%s) where local" dst_curs . execute ( q , [ self . queue_name ] ) for row in dst_curs . fetchall ( ) : seq_map [ row [ 'seq_name' ] ] = row [ 'seq_name' ] tbl_map = { } for tbl , t in self . table_map . items ( ) : tbl_map [ t . name ] = t . dest_table q = "select * from londiste.execute_start(%s, %s, %s, false, %s)" res = self . exec_cmd ( dst_curs , q , [ self . queue_name , fname , sql , s_attrs ] , commit = False ) ret = res [ 0 ] [ 'ret_code' ] if ret > 200 : self . log . warning ( "Skipping execution of '%s'" , fname ) if pgver >= 80300 : dst_curs . execute ( "set local session_replication_role = 'replica'" ) return if exec_attrs . need_execute ( dst_curs , tbl_map , seq_map ) : self . log . info ( "%s: executing sql" ) xsql = exec_attrs . process_sql ( sql , tbl_map , seq_map ) for stmt in skytools . parse_statements ( xsql ) : dst_curs . execute ( stmt ) else : self . log . info ( "%s: execution not needed on this node" ) q = "select * from londiste.execute_finish(%s, %s)" self . exec_cmd ( dst_curs , q , [ self . queue_name , fname ] , commit = False ) if pgver >= 80300 : dst_curs . execute ( "set local session_replication_role = 'replica'" )