def add_table(self, src_db, dst_db, tbl, create_flags):
        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()
        tbl_exists = skytools.exists_table(dst_curs, tbl)
        if create_flags:
            if tbl_exists:
                self.log.info('Table %s already exist, not touching' % tbl)
            else:
                if not skytools.exists_table(src_curs, tbl):
                    # table not present on provider - nowhere to get the DDL from
                    self.log.warning('Table "%s" missing on provider, skipping' % tbl)
                    return
                schema = skytools.fq_name_parts(tbl)[0]
                if not skytools.exists_schema(dst_curs, schema):
                    q = "create schema %s" % skytools.quote_ident(schema)
                    dst_curs.execute(q)
                s = skytools.TableStruct(src_curs, tbl)
                src_db.commit()
                s.create(dst_curs, create_flags, log = self.log)
        elif not tbl_exists:
            self.log.warning('Table "%s" missing on subscriber, use --create if necessary' % tbl)
            return

        # actual table registration
        q = "select * from londiste.local_add_table(%s, %s)"
        self.exec_cmd(dst_curs, q, [self.set_name, tbl])
        if self.options.expect_sync:
            q = "select * from londiste.local_set_table_state(%s, %s, NULL, 'ok')"
            self.exec_cmd(dst_curs, q, [self.set_name, tbl])
        if self.options.copy_condition:
            q = "select * from londiste.local_set_table_attrs(%s, %s, %s)"
            attrs = {'copy_condition': self.options.copy_condition}
            enc_attrs = skytools.db_urlencode(attrs)
            self.exec_cmd(dst_curs, q, [self.set_name, tbl, enc_attrs])
        dst_db.commit()

    def add_table(self, src_db, dst_db, tbl, create_flags):
        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()
        tbl_exists = skytools.exists_table(dst_curs, tbl)
        if create_flags:
            if tbl_exists:
                self.log.info('Table %s already exist, not touching' % tbl)
            else:
                if not skytools.exists_table(src_curs, tbl):
                    # table not present on provider - nowhere to get the DDL from
                    self.log.warning('Table "%s" missing on provider, skipping' % tbl)
                    return
                schema = skytools.fq_name_parts(tbl)[0]
                if not skytools.exists_schema(dst_curs, schema):
                    q = "create schema %s" % skytools.quote_ident(schema)
                    dst_curs.execute(q)
                s = skytools.TableStruct(src_curs, tbl)
                src_db.commit()
                s.create(dst_curs, create_flags, log = self.log)
        elif not tbl_exists:
            self.log.warning('Table "%s" missing on subscriber, use --create if necessary' % tbl)
            return

        q = "select * from londiste.local_add_table(%s, %s)"
        self.exec_cmd(dst_curs, q, [self.set_name, tbl])
        dst_db.commit()

    def add_table(self, src_db, dst_db, tbl, create_flags):
        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()
        tbl_exists = skytools.exists_table(dst_curs, tbl)
        if create_flags:
            if tbl_exists:
                self.log.info('Table %s already exist, not touching' % tbl)
            else:
                if not skytools.exists_table(src_curs, tbl):
                    # table not present on provider - nowhere to get the DDL from
                    self.log.warning('Table "%s" missing on provider, skipping' % tbl)
                    return
                schema = skytools.fq_name_parts(tbl)[0]
                if not skytools.exists_schema(dst_curs, schema):
                    q = "create schema %s" % skytools.quote_ident(schema)
                    dst_curs.execute(q)
                s = skytools.TableStruct(src_curs, tbl)
                src_db.commit()
                s.create(dst_curs, create_flags, log = self.log)
        elif not tbl_exists:
            self.log.warning('Table "%s" missing on subscriber, use --create if necessary' % tbl)
            return

        tgargs = []
        if self.options.trigger_arg:
            tgargs = self.options.trigger_arg
        tgflags = self.options.trigger_flags
        if tgflags:
            tgargs.append('tgflags='+tgflags)
        if self.options.no_triggers:
            tgargs.append('no_triggers')
        if self.options.merge_all:
            tgargs.append('merge_all')

        attrs = {}
        if self.options.handler:
            hstr = londiste.handler.create_handler_string(
                            self.options.handler, self.options.handler_arg)
            p = londiste.handler.build_handler(tbl, hstr, self.log)
            attrs['handler'] = hstr
            p.add(tgargs)

        # actual table registration
        q = "select * from londiste.local_add_table(%s, %s, %s)"
        self.exec_cmd(dst_curs, q, [self.set_name, tbl, tgargs])

        if self.options.expect_sync:
            q = "select * from londiste.local_set_table_state(%s, %s, NULL, 'ok')"
            self.exec_cmd(dst_curs, q, [self.set_name, tbl])
        else:
            if self.options.skip_truncate:
                attrs['skip_truncate'] = 1
            if self.options.copy_condition:
                attrs['copy_condition'] = self.options.copy_condition
        if attrs:
            enc_attrs = skytools.db_urlencode(attrs)
            q = "select * from londiste.local_set_table_attrs(%s, %s, %s)"
            self.exec_cmd(dst_curs, q, [self.set_name, tbl, enc_attrs])
        dst_db.commit()

    def check_tables(self, dcon, tables):
        """Checks that tables needed for copy are there. If not
        then creates them.

        Used by other procedures to ensure that table is there
        before they start inserting.

        The commits should not be dangerous, as we haven't done anything
        with cdr's yet, so they should still be in one TX.

        Although it would be nicer to have a lock for table creation.
        """

        dcur = dcon.cursor()
        for tbl in tables.keys():
            if not skytools.exists_table(dcur, tbl):
                if not self.part_template:
                    raise Exception('Dest table does not exists and no way to create it.')

                sql = self.part_template
                sql = sql.replace(DEST_TABLE, skytools.quote_fqident(tbl))

                # we do this to make sure that constraints for 
                # tables who contain a schema will still work
                schema_table = tbl.replace(".", "__")
                sql = sql.replace(SCHEMA_TABLE, skytools.quote_ident(schema_table))

                dcur.execute(sql)
                dcon.commit()
                self.log.info('%s: Created table %s' % (self.job_name, tbl))

    def create_temp_table(self, curs):
        if USE_REAL_TABLE:
            tempname = self.table_name + "_loadertmpx"
        else:
            # create temp table for loading
            tempname = self.table_name.replace('.', '_') + "_loadertmp"

        # check if exists
        if USE_REAL_TABLE:
            if skytools.exists_table(curs, tempname):
                self.log.debug("bulk: Using existing real table %s" % tempname)
                return tempname, quote_fqident(tempname)

            # create non-temp table
            q = "create table %s (like %s)" % (
                        quote_fqident(tempname),
                        quote_fqident(self.table_name))
            self.log.debug("bulk: Creating real table: %s" % q)
            curs.execute(q)
            return tempname, quote_fqident(tempname)
        elif USE_LONGLIVED_TEMP_TABLES:
            if skytools.exists_temp_table(curs, tempname):
                self.log.debug("bulk: Using existing temp table %s" % tempname)
                return tempname, quote_ident(tempname)

        # bizgres crashes on delete rows
        # removed arg = "on commit delete rows"
        arg = "on commit preserve rows"
        # create temp table for loading
        q = "create temp table %s (like %s) %s" % (
                quote_ident(tempname), quote_fqident(self.table_name), arg)
        self.log.debug("bulk: Creating temp table: %s" % q)
        curs.execute(q)
        return tempname, quote_ident(tempname)

    def check_tables(self, dcon, tables):
        """Checks that tables needed for copy are there. If not
        then creates them.

        Used by other procedures to ensure that table is there
        before they start inserting.

        The commits should not be dangerous, as we haven't done anything
        with cdr's yet, so they should still be in one TX.

        Although it would be nicer to have a lock for table creation.
        """

        dcur = dcon.cursor()
        exist_map = {}
        for tbl, inf in tables.items():
            if skytools.exists_table(dcur, tbl):
                continue

            sql = self.part_template
            sql = sql.replace('_DEST_TABLE', skytools.quote_fqident(inf['table']))
            sql = sql.replace('_PARENT', skytools.quote_fqident(inf['parent']))
            sql = sql.replace('_PKEY', inf['key_list'])
            # be similar to table_dispatcher
            schema_table = inf['table'].replace(".", "__")
            sql = sql.replace('_SCHEMA_TABLE', skytools.quote_ident(schema_table))

            dcur.execute(sql)
            dcon.commit()
            self.log.info('%s: Created table %s' % (self.job_name, tbl))

    def check_table(self, t1, t2, lock_db, src_db, dst_db, setup_db):
        """Get transaction to same state, then process."""

        src_tbl = t1.dest_table
        dst_tbl = t2.dest_table

        lock_curs = lock_db.cursor()
        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()

        if not skytools.exists_table(src_curs, src_tbl):
            self.log.warning("Table %s does not exist on provider side" % src_tbl)
            return
        if not skytools.exists_table(dst_curs, dst_tbl):
            self.log.warning("Table %s does not exist on subscriber side" % dst_tbl)
            return

        # lock table against changes
        try:
            if self.provider_info['node_type'] == 'root':
                self.lock_table_root(lock_db, setup_db, dst_db, src_tbl, dst_tbl)
            else:
                self.lock_table_branch(lock_db, setup_db, dst_db, src_tbl, dst_tbl)

            # take snapshot on provider side
            src_db.commit()
            src_curs.execute("SELECT 1")

            # take snapshot on subscriber side
            dst_db.commit()
            dst_curs.execute("SELECT 1")
        finally:
            # release lock
            if self.provider_info['node_type'] == 'root':
                self.unlock_table_root(lock_db, setup_db)
            else:
                self.unlock_table_branch(lock_db, setup_db)

        # do work
        bad = self.process_sync(t1, t2, src_db, dst_db)
        if bad:
            self.bad_tables += 1

        # done
        src_db.commit()
        dst_db.commit()

    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 check_part(self, curs, dst, pkey_list):
        if skytools.exists_table(curs, dst):
            return
        if not self.split_part_template:
            raise UsageError('Partition %s does not exist and split_part_template not specified' % dst)

        vals = {
            'dest': quote_fqident(dst),
            'part': quote_fqident(dst),
            'parent': quote_fqident(self.table_name),
            'pkey': ",".join(pkey_list), # quoting?
        }
        sql = self.split_part_template % vals
        curs.execute(sql)

