skytools 2.1.12-6 / usr / share / pyshared / londiste / setup.py

#! /usr/bin/env python

"""Londiste setup and sanity checker.

"""
import sys, os, skytools
from installer import *

# support set() on 2.3
if 'set' not in __builtins__:
    from sets import Set as set

__all__ = ['ProviderSetup', 'SubscriberSetup']

def find_column_types(curs, table):
    table_oid = skytools.get_table_oid(curs, table)
    if table_oid == None:
        return None

    key_sql = """
        SELECT k.attname FROM pg_index i, pg_attribute k
         WHERE i.indrelid = %d AND k.attrelid = i.indexrelid
           AND i.indisprimary AND k.attnum > 0 AND NOT k.attisdropped
        """ % table_oid

    # find columns
    q = """
        SELECT a.attname as name,
               CASE WHEN k.attname IS NOT NULL
                    THEN 'k' ELSE 'v' END AS type
          FROM pg_attribute a LEFT JOIN (%s) k ON (k.attname = a.attname)
         WHERE a.attrelid = %d AND a.attnum > 0 AND NOT a.attisdropped
         ORDER BY a.attnum
         """ % (key_sql, table_oid)
    curs.execute(q)
    rows = curs.dictfetchall()
    return rows

def make_type_string(col_rows):
    res = map(lambda x: x['type'], col_rows)
    return "".join(res)

def convertGlobs(s):
    return s.replace('?', '.').replace('*', '.*')

def glob2regex(gpat):
    plist = [convertGlobs(s) for s in gpat.split('.')]
    return '^%s$' % '[.]'.join(plist)

class CommonSetup(skytools.DBScript):
    def __init__(self, args):
        skytools.DBScript.__init__(self, 'londiste', args)
        self.set_single_loop(1)
        self.pidfile = self.pidfile + ".setup"

        self.pgq_queue_name = self.cf.get("pgq_queue_name")
        self.consumer_id = self.cf.get("pgq_consumer_id", self.job_name)
        self.fake = self.cf.getint('fake', 0)
        self.lock_timeout = self.cf.getfloat('lock_timeout', 10)

        if len(self.args) < 3:
            self.log.error("need subcommand")
            sys.exit(1)

    def set_lock_timeout(self, curs):
        ms = int(1000 * self.lock_timeout)
        if ms > 0:
            q = "SET LOCAL statement_timeout = %d" % ms
            self.log.debug(q)
            curs.execute(q)

    def run(self):
        self.admin()

    def fetch_provider_table_list(self, curs, pattern='*'):
        q = """select table_name, trigger_name
                 from londiste.provider_get_table_list(%s)
                 where table_name ~ %s"""
        curs.execute(q, [self.pgq_queue_name, glob2regex(pattern)])
        return curs.dictfetchall()

    def get_provider_table_list(self, pattern='*'):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        list = self.fetch_provider_table_list(src_curs, pattern)
        src_db.commit()
        res = []
        for row in list:
            res.append(row['table_name'])
        return res

    def get_provider_seqs(self):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        q = """SELECT * from londiste.provider_get_seq_list(%s)"""
        src_curs.execute(q, [self.pgq_queue_name])
        src_db.commit()
        res = []
        for row in src_curs.fetchall():
            res.append(row[0])
        return res

    def get_all_seqs(self, curs):
        q = """SELECT n.nspname || '.' || c.relname
                 from pg_class c, pg_namespace n
                where n.oid = c.relnamespace 
                  and c.relkind = 'S'
                  and n.nspname not in ('pgq', 'londiste', 'pgq_node')
                  and n.nspname !~ '^pg_temp_.*'
                order by 1"""
        curs.execute(q)
        res = []
        for row in curs.fetchall():
            res.append(row[0])
        return res

    def check_provider_queue(self):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        q = "select count(1) from pgq.get_queue_info(%s)"
        src_curs.execute(q, [self.pgq_queue_name])
        ok = src_curs.fetchone()[0]
        src_db.commit()
        
        if not ok:
            self.log.error('Event queue does not exist yet')
            sys.exit(1)

    def fetch_subscriber_tables(self, curs, pattern = '*'):
        q = "select * from londiste.subscriber_get_table_list(%s) where table_name ~ %s"
        curs.execute(q, [self.pgq_queue_name, glob2regex(pattern)])
        return curs.dictfetchall()

    def get_subscriber_table_list(self, pattern = '*'):
        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        list = self.fetch_subscriber_tables(dst_curs, pattern)
        dst_db.commit()
        res = []
        for row in list:
            res.append(row['table_name'])
        return res

    def init_optparse(self, parser=None):
        p = skytools.DBScript.init_optparse(self, parser)
        p.add_option("--expect-sync", action="store_true", dest="expect_sync",
                    help = "no copy needed", default=False)
        p.add_option("--skip-truncate", action="store_true", dest="skip_truncate",
                    help = "dont delete old data", default=False)
        p.add_option("--force", action="store_true",
                    help="force", default=False)
        p.add_option("--all", action="store_true",
                    help="include all tables", default=False)
        return p


