def work(self):
        """Syncer main function."""

        # 'SELECT 1' and COPY must use same snapshot, so change isolation level.
        dst_db = self.get_database('db', isolation_level = skytools.I_REPEATABLE_READ)
        pnode, ploc = self.get_provider_location(dst_db)

        dst_tables, names = self.get_tables(dst_db)

        if len(self.args) > 2:
            tlist = self.args[2:]
        else:
            tlist = names

        for tbl in tlist:
            tbl = skytools.fq_name(tbl)
            if not tbl in dst_tables:
                self.log.warning('Table not subscribed: %s' % tbl)
                continue
            t2 = dst_tables[tbl]
            if t2.merge_state != 'ok':
                self.log.warning('Table %s not synced yet, no point' % tbl)
                continue

            pnode, ploc, wname = find_copy_source(self, self.queue_name, tbl, pnode, ploc)
            self.log.info('%s: Using node %s as provider', tbl, pnode)

            if wname is None:
                wname = self.consumer_name
            self.downstream_worker_name = wname

            self.process_one_table(tbl, t2, dst_db, pnode, ploc)

        # signal caller about bad tables
        sys.exit(self.bad_tables)

    def subscriber_add_seq(self, seq_list):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        
        prov_list = self.get_provider_seqs()

        full_list = self.get_all_seqs(dst_curs)
        cur_list = self.get_subscriber_seq_list()

        if not seq_list and self.options.all:
            seq_list = prov_list
        
        for seq in seq_list:
            seq = skytools.fq_name(seq)
            if seq not in prov_list:
                self.log.error('Seq %s does not exist on provider side' % seq)
                continue
            if seq not in full_list:
                self.log.error('Seq %s does not exist on subscriber side' % seq)
                continue
            if seq in cur_list:
                self.log.info('Seq %s already subscribed' % seq)
                continue

            self.log.info('Adding sequence: %s' % seq)
            q = "select londiste.subscriber_add_seq(%s, %s)"
            dst_curs.execute(q, [self.pgq_queue_name, seq])

        dst_db.commit()

    def cmd_add_seq(self, *args):
        """Attach seqs(s) to local node."""
        dst_db = self.get_database('db')
        dst_curs = dst_db.cursor()
        src_db = self.get_provider_db()
        src_curs = src_db.cursor()

        src_seqs = self.fetch_seqs(src_curs)
        dst_seqs = self.fetch_seqs(dst_curs)
        src_db.commit()
        self.sync_seq_list(dst_curs, src_seqs, dst_seqs)
        dst_db.commit()

        args = self.expand_arg_list(dst_db, 'S', False, args)

        # pick proper create flags
        create = self.options.create_only
        if not create and self.options.create:
            create = 'full'

        fmap = {
            "full": skytools.T_SEQUENCE,
        }
        create_flags = 0
        if create:
            for f in create.split(','):
                if f not in fmap:
                    raise Exception("bad --create-only flag: " + f)
                create_flags += fmap[f]

        # seems ok
        for seq in args:
            seq = skytools.fq_name(seq)
            self.add_seq(src_db, dst_db, seq, create_flags)
        dst_db.commit()

    def solve_globbing(self, args, full_list, full_map, reverse_map):
        def glob2regex(s):
            s = s.replace('.', '[.]').replace('?', '.').replace('*', '.*')
            return '^%s$' % s

        res_map = {}
        res_list = []
        err = 0
        for a in args:
            if a.find('*') >= 0 or a.find('?') >= 0:
                if a.find('.') < 0:
                    a = 'public.' + a
                rc = re.compile(glob2regex(a))
                for x in full_list:
                    if rc.match(x):
                        if not x in res_map:
                            res_map[x] = 1
                            res_list.append(x)
            else:
                a = skytools.fq_name(a)
                if a in res_map:
                    continue
                elif a in full_map:
                    res_list.append(a)
                    res_map[a] = 1
                elif a in reverse_map:
                    self.log.info("%s already processed" % a)
                else:
                    self.log.warning("%s not available" % a)
                    err = 1
        if err:
            raise UsageError("Cannot proceed")
        return res_list

 def need_execute(self, curs, local_tables, local_seqs):
     # if no attrs, always execute
     if not self.attrs:
         return True
     
     matched = 0
     missed = 0
     good_list = []
     miss_list = []
     for m in META_MATCHERS:
         k = m.get_key()
         if k not in self.attrs:
             continue
         for v in self.attrs[k]:
             fqname = skytools.fq_name(v)
             if m.match(fqname, curs, local_tables, local_seqs):
                 matched += 1
                 good_list.append(v)
             else:
                 missed += 1
                 miss_list.append(v)
                 # should be drop out early?
     if matched > 0 and missed == 0:
         return True
     elif missed > 0 and matched == 0:
         return False
     elif missed == 0 and matched == 0:
         # should not happen, but lets restore old behaviour?
         return True
     else:
         raise Exception("SQL only partially matches local setup: matches=%r misses=%r" % (good_list, miss_list))

    def cmd_add_table(self, *args):
        """Attach table(s) to local node."""

        self.load_local_info()

        src_db = self.get_provider_db()
        if not self.is_root():
            src_curs = src_db.cursor()
            src_tbls = self.fetch_set_tables(src_curs)
            src_db.commit()

        dst_db = self.get_database('db')
        dst_curs = dst_db.cursor()
        dst_tbls = self.fetch_set_tables(dst_curs)
        if self.is_root():
            src_tbls = dst_tbls
        else:
            self.sync_table_list(dst_curs, src_tbls, dst_tbls)
        dst_db.commit()

        needs_tbl = self.handler_needs_table()
        args = self.expand_arg_list(dst_db, 'r', False, args, needs_tbl)

        # dont check for exist/not here (root handling)
        if not self.is_root() and not self.options.expect_sync and not self.options.find_copy_node:
            problems = False
            for tbl in args:
                tbl = skytools.fq_name(tbl)
                if (tbl in src_tbls) and not src_tbls[tbl]['local']:
                    self.log.error("Table %s does not exist on provider, need to switch to different provider" % tbl)
                    problems = True
            if problems:
                self.log.error("Problems, canceling operation")
                sys.exit(1)

        # pick proper create flags
        if self.options.create_full:
            create_flags = skytools.T_ALL
        elif self.options.create:
            create_flags = skytools.T_TABLE | skytools.T_PKEY
        else:
            create_flags = 0

        # sanity check
        if self.options.dest_table and len(args) > 1:
            self.log.error("--dest-table can be given only for single table")
            sys.exit(1)

        # not implemented
        if self.options.find_copy_node and create_flags != 0:
            self.log.error("--find-copy-node does not work with --create")
            sys.exit(1)

        # seems ok
        for tbl in args:
            self.add_table(src_db, dst_db, tbl, create_flags, src_tbls)

        # wait
        if self.options.wait_sync:
            self.wait_for_sync(dst_db)

 def restore_triggers(self, tbl, triggers=None):
     tbl = skytools.fq_name(tbl)
     if tbl not in self.get_subscriber_table_list():
         self.log.error("Table %s is not in the subscriber queue." % tbl)
         sys.exit(1)
         
     dst_db = self.get_database('subscriber_db')
     dst_curs = dst_db.cursor()
     
     if not triggers:
         q = "select count(1) from londiste.subscriber_get_table_pending_triggers(%s)"
         dst_curs.execute(q, [tbl])
         if not dst_curs.fetchone()[0]:
             self.log.info("No pending triggers found for %s." % tbl)
         else:
             q = "select londiste.subscriber_restore_all_table_triggers(%s)"
             dst_curs.execute(q, [tbl])
     else:
         for trigger in triggers:
             q = "select count(1) from londiste.find_table_triggers(%s) where trigger_name=%s"
             dst_curs.execute(q, [tbl, trigger])
             if dst_curs.fetchone()[0]:
                 self.log.info("Trigger %s on %s is already active." % (trigger, tbl))
                 continue
                 
             q = "select count(1) from londiste.subscriber_get_table_pending_triggers(%s) where trigger_name=%s"
             dst_curs.execute(q, [tbl, trigger])
             if not dst_curs.fetchone()[0]:
                 self.log.info("Trigger %s not found on %s" % (trigger, tbl))
                 continue
                 
             q = "select londiste.subscriber_restore_table_trigger(%s, %s)"
             dst_curs.execute(q, [tbl, trigger])
     dst_db.commit()

    def cmd_add_seq(self, *args):
        """Attach seqs(s) to local node."""
        dst_db = self.get_database('db')
        dst_curs = dst_db.cursor()
        src_db = self.get_provider_db()
        src_curs = src_db.cursor()

        src_seqs = self.fetch_seqs(src_curs)
        dst_seqs = self.fetch_seqs(dst_curs)
        src_db.commit()
        self.sync_seq_list(dst_curs, src_seqs, dst_seqs)
        dst_db.commit()

        args = self.expand_arg_list(dst_db, 'S', False, args)

        # pick proper create flags
        if self.options.create_full:
            create_flags = skytools.T_SEQUENCE
        elif self.options.create:
            create_flags = skytools.T_SEQUENCE
        else:
            create_flags = 0

        # seems ok
        for seq in args:
            seq = skytools.fq_name(seq)
            self.add_seq(src_db, dst_db, seq, create_flags)
        dst_db.commit()

    def cmd_add_table(self, *args):
        """Attach table(s) to local node."""

        dst_db = self.get_database('db')
        dst_curs = dst_db.cursor()
        src_db = self.get_provider_db()
        src_curs = src_db.cursor()

        src_tbls = self.fetch_set_tables(src_curs)
        dst_tbls = self.fetch_set_tables(dst_curs)
        src_db.commit()
        self.sync_table_list(dst_curs, src_tbls, dst_tbls)
        dst_db.commit()

        # dont check for exist/not here (root handling)
        problems = False
        for tbl in args:
            tbl = skytools.fq_name(tbl)
            if (tbl in src_tbls) and not src_tbls[tbl]:
                self.log.error("Table %s does not exist on provider, need to switch to different provider" % tbl)
                problems = True
        if problems:
            self.log.error("Problems, canceling operation")
            sys.exit(1)

        # pick proper create flags
        create = self.options.create_only
        if not create and self.options.create:
            create = 'full'

        fmap = {
            "full": skytools.T_ALL,
            "pkey": skytools.T_PKEY,
        }
        create_flags = 0
        if create:
            for f in create.split(','):
                if f not in fmap:
                    raise Exception("bad --create-only flag: " + f)
            create_flags += fmap[f]

        # seems ok
        for tbl in args:
            tbl = skytools.fq_name(tbl)
            self.add_table(src_db, dst_db, tbl, create_flags)

    def work(self):
        """Syncer main function."""

        # 'SELECT 1' and COPY must use same snapshot, so change isolation level.
        dst_db = self.get_database('db', isolation_level = skytools.I_REPEATABLE_READ)
        provider_loc = self.get_provider_location(dst_db)

        lock_db = self.get_database('lock_db', connstr = provider_loc)
        setup_db = self.get_database('setup_db', autocommit = 1, connstr = provider_loc)

        src_db = self.get_database('provider_db', connstr = provider_loc,
                                   isolation_level = skytools.I_REPEATABLE_READ)

        setup_curs = setup_db.cursor()

        # provider node info
        self.provider_node = self.get_provider_info(setup_curs)

        src_tables, ignore = self.get_tables(src_db)
        dst_tables, names = self.get_tables(dst_db)

        if len(self.args) > 2:
            tlist = self.args[2:]
        else:
            tlist = names

        for tbl in tlist:
            tbl = skytools.fq_name(tbl)
            if not tbl in dst_tables:
                self.log.warning('Table not subscribed: %s' % tbl)
                continue
            if not tbl in src_tables:
                self.log.warning('Table not available on provider: %s' % tbl)
                continue
            t1 = src_tables[tbl]
            t2 = dst_tables[tbl]

            if t1.merge_state != 'ok':
                self.log.warning('Table %s not ready yet on provider' % tbl)
                continue
            if t2.merge_state != 'ok':
                self.log.warning('Table %s not synced yet, no point' % tbl)
                continue

            self.check_consumer(setup_db)

            self.check_table(t1, t2, lock_db, src_db, dst_db, setup_db)
            lock_db.commit()
            src_db.commit()
            dst_db.commit()

        # signal caller about bad tables
        sys.exit(self.bad_tables)

 def subscriber_remove_tables(self, table_list):
     subscriber_tables = self.get_subscriber_table_list()
     if not table_list and self.options.all:
         table_list = ['*.*']
         
     for tbl in table_list:
         tbls = self.get_subscriber_table_list(skytools.fq_name(tbl))
         for tbl in tbls:
             if tbl in subscriber_tables:
                 self.log.info("Removing: %s" % tbl)
                 self.subscriber_remove_one_table(tbl)
             else:
                 self.log.info("Table %s already removed" % tbl)

    def subscriber_add_tables(self, table_list):
        provider_tables = self.get_provider_table_list()
        subscriber_tables = self.get_subscriber_table_list()

        if not table_list and self.options.all:
            table_list = ['*.*']
            for tbl in provider_tables:
                if tbl not in subscriber_tables:
                    table_list.append(tbl)
        
        tbls = []
        for tbl in table_list:
            more = self.find_missing_subscriber_tables(skytools.fq_name(tbl))
            if more == []:
                self.log.info("No tables found that match %s" % tbl)
            tbls.extend(more)
        tbls = list(set(tbls))

        err = 0
        table_list = []
        for tbl in tbls:
            if tbl not in provider_tables:
                err = 1
                self.log.error("Table %s not attached to queue" % tbl)
                if not self.options.force:
                    continue
            table_list.append(tbl)
                
        if err:
            if self.options.force:
                self.log.warning('--force used, ignoring errors')

        err = self.check_tables(table_list)
        if err:
            if self.options.force:
                self.log.warning('--force used, ignoring errors')
            else:
                sys.exit(1)

        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        for tbl in table_list:
            if tbl in subscriber_tables:
                self.log.info("Table %s already added" % tbl)
            else:
                self.log.info("Adding %s" % tbl)
                self.subscriber_add_one_table(dst_curs, tbl)
            dst_db.commit()

    def provider_remove_tables(self, table_list):
        self.check_provider_queue()

        cur_list = self.get_provider_table_list()
        if not table_list and self.options.all:
            table_list = cur_list

        for tbl in table_list:
            tbls = self.get_provider_table_list(skytools.fq_name(tbl))
            for tbl in tbls:
                if tbl not in cur_list:
                    self.log.info('%s already removed' % tbl)
                else:
                    self.log.info("Removing %s" % tbl)
                    self.provider_remove_table(tbl)
        self.provider_notify_change()

    def provider_add_tables(self, table_list):
        self.check_provider_queue()

        if self.options.all and not table_list:
            table_list = ['*.*']

        cur_list = self.get_provider_table_list()
        for tbl in table_list:
            tbls = self.find_missing_provider_tables(skytools.fq_name(tbl))
            
            for tbl in tbls:
                if tbl not in cur_list:
                    self.log.info('Adding %s' % tbl)
                    self.provider_add_table(tbl)
                else:
                    self.log.info("Table %s already added" % tbl)
        self.provider_notify_change()

    def subscriber_remove_seq(self, seq_list):
        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        cur_list = self.get_subscriber_seq_list()

        if not seq_list and self.options.all:
            seq_list = cur_list

        for seq in seq_list:
            seq = skytools.fq_name(seq)
            if seq not in cur_list:
                self.log.warning('Seq %s not subscribed')
            else:
                self.log.info('Removing sequence: %s' % seq)
                q = "select londiste.subscriber_remove_seq(%s, %s)"
                dst_curs.execute(q, [self.pgq_queue_name, seq])
        dst_db.commit()

 def clean_subscriber_tables(self, table_list):
     """Returns fully-quelifies table list of tables
     that are registered on subscriber.
     """
     subscriber_tables = self.get_subscriber_table_list()
     if not table_list and self.options.all:
         table_list = subscriber_tables
     else:
         newlist = []
         for tbl in table_list:
             tbl = skytools.fq_name(tbl)
             if tbl in subscriber_tables:
                 newlist.append(tbl)
             else:
                 #self.log.warning("table %s not subscribed" % tbl)
                 pass
         table_list = newlist
     return table_list

    def cmd_change_handler(self, tbl):
        """Change handler (table_attrs) of the replicated table."""

        self.load_local_info()

        tbl = skytools.fq_name(tbl)

        db = self.get_database('db')
        curs = db.cursor()
        q = "select table_attrs, dest_table "\
            " from londiste.get_table_list(%s) "\
            " where table_name = %s and local"
        curs.execute(q, [self.set_name, tbl])
        if curs.rowcount == 0:
            self.log.error("Table %s not found on this node", tbl)
            sys.exit(1)

        attrs, dest_table = curs.fetchone()
        attrs = skytools.db_urldecode(attrs or '')
        old_handler = attrs.get('handler')

        tgargs = self.build_tgargs()
        if self.options.handler:
            new_handler = self.build_handler(tbl, tgargs, dest_table)
        else:
            new_handler = None

        if old_handler == new_handler:
            self.log.info("Handler is already set to desired value, nothing done")
            sys.exit(0)

        if new_handler:
            attrs['handler'] = new_handler
        elif 'handler' in attrs:
            del attrs['handler']

        args = [self.set_name, tbl, tgargs, None]
        if attrs:
            args[3] = skytools.db_urlencode(attrs)

        q = "select * from londiste.local_change_handler(%s, %s, %s, %s)"
        self.exec_cmd(curs, q, args)
        db.commit()

 def process_sql(self, sql, local_tables, local_seqs):
     """Replace replacement tags in sql with actual local names."""
     for k, vlist in self.attrs.items():
         m = META_KEYS[k]
         if not m.local_rename():
             continue
         for v in vlist:
             repname = '@%[email protected]' % v
             fqname = skytools.fq_name(v)
             if fqname in local_tables:
                 localname = local_tables[fqname]
             elif fqname in local_seqs:
                 localname = local_seqs[fqname]
             else:
                 # should not happen
                 raise Exception("bug: lost table: "+v)
             qdest = skytools.quote_fqident(localname)
             sql = sql.replace(repname, qdest)
     return sql

    def reload(self):
        skytools.DBScript.reload(self)

        self.pgq_lazy_fetch = self.cf.getint("pgq_lazy_fetch", self.default_lazy_fetch)

        # set following ones to None if not set
        self.pgq_min_count = self.cf.getint("pgq_batch_collect_events", 0) or None
        self.pgq_min_interval = self.cf.get("pgq_batch_collect_interval", '') or None
        self.pgq_min_lag = self.cf.get("pgq_keep_lag", '') or None

        # filter out specific tables only
        tfilt = []
        for t in self.cf.getlist('table_filter', ''):
            tfilt.append(skytools.quote_literal(skytools.fq_name(t)))
        if len(tfilt) > 0:
            expr = "ev_extra1 in (%s)" % ','.join(tfilt)
            self.consumer_filter = expr

        self.keepalive_stats = self.cf.getint("keepalive_stats", 300)

    def cmd_resync(self, *args):
        """Reload data from provider node."""
        db = self.get_database('db')
        args = self.expand_arg_list(db, 'r', True, args)

        if not self.options.find_copy_node:
            self.load_local_info()
            src_db = self.get_provider_db()
            src_curs = src_db.cursor()
            src_tbls = self.fetch_set_tables(src_curs)
            src_db.commit()

            problems = 0
            for tbl in args:
                tbl = skytools.fq_name(tbl)
                if tbl not in src_tbls or not src_tbls[tbl]['local']:
                    self.log.error("Table %s does not exist on provider, need to switch to different provider", tbl)
                    problems += 1
            if problems > 0:
                self.log.error("Problems, cancelling operation")
                sys.exit(1)

        if self.options.find_copy_node or self.options.copy_node:
            q = "select table_name, table_attrs from londiste.get_table_list(%s) where local"
            cur = db.cursor()
            cur.execute(q, [self.set_name])
            for row in cur.fetchall():
                if row['table_name'] not in args:
                    continue
                attrs = skytools.db_urldecode (row['table_attrs'] or '')

                if self.options.find_copy_node:
                    attrs['copy_node'] = '?'
                elif self.options.copy_node:
                    attrs['copy_node'] = self.options.copy_node

                attrs = skytools.db_urlencode (attrs)
                q = "select * from londiste.local_set_table_attrs (%s, %s, %s)"
                self.exec_cmd(db, q, [self.set_name, row['table_name'], attrs])

        q = "select * from londiste.local_set_table_state(%s, %s, null, null)"
        self.exec_cmd_many(db, q, [self.set_name], args)