def is_txid_sane(curs):
    curs.execute("select txid_current()")
    txid = curs.fetchone()[0]

    # on 8.2 theres no such table
    if not skytools.exists_table(curs, 'txid.epoch'):
        return 1

    curs.execute("select epoch, last_value from txid.epoch")
    epoch, last_val = curs.fetchone()
    stored_val = (epoch << 32) | last_val

    if stored_val <= txid:
        return 1
    else:
        return 0

 def find_dist_fields(self, curs):
     if not skytools.exists_table(curs, "pg_catalog.gp_distribution_policy"):
         return []
     schema, name = skytools.fq_name_parts(self.table_name)
     q = "select a.attname"\
         "  from pg_class t, pg_namespace n, pg_attribute a,"\
         "       gp_distribution_policy p"\
         " where n.oid = t.relnamespace"\
         "   and p.localoid = t.oid"\
         "   and a.attrelid = t.oid"\
         "   and a.attnum = any(p.attrnums)"\
         "   and n.nspname = %s and t.relname = %s"
     curs.execute(q, [schema, name])
     res = []
     for row in curs.fetchall():
         res.append(row[0])
     return res

    def __init__(self, curs, table_name):
        """Initializes class by loading info about table_name from database."""

        BaseStruct.__init__(self, curs, table_name)

        self.table_name = table_name

        # fill args
        schema, name = skytools.fq_name_parts(table_name)
        args = {
            'schema': schema,
            'table': name,
            'fqname': self.fqname,
            'fq2name': skytools.quote_literal(self.fqname),
            'oid': skytools.get_table_oid(curs, table_name),
            'pg_class_oid': skytools.get_table_oid(curs, 'pg_catalog.pg_class'),
        }

        # load table struct
        self.col_list = self._load_elem(curs, self.name, args, TColumn)
        # if db is GP then read also table distribution keys
        if skytools.exists_table(curs, "pg_catalog.gp_distribution_policy"):
            self.dist_key_list = self._load_elem(curs, self.name, args,
                                                 TGPDistKey)
        else:
            self.dist_key_list = None
        self.object_list = [ TTable(table_name, self.col_list,
                                    self.dist_key_list) ]
        self.seq_list = []

        # load seqs
        for col in self.col_list:
            if col.seqname:
                fqname = quote_fqident(col.seqname)
                owner = self.fqname + '.' + quote_ident(col.name)
                seq_args = { 'fqname': fqname, 'owner': skytools.quote_literal(owner) }
                self.seq_list += self._load_elem(curs, col.seqname, seq_args, TSeq)
        self.object_list += self.seq_list

        # load additional objects
        to_load = [TColumnDefault, TConstraint, TIndex, TTrigger,
                   TRule, TGrant, TOwner, TParent]
        for eclass in to_load:
            self.object_list += self._load_elem(curs, self.name, args, eclass)

    def check_part(self, dst, part_time):
        """Create part table if not exists.

        It part_template present, execute it
        else if part function present in db, call it
        else clone master table"""
        curs = self.dst_curs
        if skytools.exists_table(curs, dst):
            return
        dst = quote_fqident(dst)
        vals = {
            "dest": dst,
            "part": dst,
            "parent": self.quoted_name,
            "pkeys": ",".join(self.pkeys),  # quoting?
            # we do this to make sure that constraints for
            # tables who contain a schema will still work
            "schema_table": dst.replace(".", "__"),
            "part_field": self.conf.part_field,
            "part_time": part_time,
            "period": self.conf.period,
        }

        def exec_with_vals(tmpl):
            if tmpl:
                sql = tmpl % vals
                curs.execute(sql)
                return True
            return False

        exec_with_vals(self.conf.pre_part)
        if not exec_with_vals(self.conf.part_template):
            self.log.debug("part_template not provided, using part func")
            # if part func exists call it with val arguments
            if skytools.exists_function(curs, PART_FUNC, len(PART_FUNC_ARGS)):
                self.log.debug("check_part.exec: func:%s, args: %s" % (PART_FUNC_CALL, vals))
                curs.execute(PART_FUNC_CALL, vals)
            else:
                self.log.debug("part func %s not found, cloning table" % PART_FUNC)
                struct = TableStruct(curs, self.table_name)
                struct.create(curs, T_ALL, dst)
        exec_with_vals(self.conf.post_part)
        self.log.info("Created table: %s" % dst)

    def check_part(self, dst, part_time):
        """Create part table if not exists.

        It part_template present, execute it
        else if part function present in db, call it
        else clone master table"""
        curs = self.dst_curs
        if skytools.exists_table(curs, dst):
            return
        dst = quote_fqident(dst)
        vals = {'dest': dst,
                'part': dst,
                'parent': self.fq_dest_table,
                'pkeys': ",".join(self.pkeys), # quoting?
                # we do this to make sure that constraints for
                # tables who contain a schema will still work
                'schema_table': dst.replace(".", "__"),
                'part_field': self.conf.part_field,
                'part_time': part_time,
                'period': self.conf.period,
                }
        def exec_with_vals(tmpl):
            if tmpl:
                sql = tmpl % vals
                curs.execute(sql)
                return True
            return False
        exec_with_vals(self.conf.pre_part)
        if not exec_with_vals(self.conf.part_template):
            self.log.debug('part_template not provided, using part func')
            # if part func exists call it with val arguments
            pfargs = ', '.join('%%(%s)s' % arg for arg in PART_FUNC_ARGS)
            pfcall = 'select %s(%s)' % (self.conf.part_func, pfargs)
            if skytools.exists_function(curs, self.conf.part_func, len(PART_FUNC_ARGS)):
                self.log.debug('check_part.exec: func:%s, args: %s' % (pfcall, vals))
                curs.execute(pfcall, vals)
            else:
                self.log.debug('part func %s not found, cloning table' % self.conf.part_func)
                struct = TableStruct(curs, self.dest_table)
                struct.create(curs, T_ALL, dst)
        exec_with_vals(self.conf.post_part)
        self.log.info("Created table: %s" % dst)

    def add_table(self, src_db, dst_db, tbl, create_flags, src_tbls):
        # use full names
        tbl = skytools.fq_name(tbl)
        dest_table = self.options.dest_table or tbl
        dest_table = skytools.fq_name(dest_table)

        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()
        tbl_exists = skytools.exists_table(dst_curs, dest_table)

        if dest_table == tbl:
            desc = tbl
        else:
            desc = "%s(%s)" % (tbl, dest_table)

        if create_flags:
            if tbl_exists:
                self.log.info('Table %s already exist, not touching' % desc)
            else:
                src_dest_table = src_tbls[tbl]['dest_table']
                if not skytools.exists_table(src_curs, src_dest_table):
                    # table not present on provider - nowhere to get the DDL from
                    self.log.warning('Table %s missing on provider, cannot create, skipping' % desc)
                    return
                schema = skytools.fq_name_parts(dest_table)[0]
                if not skytools.exists_schema(dst_curs, schema):
                    q = "create schema %s" % skytools.quote_ident(schema)
                    dst_curs.execute(q)
                s = skytools.TableStruct(src_curs, src_dest_table)
                src_db.commit()

                # create, using rename logic only when necessary
                newname = None
                if src_dest_table != dest_table:
                    newname = dest_table
                s.create(dst_curs, create_flags, log = self.log, new_table_name = newname)

        tgargs = []
        if self.options.trigger_arg:
            tgargs = self.options.trigger_arg
        tgflags = self.options.trigger_flags
        if tgflags:
            tgargs.append('tgflags='+tgflags)
        if self.options.no_triggers:
            tgargs.append('no_triggers')
        if self.options.merge_all:
            tgargs.append('merge_all')
        if self.options.no_merge:
            tgargs.append('no_merge')

        attrs = {}
        if self.options.handler:
            hstr = londiste.handler.create_handler_string(
                            self.options.handler, self.options.handler_arg)
            p = londiste.handler.build_handler(tbl, hstr, self.options.dest_table)
            attrs['handler'] = hstr
            p.add(tgargs)

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

        if self.options.expect_sync:
            tgargs.append('expect_sync')

        if not self.options.expect_sync:
            if self.options.skip_truncate:
                attrs['skip_truncate'] = 1

        if self.options.max_parallel_copy:
            attrs['max_parallel_copy'] = self.options.max_parallel_copy

        # actual table registration
        args = [self.set_name, tbl, tgargs, None, None]
        if attrs:
            args[3] = skytools.db_urlencode(attrs)
        if dest_table != tbl:
            args[4] = dest_table
        q = "select * from londiste.local_add_table(%s, %s, %s, %s, %s)"
        self.exec_cmd(dst_curs, q, args)
        dst_db.commit()

    def add_table(self, src_db, dst_db, tbl, create_flags):
        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()
        tbl_exists = skytools.exists_table(dst_curs, tbl)
        if create_flags:
            if tbl_exists:
                self.log.info("Table %s already exist, not touching" % tbl)
            else:
                if not skytools.exists_table(src_curs, tbl):
                    # table not present on provider - nowhere to get the DDL from
                    self.log.warning('Table "%s" missing on provider, skipping' % tbl)
                    return
                schema = skytools.fq_name_parts(tbl)[0]
                if not skytools.exists_schema(dst_curs, schema):
                    q = "create schema %s" % skytools.quote_ident(schema)
                    dst_curs.execute(q)
                s = skytools.TableStruct(src_curs, tbl)
                src_db.commit()
                s.create(dst_curs, create_flags, log=self.log)

        tgargs = []
        if self.options.trigger_arg:
            tgargs = self.options.trigger_arg
        tgflags = self.options.trigger_flags
        if tgflags:
            tgargs.append("tgflags=" + tgflags)
        if self.options.no_triggers:
            tgargs.append("no_triggers")
        if self.options.merge_all:
            tgargs.append("merge_all")
        if self.options.no_merge:
            tgargs.append("no_merge")

        attrs = {}
        if self.options.handler:
            hstr = londiste.handler.create_handler_string(self.options.handler, self.options.handler_arg)
            p = londiste.handler.build_handler(tbl, hstr, self.log)
            attrs["handler"] = hstr
            p.add(tgargs)

        if self.options.expect_sync:
            tgargs.append("expect_sync")

        if not self.options.expect_sync:
            if self.options.skip_truncate:
                attrs["skip_truncate"] = 1
            if self.options.copy_condition:
                attrs["copy_condition"] = self.options.copy_condition

        if self.options.max_parallel_copy:
            attrs["max_parallel_copy"] = self.options.max_parallel_copy

        args = [self.set_name, tbl, tgargs]

        if attrs:
            args.append(skytools.db_urlencode(attrs))

        q = "select * from londiste.local_add_table(%s)" % ",".join(["%s"] * len(args))

        # actual table registration
        self.exec_cmd(dst_curs, q, args)
        dst_db.commit()

 def match(self, objname, curs, tables, seqs):
     if objname not in tables:
         return False
     dest_name = tables[objname]
     return skytools.exists_table(curs, dest_name)

