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 mk_insert_sql(row, tbl, pkey_list=None, field_map=None):
    """Generate INSERT statement from dict data.

    >>> from collections import OrderedDict
    >>> row = OrderedDict([('id',1), ('data', None)])
    >>> mk_insert_sql(row, 'tbl')
    "insert into public.tbl (id, data) values ('1', null);"
    >>> mk_insert_sql(row, 'tbl', ['x'], OrderedDict([('id', 'id_'), ('data', 'data_')]))
    "insert into public.tbl (id_, data_) values ('1', null);"

    """

    col_list = []
    val_list = []
    if field_map:
        for src, dst in field_map.items():
            col_list.append(skytools.quote_ident(dst))
            val_list.append(skytools.quote_literal(row[src]))
    else:
        for c, v in row.items():
            col_list.append(skytools.quote_ident(c))
            val_list.append(skytools.quote_literal(v))
    col_str = ", ".join(col_list)
    val_str = ", ".join(val_list)
    return "insert into %s (%s) values (%s);" % (
                    skytools.quote_fqident(tbl), col_str, val_str)

    def gen_copy_tbl(self, tbl, src_curs, dst_curs):
        """Create COPY expession from common fields."""
        self.pkey_list = get_pkey_list(src_curs, tbl)
        dst_pkey = get_pkey_list(dst_curs, tbl)
        if dst_pkey != self.pkey_list:
            self.log.error('pkeys do not match')
            sys.exit(1)

        src_cols = get_column_list(src_curs, tbl)
        dst_cols = get_column_list(dst_curs, tbl)
        field_list = []
        for f in self.pkey_list:
            field_list.append(f)
        for f in src_cols:
            if f in self.pkey_list:
                continue
            if f in dst_cols:
                field_list.append(f)

        self.common_fields = field_list

        fqlist = [skytools.quote_ident(col) for col in field_list]

        tbl_expr = "%s (%s)" % (skytools.quote_ident(tbl), ",".join(fqlist))

        self.log.debug("using copy expr: %s" % tbl_expr)

        return tbl_expr

def mk_update_sql(row, tbl, pkey_list, field_map = None):
    r"""Generate UPDATE statement from dict data.

    >>> mk_update_sql({'id': 0, 'id2': '2', 'data': 'str\\'}, 'Table', ['id', 'id2'])
    'update only public."Table" set data = E\'str\\\\\' where id = \'0\' and id2 = \'2\';'
    """

    if len(pkey_list) < 1:
        raise Exception("update needs pkeys")
    set_list = []
    whe_list = []
    pkmap = {}
    for k in pkey_list:
        pkmap[k] = 1
        new_k = field_map and field_map[k] or k
        col = skytools.quote_ident(new_k)
        val = skytools.quote_literal(row[k])
        whe_list.append("%s = %s" % (col, val))

    if field_map:
        for src, dst in field_map.iteritems():
            if src not in pkmap:
                col = skytools.quote_ident(dst)
                val = skytools.quote_literal(row[src])
                set_list.append("%s = %s" % (col, val))
    else:
        for col, val in row.iteritems():
            if col not in pkmap:
                col = skytools.quote_ident(col)
                val = skytools.quote_literal(val)
                set_list.append("%s = %s" % (col, val))
    return "update only %s set %s where %s;" % (skytools.quote_fqident(tbl),
            ", ".join(set_list), " and ".join(whe_list))

 def _sync_listen(self, new_clist):
     if not new_clist and not self.listen_channel_list:
         return
     curs = self.conn.cursor()
     for ch in self.listen_channel_list:
         if ch not in new_clist:
             curs.execute("UNLISTEN %s" % skytools.quote_ident(ch))
     for ch in new_clist:
         if ch not in self.listen_channel_list:
             curs.execute("LISTEN %s" % skytools.quote_ident(ch))
     if self.isolation_level != skytools.I_AUTOCOMMIT:
         self.conn.commit()
     self.listen_channel_list = new_clist[:]

 def _where(self):
     tmpl = "%(tbl)s.%(col)s = t.%(col)s"
     stmt = (tmpl % {'col': quote_ident(f),
                      'tbl': self.qtable,
                     }
             for f in self.keys)
     return ' and '.join(stmt)

    def gen_copy_tbl(self, src_curs, dst_curs, where):
        """Create COPY expession from common fields."""
        self.pkey_list = skytools.get_table_pkeys(src_curs, self.table_name)
        dst_pkey = skytools.get_table_pkeys(dst_curs, self.table_name)
        if dst_pkey != self.pkey_list:
            self.log.error('pkeys do not match')
            sys.exit(1)

        src_cols = skytools.get_table_columns(src_curs, self.table_name)
        dst_cols = skytools.get_table_columns(dst_curs, self.table_name)
        field_list = []
        for f in self.pkey_list:
            field_list.append(f)
        for f in src_cols:
            if f in self.pkey_list:
                continue
            if f in dst_cols:
                field_list.append(f)

        self.common_fields = field_list

        fqlist = [skytools.quote_ident(col) for col in field_list]

        tbl_expr = "select %s from %s" % (",".join(fqlist), self.fq_table_name)
        if where:
            tbl_expr += ' where ' + where
        tbl_expr = "COPY (%s) TO STDOUT" % tbl_expr

        self.log.debug("using copy expr: %s" % tbl_expr)

        return tbl_expr

 def get_create_sql(self, curs, new_table_name=None):
     """Generate creation SQL."""
     if new_table_name:
         # fixme: seems broken
         iname = find_new_name(curs, self.name)
         tname = new_table_name
         pnew = "INDEX %s ON %s " % (quote_ident(iname), quote_fqident(tname))
         rx = r"\bINDEX[ ][a-z0-9._]+[ ]ON[ ][a-z0-9._]+[ ]"
         sql = rx_replace(rx, self.defn, pnew)
     else:
         sql = self.defn
         iname = self.local_name
         tname = self.table_name
     if self.is_clustered:
         sql += " ALTER TABLE ONLY %s\n  CLUSTER ON %s;" % (quote_fqident(tname), quote_ident(iname))
     return sql

