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 change_config(self):
        if len(self.args) < 3:
            list = self.get_queue_list()
            for qname in list:
                self.show_config(qname)
            return

        qname = self.args[2]
        if len(self.args) == 3:
            self.show_config(qname)
            return

        alist = []
        for el in self.args[3:]:
            k, v = el.split('=')
            if k not in config_allowed_list:
                qk = "queue_" + k
                if qk not in config_allowed_list:
                    raise Exception('unknown config var: '+k)
                k = qk
            expr = "%s=%s" % (k, skytools.quote_literal(v))
            alist.append(expr)
        self.log.info('Change queue %s config to: %s' % (qname, ", ".join(alist)))
        sql = "update pgq.queue set %s where queue_name = %s" % ( 
                        ", ".join(alist), skytools.quote_literal(qname))
        self.exec_sql(sql, [])

 def _new_obj_sql(self, queue, objname, objkind):
     args = {
         "queue": skytools.quote_literal(queue),
         "obj": objname,
         "ifield": objname + "_name",
         "itable": "londiste." + objname + "_info",
         "kind": skytools.quote_literal(objkind),
     }
     q = (
         """select quote_ident(n.nspname) || '.' || quote_ident(r.relname)
         from pg_catalog.pg_class r
         join pg_catalog.pg_namespace n on (n.oid = r.relnamespace)
         left join %(itable)s i
              on (i.queue_name = %(queue)s and
                  i.%(ifield)s = (n.nspname || '.' || r.relname))
         where r.relkind = %(kind)s
           and n.nspname not in ('pg_catalog', 'information_schema', 'pgq', 'londiste', 'pgq_node', 'pgq_ext')
           and n.nspname !~ 'pg_.*'
           and i.%(ifield)s is null
         union all
         select londiste.quote_fqname(%(ifield)s) from %(itable)s
          where queue_name = %(queue)s and not local
         order by 1 """
         % args
     )
     return q

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 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 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 get_batch_list(self):
     if not self.cur_queue:
         return []
     qname = skytools.quote_literal(self.cur_queue)
     q = "select current_batch::text from pgq.get_consumer_info(%s)"\
         " where current_batch is not null order by 1" % qname
     return self._ccache('batch_list', q, 'pgq')

def _gen_dict_insert(tbl, row, fields, qfields):
    tmp = []
    for f in fields:
        v = row.get(f)
        tmp.append(skytools.quote_literal(v))
    fmt = "insert into %s (%s) values (%s);"
    return fmt % (tbl, ",".join(qfields), ",".join(tmp))

 def addcmp(self, list, f, v):
     if v is None:
         s = "%s is null" % f
     else:
         vq = skytools.quote_literal(v)
         s = "%s = %s" % (f, vq)
     list.append(s)

    def process_remote_batch(self, src_db, tick_id, ev_list, dst_db):
        "All work for a batch.  Entry point from SetConsumer."

        # this part can play freely with transactions

        if not self.code_check_done:
            self.check_code(dst_db)
            self.code_check_done = 1

        self.sync_database_encodings(src_db, dst_db)

        self.cur_tick = self.batch_info["tick_id"]
        self.prev_tick = self.batch_info["prev_tick_id"]

        dst_curs = dst_db.cursor()
        self.load_table_state(dst_curs)
        self.sync_tables(src_db, dst_db)

        self.copy_snapshot_cleanup(dst_db)

        # only main thread is allowed to restore fkeys
        if not self.copy_thread:
            self.restore_fkeys(dst_db)

        for p in self.used_plugins.values():
            p.reset()
        self.used_plugins = {}

        # now the actual event processing happens.
        # they must be done all in one tx in dst side
        # and the transaction must be kept open so that
        # the cascade-consumer can save last tick and commit.

        self.sql_list = []
        CascadedWorker.process_remote_batch(self, src_db, tick_id, ev_list, dst_db)
        self.flush_sql(dst_curs)

        for p in self.used_plugins.values():
            p.finish_batch(self.batch_info, dst_curs)
        self.used_plugins = {}

        # finalize table changes
        self.save_table_state(dst_curs)

        # store event filter
        if self.cf.getboolean("local_only", False):
            if self.copy_thread:
                _filterlist = skytools.quote_literal(self.copy_table_name)
            else:
                _filterlist = ",".join(map(skytools.quote_literal, self.table_map.keys()))
            self.consumer_filter = (
                """
((ev_type like 'pgq%%' or ev_type like 'londiste%%')
or (ev_extra1 in (%s)))
"""
                % _filterlist
            )
        else:
            self.consumer_filter = None

 def addcmp(self, list, f, v):
     """Add quoted comparision."""
     if v is None:
         s = "%s is null" % f
     else:
         vq = skytools.quote_literal(v)
         s = "%s = %s" % (f, vq)
     list.append(s)

 def get_known_seq_list(self):
     if not self.cur_queue:
         return []
     qname = skytools.quote_literal(self.cur_queue)
     q = "select londiste.quote_fqname(seq_name)"\
         " from londiste.seq_info"\
         " where queue_name = %s order by 1" % qname
     return self._ccache('known_seq_list', q, 'londiste')

 def __str__(self):
     if self.conf.param_type == PARAM_INLINE:
         return skytools.quote_literal(self.value)
     elif self.conf.param_type == PARAM_DBAPI:
         return "%s"
     elif self.conf.param_type == PARAM_PLPY:
         return "$%d" % self.pos
     else:
         raise Exception("bad QArgConf.param_type")

    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 get_known_seq_list(self):
     if not self.cur_queue:
         return []
     qname = skytools.quote_literal(self.cur_queue)
     q = (
         "select londiste.quote_fqname(seq_name)"
         " from londiste.seq_info"
         " where queue_name = %s order by 1" % qname
     )
     return self._ccache("known_seq_list", q, "londiste")

 def got_missed_insert(self, tbl, src_row):
     self.cnt_insert += 1
     fld_list = self.common_fields
     val_list = []
     for f in fld_list:
         v = unescape(src_row[f])
         val_list.append(skytools.quote_literal(v))
     q = "insert into %s (%s) values (%s);" % (
             tbl, ", ".join(fld_list), ", ".join(val_list))
     self.show_fix(tbl, q, 'insert')

def _gen_list_insert(tbl, row, fields, qfields):
    tmp = []
    for i in range(len(fields)):
        try:
            v = row[i]
        except IndexError:
            v = None
        tmp.append(skytools.quote_literal(v))
    fmt = "insert into %s (%s) values (%s);"
    return fmt % (tbl, ",".join(qfields), ",".join(tmp))

 def mk_insert_sql(self, tbl, key_list, data):
     # generate insert command
     col_list = []
     val_list = []
     for c, v in data.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 mk_insert_sql(row, tbl, pkey_list=None, field_map=None):
    """Generate INSERT statement from dict data.

    >>> mk_insert_sql({'id': '1', 'data': None}, 'tbl')
    "insert into public.tbl (data, id) values (null, '1');"
    """

    col_list = []
    val_list = []
    if field_map:
        for src, dst in field_map.iteritems():
            col_list.append(skytools.quote_ident(dst))
            val_list.append(skytools.quote_literal(row[src]))
    else:
        for c, v in row.iteritems():
            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 mk_delete_sql(row, tbl, pkey_list, field_map = None):
    """Generate DELETE statement from dict data."""

    if len(pkey_list) < 1:
        raise Exception("delete needs pkeys")
    whe_list = []
    for k in pkey_list:
        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))
    whe_str = " and ".join(whe_list)
    return "delete from only %s where %s;" % (skytools.quote_fqident(tbl), whe_str)