def detect_londiste215(curs):
    return skytools.exists_table(curs, 'londiste.subscriber_pending_fkeys')

    def process_one_table(self, dst_db, tbl, cache):

        del_list = cache.final_del_list
        ins_list = cache.final_ins_list
        upd_list = cache.final_upd_list
        col_list = cache.col_list
        real_update_count = len(upd_list)

        self.log.debug("process_one_table: %s  (I/U/D = %d/%d/%d)" % (
                       tbl, len(ins_list), len(upd_list), len(del_list)))

        if tbl in self.remap_tables:
            old = tbl
            tbl = self.remap_tables[tbl]
            self.log.debug("Redirect %s to %s" % (old, tbl))

        # hack to unbroke stuff
        if self.load_method == METH_MERGED:
            upd_list += ins_list
            ins_list = []

        # check if interesting table
        curs = dst_db.cursor()
        if not skytools.exists_table(curs, tbl):
            self.log.warning("Ignoring events for table: %s" % tbl)
            return

        # fetch distribution fields
        dist_fields = find_dist_fields(curs, tbl)
        extra_fields = []
        for fld in dist_fields:
            if fld not in cache.pkey_list:
                extra_fields.append(fld)
        self.log.debug("PKey fields: %s  Extra fields: %s" % (
                       ",".join(cache.pkey_list), ",".join(extra_fields)))

        # create temp table
        temp = self.create_temp_table(curs, tbl)
        
        # where expr must have pkey and dist fields
        klist = []
        for pk in cache.pkey_list + extra_fields:
            exp = "%s.%s = %s.%s" % (quote_fqident(tbl), quote_ident(pk),
                                     quote_fqident(temp), quote_ident(pk))
            klist.append(exp)
        whe_expr = " and ".join(klist)

        # create del sql
        del_sql = "delete from only %s using %s where %s" % (
                  quote_fqident(tbl), quote_fqident(temp), whe_expr)

        # create update sql
        slist = []
        key_fields = cache.pkey_list + extra_fields
        for col in cache.col_list:
            if col not in key_fields:
                exp = "%s = %s.%s" % (quote_ident(col), quote_fqident(temp), quote_ident(col))
                slist.append(exp)
        upd_sql = "update only %s set %s from %s where %s" % (
                    quote_fqident(tbl), ", ".join(slist), quote_fqident(temp), whe_expr)

        # insert sql
        colstr = ",".join([quote_ident(c) for c in cache.col_list])
        ins_sql = "insert into %s (%s) select %s from %s" % (
                  quote_fqident(tbl), colstr, colstr, quote_fqident(temp))

        # process deleted rows
        if len(del_list) > 0:
            self.log.info("Deleting %d rows from %s" % (len(del_list), tbl))
            # delete old rows
            q = "truncate %s" % quote_fqident(temp)
            self.log.debug(q)
            curs.execute(q)
            # copy rows
            self.log.debug("COPY %d rows into %s" % (len(del_list), temp))
            skytools.magic_insert(curs, temp, del_list, col_list)
            # delete rows
            self.log.debug(del_sql)
            curs.execute(del_sql)
            self.log.debug("%s - %d" % (curs.statusmessage, curs.rowcount))
            self.log.debug(curs.statusmessage)
            if len(del_list) != curs.rowcount:
                self.log.warning("Delete mismatch: expected=%s updated=%d"
                        % (len(del_list), curs.rowcount))

        # process updated rows
        if len(upd_list) > 0:
            self.log.info("Updating %d rows in %s" % (len(upd_list), tbl))
            # delete old rows
            q = "truncate %s" % quote_fqident(temp)
            self.log.debug(q)
            curs.execute(q)
            # copy rows
            self.log.debug("COPY %d rows into %s" % (len(upd_list), temp))
            skytools.magic_insert(curs, temp, upd_list, col_list)
            if self.load_method == METH_CORRECT:
                # update main table
                self.log.debug(upd_sql)
                curs.execute(upd_sql)
                self.log.debug(curs.statusmessage)