def magic_insert(curs, tablename, data, fields = None, use_insert = 0, quoted_table = False):
    r"""Copy/insert a list of dict/list data to database.

    If curs == None, then the copy or insert statements are returned
    as string.  For list of dict the field list is optional, as its
    possible to guess them from dict keys.

    Example:
    >>> magic_insert(None, 'tbl', [[1, '1'], [2, '2']], ['col1', 'col2'])
    'COPY public.tbl (col1,col2) FROM STDIN;\n1\t1\n2\t2\n\\.\n'
    """
    if len(data) == 0:
        return

    # decide how to process
    if hasattr(data[0], 'keys'):
        if fields == None:
            fields = data[0].keys()
        if use_insert:
            row_func = _gen_dict_insert
        else:
            row_func = _gen_dict_copy
    else:
        if fields == None:
            raise Exception("Non-dict data needs field list")
        if use_insert:
            row_func = _gen_list_insert
        else:
            row_func = _gen_list_copy

    qfields = [skytools.quote_ident(f) for f in fields]
    if quoted_table:
        qtablename = tablename
    else:
        qtablename = skytools.quote_fqident(tablename)

    # init processing
    buf = StringIO()
    if curs == None and use_insert == 0:
        fmt = "COPY %s (%s) FROM STDIN;\n"
        buf.write(fmt % (qtablename, ",".join(qfields)))

    # process data
    for row in data:
        buf.write(row_func(qtablename, row, fields, qfields))
        buf.write("\n")

    # if user needs only string, return it
    if curs == None:
        if use_insert == 0:
            buf.write("\\.\n")
        return buf.getvalue()

    # do the actual copy/inserts
    if use_insert:
        curs.execute(buf.getvalue())
    else:
        buf.seek(0)
        hdr = "%s (%s)" % (qtablename, ",".join(qfields))
        curs.copy_from(buf, hdr)

 def got_missed_delete(self, tbl, dst_row):
     self.cnt_delete += 1
     whe_list = []
     for f in self.pkey_list:
         self.addcmp(whe_list, skytools.quote_ident(f), unescape(dst_row[f]))
     q = "delete from only %s where %s;" % (skytools.quote_fqident(tbl), " and ".join(whe_list))
     self.show_fix(tbl, q, 'delete')

    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

        # 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 load_common_columns(self, src_tbl, dst_tbl, src_curs, dst_curs):
        """Get common fields, put pkeys in start."""

        self.pkey_list = skytools.get_table_pkeys(src_curs, src_tbl)
        dst_pkey = skytools.get_table_pkeys(dst_curs, dst_tbl)
        if dst_pkey != self.pkey_list:
            self.log.error('pkeys do not match')
            sys.exit(1)

        src_cols = skytools.get_table_columns(src_curs, src_tbl)
        dst_cols = skytools.get_table_columns(dst_curs, dst_tbl)
        field_list = []
        for f in self.pkey_list:
            field_list.append(f)
        for f in src_cols:
            if f in self.pkey_list:
                continue
            if f in dst_cols:
                field_list.append(f)

        self.common_fields = field_list

        fqlist = [skytools.quote_ident(col) for col in field_list]
        self.fq_common_fields = fqlist

        cols = ",".join(fqlist)
        self.log.debug("using columns: %s" % cols)

    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 mk_delete_sql(self, tbl, key_list, data):
     # generate delete command
     whe_list = []
     for k in key_list:
         whe_list.append("%s = %s" % (skytools.quote_ident(k), skytools.quote_literal(data[k])))
     whe_str = " and ".join(whe_list)
     return "delete from %s where %s;" % (skytools.quote_fqident(tbl), whe_str)

    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 add(self, trigger_arg_list):
        """Let trigger put hash into extra3"""

        arg = "ev_extra3='hash='||hashtext(%s)" % skytools.quote_ident(self.key)
        trigger_arg_list.append(arg)

        BaseHandler.add(self, trigger_arg_list)

    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 calc_cols(self, src_curs, src_tbl, dst_curs, dst_tbl):
        cols1 = self.load_cols(src_curs, src_tbl)
        cols2 = self.load_cols(dst_curs, dst_tbl)

        qcols = []
        for c in self.calc_common(cols1, cols2):
            qcols.append(skytools.quote_ident(c))
        return "(%s)" % ",".join(qcols)

 def got_missed_delete(self, dst_row, fn):
     """Create sql for missed delete."""
     self.cnt_delete += 1
     whe_list = []
     for f in self.pkey_list:
         self.addcmp(whe_list, skytools.quote_ident(f), skytools.unescape_copy(dst_row[f]))
     q = "delete from only %s where %s;" % (self.fq_table_name, " and ".join(whe_list))
     self.show_fix(q, 'delete', fn)