#
# Provider commands
#

class ProviderSetup(CommonSetup):

    def admin(self):
        cmd = self.args[2]
        if cmd == "tables":
            self.provider_show_tables()
        elif cmd == "add":
            self.provider_add_tables(self.args[3:])
        elif cmd == "remove":
            self.provider_remove_tables(self.args[3:])
        elif cmd == "add-seq":
            self.provider_add_seq_list(self.args[3:])
        elif cmd == "remove-seq":
            self.provider_remove_seq_list(self.args[3:])
        elif cmd == "install":
            self.provider_install()
        elif cmd == "seqs":
            self.provider_list_seqs()
        else:
            self.log.error('bad subcommand')
            sys.exit(1)

    def provider_list_seqs(self):
        list = self.get_provider_seqs()
        for seq in list:
            print seq

    def provider_get_all_seqs(self):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        list = self.get_all_seqs(src_curs)
        src_db.commit()
        return list

    def provider_add_seq_list(self, seq_list):
        if not seq_list and self.options.all:
            seq_list = self.provider_get_all_seqs()

        cur_list = self.get_provider_seqs()
        gotnew = False
        for seq in seq_list:
            seq = skytools.fq_name(seq)
            if seq in cur_list:
                self.log.info('Seq %s already subscribed' % seq)
                continue
            gotnew = True
            self.provider_add_seq(seq)

        #if gotnew:
        #    self.provider_notify_change()

    def provider_remove_seq_list(self, seq_list):
        if not seq_list and self.options.all:
            seq_list = self.get_provider_seqs()

        for seq in seq_list:
            self.provider_remove_seq(seq)
        self.provider_notify_change()

    def provider_install(self):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        install_provider(src_curs, self.log)

        # create event queue
        q = "select pgq.create_queue(%s)"
        self.exec_provider(q, [self.pgq_queue_name])

    def find_missing_provider_tables(self, pattern='*'):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        q = """select schemaname || '.' || tablename as full_name from pg_tables
                where schemaname not in ('pgq', 'londiste', 'pg_catalog', 'information_schema')
                  and schemaname !~ 'pg_.*'
                  and (schemaname || '.' || tablename) ~ %s
                except select table_name from londiste.provider_get_table_list(%s)"""
        src_curs.execute(q, [glob2regex(pattern), self.pgq_queue_name])
        rows = src_curs.fetchall()
        src_db.commit()
        list = []
        for row in rows:
            list.append(row[0])
        return list
                
    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 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_table(self, tbl):

        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        pg_vers = src_curs.connection.server_version

        q = "select londiste.provider_add_table(%s, %s)"
        self.exec_provider(q, [self.pgq_queue_name, tbl])

        # detect dangerous triggers
        if pg_vers >= 90100:
            q = """
            select tg.trigger_name
                from londiste.provider_table tbl,
                     information_schema.triggers tg
                where tbl.queue_name = %s
                  and tbl.table_name = %s
                  and tg.event_object_schema = %s
                  and tg.event_object_table = %s
                  and tg.action_timing = 'AFTER'
                  and tg.trigger_name != tbl.trigger_name
                  and tg.trigger_name < tbl.trigger_name
                  and substring(tg.trigger_name from 1 for 10) != '_londiste_'
                  and substring(tg.trigger_name from char_length(tg.trigger_name) - 6) != '_logger'
            """
        else:
            q = """
            select tg.trigger_name
                from londiste.provider_table tbl,
                     information_schema.triggers tg
                where tbl.queue_name = %s
                  and tbl.table_name = %s
                  and tg.event_object_schema = %s
                  and tg.event_object_table = %s
                  and tg.condition_timing = 'AFTER'
                  and tg.trigger_name != tbl.trigger_name
                  and tg.trigger_name < tbl.trigger_name
                  and substring(tg.trigger_name from 1 for 10) != '_londiste_'
                  and substring(tg.trigger_name from char_length(tg.trigger_name) - 6) != '_logger'
            """

        sname, tname = skytools.fq_name_parts(tbl)
        src_curs.execute(q, [self.pgq_queue_name, tbl, sname, tname])
        for r in src_curs.fetchall():
            self.log.warning("Table %s has AFTER trigger '%s' which runs before Londiste trigger.  "\
                             "If it modifies data, then events will appear in queue in wrong order." % (
                                 tbl, r[0]))
        src_db.commit()

    def provider_remove_table(self, tbl):
        q = "select londiste.provider_remove_table(%s, %s)"
        self.exec_provider(q, [self.pgq_queue_name, tbl])

    def provider_show_tables(self):
        self.check_provider_queue()
        list = self.get_provider_table_list()
        for tbl in list:
            print tbl

    def provider_notify_change(self):
        q = "select londiste.provider_notify_change(%s)"
        self.exec_provider(q, [self.pgq_queue_name])

    def provider_add_seq(self, seq):
        seq = skytools.fq_name(seq)
        q = "select londiste.provider_add_seq(%s, %s)"
        self.exec_provider(q, [self.pgq_queue_name, seq])

    def provider_remove_seq(self, seq):
        seq = skytools.fq_name(seq)
        q = "select londiste.provider_remove_seq(%s, %s)"
        self.exec_provider(q, [self.pgq_queue_name, seq])

    def exec_provider(self, sql, args):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()

        self.set_lock_timeout(src_curs)

        src_curs.execute(sql, args)

        if self.fake:
            src_db.rollback()
        else:
            src_db.commit()