#.........这里部分代码省略.........

    def check_table(self, tbl, lock_db, src_db, dst_db, setup_curs):
        """Get transaction to same state, then process."""


        lock_curs = lock_db.cursor()
        src_curs = src_db.cursor()
        dst_curs = dst_db.cursor()

        if not skytools.exists_table(src_curs, tbl):
            self.log.warning("Table %s does not exist on provider side" % tbl)
            return
        if not skytools.exists_table(dst_curs, tbl):
            self.log.warning("Table %s does not exist on subscriber side" % tbl)
            return

        # lock table in separate connection
        self.log.info('Locking %s' % tbl)
        lock_db.commit()
        self.set_lock_timeout(lock_curs)
        lock_time = time.time()
        lock_curs.execute("LOCK TABLE %s IN SHARE MODE" % skytools.quote_fqident(tbl))

        # now wait until consumer has updated target table until locking
        self.log.info('Syncing %s' % tbl)

        # consumer must get futher than this tick
        tick_id = self.force_tick(setup_curs)
        # try to force second tick also
        self.force_tick(setup_curs)

        # take server time
        setup_curs.execute("select to_char(now(), 'YYYY-MM-DD HH24:MI:SS.MS')")
        tpos = setup_curs.fetchone()[0]
        # now wait
        while 1:
            time.sleep(0.5)

            q = "select now() - lag > timestamp %s, now(), lag"\
                " from pgq.get_consumer_info(%s, %s)"
            setup_curs.execute(q, [tpos, self.queue_name, self.consumer_name])
            res = setup_curs.fetchall()

            if len(res) == 0:
                raise Exception('No such consumer')

            row = res[0]
            self.log.debug("tpos=%s now=%s lag=%s ok=%s" % (tpos, row[1], row[2], row[0]))
            if row[0]:
                break

            # limit lock time
            if time.time() > lock_time + self.lock_timeout and not self.options.force:
                self.log.error('Consumer lagging too much, exiting')
                lock_db.rollback()
                sys.exit(1)
        
        # take snapshot on provider side
        src_db.commit()
        src_curs.execute("SELECT 1")

        # take snapshot on subscriber side
        dst_db.commit()
        dst_curs.execute("SELECT 1")

        # release lock
        lock_db.commit()

        # do work
        self.process_sync(tbl, src_db, dst_db)

        # done
        src_db.commit()
        dst_db.commit()