#
# Subscriber commands
#

class SubscriberSetup(CommonSetup):

    def admin(self):
        cmd = self.args[2]
        if cmd == "tables":
            self.subscriber_show_tables()
        elif cmd == "missing":
            self.subscriber_missing_tables()
        elif cmd == "add":
            self.subscriber_add_tables(self.args[3:])
        elif cmd == "remove":
            self.subscriber_remove_tables(self.args[3:])
        elif cmd == "resync":
            self.subscriber_resync_tables(self.args[3:])
        elif cmd == "register":
            self.subscriber_register()
        elif cmd == "unregister":
            self.subscriber_unregister()
        elif cmd == "install":
            self.subscriber_install()
        elif cmd == "check":
            self.check_tables(self.get_provider_table_list())
        elif cmd in ["fkeys", "triggers"]:
            self.collect_meta(self.get_provider_table_list(), cmd, self.args[3:])
        elif cmd == "seqs":
            self.subscriber_list_seqs()
        elif cmd == "add-seq":
            self.subscriber_add_seq(self.args[3:])
        elif cmd == "remove-seq":
            self.subscriber_remove_seq(self.args[3:])
        elif cmd == "restore-triggers":
            self.restore_triggers(self.args[3], self.args[4:])
        else:
            self.log.error('bad subcommand: ' + cmd)
            sys.exit(1)

    def collect_meta(self, table_list, meta, args):
        """Display fkey/trigger info."""

        if args == []:
            args = ['pending', 'active']
            
        field_map = {'triggers': ['table_name', 'trigger_name', 'trigger_def'],
                     'fkeys': ['from_table', 'to_table', 'fkey_name', 'fkey_def']}
        
        query_map = {'pending': "select %s from londiste.subscriber_get_table_pending_%s(%%s)",
                     'active' : "select %s from londiste.find_table_%s(%%s)"}

        table_list = self.clean_subscriber_tables(table_list)
        if len(table_list) == 0:
            self.log.info("No tables, no fkeys")
            return

        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()

        for which in args:
            union_list = []
            fields = field_map[meta]
            q = query_map[which] % (",".join(fields), meta)
            for tbl in table_list:
                union_list.append(q % skytools.quote_literal(tbl))

            # use union as fkey may appear in duplicate
            sql = " union ".join(union_list) + " order by 1"
            desc = "%s %s" % (which, meta)
            self.display_table(desc, dst_curs, fields, sql)
        dst_db.commit()

    def display_table(self, desc, curs, fields, sql, args = []):
        """Display multirow query as a table."""

        curs.execute(sql, args)
        rows = curs.dictfetchall()
        if len(rows) == 0:
            return 0
        
        widths = [15] * len(fields)
        for row in rows:
            for i, k in enumerate(fields):
                widths[i] = widths[i] > len(row[k]) and widths[i] or len(row[k])
        widths = [w + 2 for w in widths]

        fmt = '%%-%ds' * (len(widths) - 1) + '%%s'
        fmt = fmt % tuple(widths[:-1])
        print desc
        print fmt % tuple(fields)
        print fmt % tuple(['-'*15] * len(fields))
            
        for row in rows:
            print fmt % tuple([row[k] for k in fields])
        print '\n'
        return 1

    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 check_tables(self, table_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()

        failed = 0
        for tbl in table_list:
            self.log.info('Checking %s' % tbl)
            if not skytools.exists_table(src_curs, tbl):
                self.log.error('Table %s missing from provider side' % tbl)
                failed += 1
            elif not skytools.exists_table(dst_curs, tbl):
                self.log.error('Table %s missing from subscriber side' % tbl)
                failed += 1
            else:
                failed += self.check_table_columns(src_curs, dst_curs, tbl)

        src_db.commit()
        dst_db.commit()

        return failed

    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 check_table_columns(self, src_curs, dst_curs, tbl):
        src_colrows = find_column_types(src_curs, tbl)
        dst_colrows = find_column_types(dst_curs, tbl)

        src_cols = make_type_string(src_colrows)
        dst_cols = make_type_string(dst_colrows)
        if src_cols.find('k') < 0:
            self.log.error('provider table %s has no primary key (%s)' % (
                             tbl, src_cols))
            return 1
        if dst_cols.find('k') < 0:
            self.log.error('subscriber table %s has no primary key (%s)' % (
                             tbl, dst_cols))
            return 1

        if src_cols != dst_cols:
            self.log.warning('table %s structure is not same (%s/%s)'\
                 ', trying to continue' % (tbl, src_cols, dst_cols))

        err = 0
        for row in src_colrows:
            found = 0
            for row2 in dst_colrows:
                if row2['name'] == row['name']:
                    found = 1
                    break
            if not found:
                err = 1
                self.log.error('%s: column %s on provider not on subscriber'
                                    % (tbl, row['name']))
            elif row['type'] != row2['type']:
                err = 1
                self.log.error('%s: pk different on column %s'
                                    % (tbl, row['name']))

        return err

    def subscriber_install(self):
        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()

        install_subscriber(dst_curs, self.log)

        if self.fake:
            self.log.debug('rollback')
            dst_db.rollback()
        else:
            self.log.debug('commit')
            dst_db.commit()

    def subscriber_register(self):
        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        src_curs.execute("select pgq.register_consumer(%s, %s)",
            [self.pgq_queue_name, self.consumer_id])
        src_db.commit()

    def subscriber_unregister(self):
        q = "select londiste.subscriber_set_table_state(%s, %s, NULL, NULL)"

        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        tbl_rows = self.fetch_subscriber_tables(dst_curs)
        for row in tbl_rows:
            dst_curs.execute(q, [self.pgq_queue_name, row['table_name']])
        dst_db.commit()

        src_db = self.get_database('provider_db')
        src_curs = src_db.cursor()
        src_curs.execute("select pgq.unregister_consumer(%s, %s)",
            [self.pgq_queue_name, self.consumer_id])
        src_db.commit()

    def subscriber_show_tables(self):
        """print out subscriber table list, with state and snapshot"""
        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        list = self.fetch_subscriber_tables(dst_curs)
        dst_db.commit()

        format = "%-30s   %20s"
        print format % ("Table", "State")
        for tbl in list:
            print format % (tbl['table_name'],
                            tbl['merge_state'] or '-')

    def subscriber_missing_tables(self):
        provider_tables = self.get_provider_table_list()
        subscriber_tables = self.get_subscriber_table_list()
        for tbl in provider_tables:
            if tbl not in subscriber_tables:
                print "Table: %s" % tbl
        provider_seqs = self.get_provider_seqs()
        subscriber_seqs = self.get_subscriber_seq_list()
        for seq in provider_seqs:
            if seq not in subscriber_seqs:
                print "Sequence: %s" % seq

    def find_missing_subscriber_tables(self, pattern='*'):
        src_db = self.get_database('subscriber_db')
        src_curs = src_db.cursor()
        q = """select schemaname || '.' || tablename as full_name from pg_tables
                where schemaname not in ('pgq', 'londiste', 'pg_catalog', 'information_schema')
                  and schemaname !~ 'pg_.*'
                  and schemaname || '.' || tablename ~ %s
                except select table_name from londiste.provider_get_table_list(%s)"""
        src_curs.execute(q, [glob2regex(pattern), self.pgq_queue_name])
        rows = src_curs.fetchall()
        src_db.commit()
        list = []
        for row in rows:
            list.append(row[0])
        return list

    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 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_resync_tables(self, table_list):
        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        list = self.fetch_subscriber_tables(dst_curs)

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

        for tbl in table_list:
            tbl = skytools.fq_name(tbl)
            tbl_row = None
            for row in list:
                if row['table_name'] == tbl:
                    tbl_row = row
                    break
            if not tbl_row:
                self.log.warning("Table %s not found" % tbl)
            elif tbl_row['merge_state'] != 'ok':
                self.log.warning("Table %s is not in stable state" % tbl)
            else:
                self.log.info("Resyncing %s" % tbl)
                q = "select londiste.subscriber_set_table_state(%s, %s, NULL, NULL)"
                dst_curs.execute(q, [self.pgq_queue_name, tbl])
        dst_db.commit()

    def subscriber_add_one_table(self, dst_curs, tbl):
        q_add = "select londiste.subscriber_add_table(%s, %s)"
        q_triggers = "select londiste.subscriber_drop_all_table_triggers(%s)"

        if self.options.expect_sync and self.options.skip_truncate:
            self.log.error("Too many options: --expect-sync and --skip-truncate")
            sys.exit(1)

        dst_curs.execute(q_add, [self.pgq_queue_name, tbl])
        if self.options.expect_sync:
            q = "select londiste.subscriber_set_table_state(%s, %s, null, 'ok')"
            dst_curs.execute(q, [self.pgq_queue_name, tbl])
            return

        dst_curs.execute(q_triggers, [tbl])
        if self.options.skip_truncate:
            q = "select londiste.subscriber_set_skip_truncate(%s, %s, true)"
            dst_curs.execute(q, [self.pgq_queue_name, tbl])

    def subscriber_remove_one_table(self, tbl):
        q_remove = "select londiste.subscriber_remove_table(%s, %s)"
        q_triggers = "select londiste.subscriber_restore_all_table_triggers(%s)"

        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        dst_curs.execute(q_remove, [self.pgq_queue_name, tbl])
        dst_curs.execute(q_triggers, [tbl])
        dst_db.commit()

    def get_subscriber_seq_list(self):
        dst_db = self.get_database('subscriber_db')
        dst_curs = dst_db.cursor()
        q = "SELECT * from londiste.subscriber_get_seq_list(%s)"
        dst_curs.execute(q, [self.pgq_queue_name])
        list = dst_curs.fetchall()
        dst_db.commit()
        res = []
        for row in list:
            res.append(row[0])
        return res

    def subscriber_list_seqs(self):
        list = self.get_subscriber_seq_list()
        for seq in list:
            print seq

    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 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()