def process_local_event(self, db, batch_id, ev):
        if ev.ev_type[:2] not in ('I:', 'U:', 'D:'):
            return

        if ev.ev_data is None:
            payload = {}
        else:
            payload = skytools.db_urldecode(ev.ev_data)

        payload['pgq.tick_id'] = self.batch_info['cur_tick_id']
        payload['pgq.ev_id'] = ev.ev_id
        payload['pgq.ev_time'] = ev.ev_time
        payload['pgq.ev_type'] = ev.ev_type
        payload['pgq.ev_data'] = ev.ev_data
        payload['pgq.ev_extra1'] = ev.ev_extra1
        payload['pgq.ev_extra2'] = ev.ev_extra2
        payload['pgq.ev_extra3'] = ev.ev_extra3
        payload['pgq.ev_extra4'] = ev.ev_extra4

        self.log.debug(self.dst_query, payload)
        retries, curs = self.execute_with_retry('dst_db', self.dst_query, payload,
                                                exceptions = (psycopg2.OperationalError,))
        if curs.statusmessage[:6] == 'SELECT':
            res = curs.fetchall()
            self.log.debug(res)
        else:
            self.log.debug(curs.statusmessage)

    def process_event(self, ev, sql_queue_func, arg):
        if len(ev.ev_type) < 2 or ev.ev_type[1] != ":":
            raise Exception("Unsupported event type: %s/extra1=%s/data=%s" % (ev.ev_type, ev.ev_extra1, ev.ev_data))
        op = ev.ev_type[0]
        if op not in "IUD":
            raise Exception("Unknown event type: " + ev.ev_type)
        # pkey_list = ev.ev_type[2:].split(',')
        data = skytools.db_urldecode(ev.ev_data)

        # get pkey value
        if self.pkey_list is None:
            # self.pkey_list = pkey_list
            self.pkey_list = ev.ev_type[2:].split(",")
        if len(self.pkey_list) > 0:
            pk_data = tuple(data[k] for k in self.pkey_list)
        elif op == "I":
            # fake pkey, just to get them spread out
            pk_data = self.fake_seq
            self.fake_seq += 1
        else:
            raise Exception("non-pk tables not supported: %s" % self.table_name)

        # get full column list, detect added columns
        if not self.col_list:
            self.col_list = data.keys()
        elif self.col_list != data.keys():
            # ^ supposedly python guarantees same order in keys()
            self.col_list = data.keys()

        # keep all versions of row data
        ev = BulkEvent(op, data, pk_data)
        if ev.pk_data in self.pkey_ev_map:
            self.pkey_ev_map[ev.pk_data].append(ev)
        else:
            self.pkey_ev_map[ev.pk_data] = [ev]

    def process_local_event(self, db, batch_id, ev):
        curs = self.get_database('dst_db', autocommit = 1).cursor()

        if ev.ev_type[:2] not in ('I:', 'U:', 'D:'):
            return

        if ev.ev_data is None:
            payload = {}
        else:
            payload = skytools.db_urldecode(ev.ev_data)

        payload['pgq.tick_id'] = self.batch_info['cur_tick_id']
        payload['pgq.ev_id'] = ev.ev_id
        payload['pgq.ev_time'] = ev.ev_time
        payload['pgq.ev_type'] = ev.ev_type
        payload['pgq.ev_data'] = ev.ev_data
        payload['pgq.ev_extra1'] = ev.ev_extra1
        payload['pgq.ev_extra2'] = ev.ev_extra2
        payload['pgq.ev_extra3'] = ev.ev_extra3
        payload['pgq.ev_extra4'] = ev.ev_extra4

        self.log.debug(self.dst_query, payload)
        curs.execute(self.dst_query, payload)
        if curs.statusmessage[:6] == 'SELECT':
            res = curs.fetchall()
            self.log.debug(res)
        else:
            self.log.debug(curs.statusmessage)

    def register_copy_consumer(self):
        dst_db = self.get_database('db')
        dst_curs = dst_db.cursor()

        # fetch table attrs
        q = "select * from londiste.get_table_list(%s) where table_name = %s"
        dst_curs.execute(q, [ self.queue_name, self.copy_table_name ])
        rows = dst_curs.fetchall()
        attrs = {}
        if len(rows) > 0:
            v_attrs = rows[0]['table_attrs']
            if v_attrs:
                attrs = skytools.db_urldecode(v_attrs)

        # do we have node here?
        if 'copy_node' in attrs:
            # take node from attrs
            source_node = attrs['copy_node']
            q = "select * from pgq_node.get_queue_locations(%s) where node_name = %s"
            dst_curs.execute(q, [ self.queue_name, source_node ])
            rows = dst_curs.fetchall()
            if len(rows):
                source_location = rows[0]['node_location']
        else:
            # fetch parent consumer state
            q = "select * from pgq_node.get_consumer_state(%s, %s)"
            rows = self.exec_cmd(dst_db, q, [ self.queue_name, self.old_consumer_name ])
            state = rows[0]
            source_node = state['provider_node']
            source_location = state['provider_location']

        self.log.info("Using '%s' as source node", source_node)
        self.register_consumer(source_location)

    def make_sql(self, tbl, ev):
        """Return SQL statement(s) for that event."""
        
        # parse data
        data = skytools.db_urldecode(ev.data)
            
        # parse tbl info
        if ev.type.find(':') > 0:
            op, keys = ev.type.split(':')
        else:
            op = ev.type
            keys = ev.extra2
        ev.key_list = keys
        key_list = keys.split(',')
        if self.keep_latest and len(key_list) == 0:
            raise Exception('No pkey on table %s' % tbl)

        # generate sql
        if op in ('I', 'U'):
            if self.keep_latest:
                sql = "%s %s" % (self.mk_delete_sql(tbl, key_list, data),
                                 self.mk_insert_sql(tbl, key_list, data))
            else:
                sql = self.mk_insert_sql(tbl, key_list, data)
        elif op == "D":
            if not self.keep_latest:
                raise Exception('Delete op not supported if mode=keep_all')

            sql = self.mk_delete_sql(tbl, key_list, data)
        else:
            raise Exception('Unknown row op: %s' % op)
        return sql

 def process_batch(self, res, mcur, bres):
     """ Process events in autocommit mode reading results back and trying to make some sense out of them
     """
     try:
         count = 0
         item = bres.copy()
         for i in res:   # for each row in read query result
             item.update(i)
             mcur.execute(self.sql_modify, item)
             self.log.debug(mcur.query)
             if mcur.statusmessage.startswith('SELECT'): # if select was used we can expect some result
                 mres = mcur.fetchall()
                 for r in mres:
                     if 'stats' in r: # if specially handled column 'stats' is present
                         for k, v in skytools.db_urldecode(r['stats'] or '').items():
                             self.stat_increase(k, int(v))
                     self.log.debug(r)
             else:
                 self.stat_increase('processed', mcur.rowcount)
                 self.log.debug(mcur.statusmessage)
             if 'cnt' in item:
                 count += item['cnt']
                 self.stat_increase("count", item['cnt'])
             else:
                 count += 1
                 self.stat_increase("count")
             if self.last_sigint:
                 break
         return count, item
     except: # process has crashed, run sql_crash and re-raise the exception
         if self.sql_crash:
             dbc = self.get_database("dbcrash", autocommit=1)
             ccur = dbc.cursor()
             ccur.execute(self.sql_crash, item)
         raise

    def process_event(self, ev, sql_queue_func, arg):
        """Process a event.
        Event should be added to sql_queue or executed directly.
        """
        if self.conf.table_mode == 'ignore':
            return
        # get data
        data = skytools.db_urldecode(ev.data)
        if len(ev.ev_type) < 2 or ev.ev_type[1] != ':':
            raise Exception('Unsupported event type: %s/extra1=%s/data=%s' % (
                            ev.ev_type, ev.ev_extra1, ev.ev_data))
        op, pkeys = ev.type.split(':', 1)
        if op not in 'IUD':
            raise Exception('Unknown event type: %s' % ev.ev_type)
        # process only operations specified
        if not op in self.conf.event_types:
            return
        self.log.debug('dispatch.process_event: %s/%s' % (
            ev.ev_type, ev.ev_data))
        if self.pkeys is None:
            self.pkeys = self.filter_pkeys(pkeys.split(','))
        data = self.filter_data(data)
        # prepare split table when needed
        if self.conf.table_mode == 'part':
            dst, part_time = self.split_format(ev, data)
            if dst not in self.row_handler.table_map:
                self.check_part(dst, part_time)
        else:
            dst = self.table_name

        if dst not in self.row_handler.table_map:
            self.row_handler.add_table(dst, LOADERS[self.conf.load_mode],
                                    self.pkeys, self.conf)
        self.row_handler.process(dst, op, data)

    def loaded_state(self, row):
        """Update object with info from db."""

        self.log.debug("loaded_state: %s: %s / %s",
                       self.name, row['merge_state'], row['custom_snapshot'])
        self.change_snapshot(row['custom_snapshot'], 0)
        self.state = self.parse_state(row['merge_state'])
        self.changed = 0
        if row['table_attrs']:
            self.table_attrs = skytools.db_urldecode(row['table_attrs'])
        else:
            self.table_attrs = {}
        self.copy_role = row['copy_role']
        self.dropped_ddl = row['dropped_ddl']
        if row['merge_state'] == "?":
            self.changed = 1

        self.copy_pos = int(row.get('copy_pos','0'))
        self.max_parallel_copy = int(self.table_attrs.get('max_parallel_copy',
                                                        self.max_parallel_copy))

        if 'dest_table' in row and row['dest_table']:
            self.dest_table = row['dest_table']
        else:
            self.dest_table = self.name

        hstr = self.table_attrs.get('handlers', '') # compat
        hstr = self.table_attrs.get('handler', hstr)
        self.plugin = build_handler(self.name, hstr, self.dest_table)

def ts_conflict_handler(gd, args):
    """Conflict handling based on timestamp column."""

    conf = skytools.db_urldecode(args[0])
    timefield = conf['timefield']
    ev_type = args[1]
    ev_data = args[2]
    ev_extra1 = args[3]
    ev_extra2 = args[4]
    ev_extra3 = args[5]
    ev_extra4 = args[6]
    altpk = None
    if 'altpk' in conf:
        altpk = conf['altpk'].split(',')

    def ts_canapply(rnew, rold):
        return canapply_tstamp_helper(rnew, rold, timefield)

    return applyrow(ev_extra1, ev_type, ev_data,
                    backup_row = ev_extra2,
                    alt_pkey_cols = altpk,
                    fkey_ref_table = conf.get('fkey_ref_table'),
                    fkey_ref_cols = conf.get('fkey_ref_cols'),
                    fkey_cols = conf.get('fkey_cols'),
                    fn_canapply = ts_canapply)

 def process_event(self, event, hbase):
   if event.ev_extra1 in self.table_mappings:
     table_mapping = self.table_mappings[event.ev_extra1]
   else:
     self.log.info("table name not found in config, skipping event")
     return
   #hbase.validate_table_name(table_mapping.hbase_table_name)
   #hbase.validate_column_descriptors(table_mapping.hbase_table_name, table_mapping.hbase_column_descriptors)
   event_data = skytools.db_urldecode(event.data)
   event_type = event.type.split(':')[0]
   
   batch = BatchMutation()
   batch.row = table_mapping.hbase_row_prefix + str(event_data[table_mapping.psql_key_column])
       
   batch.mutations = []
   for psql_column, hbase_column in zip(table_mapping.psql_columns, table_mapping.hbase_column_descriptors):
     if event_type == INSERT or event_type == UPDATE:
       m = Mutation()
       m.column = hbase_column
       m.value = str(event_data[psql_column])
     elif event_type == DELETE:
       # delete this column entry
       m = Mutation()
       m.isDelete = True
       m.column = hbase_column
     else:
       raise Exception("Invalid event type: %s, event data was: %s" % (event_type, str(event_data)))
     batch.mutations.append(m)
   hbase.client.mutateRow(table_mapping.hbase_table_name, batch.row, batch.mutations)
   event.tag_done()

 def __init__(self, row):
     self.table_name = row["table_name"]
     self.dest_table = row["dest_table"] or row["table_name"]
     self.merge_state = row["merge_state"]
     attrs = row["table_attrs"] or ""
     self.table_attrs = skytools.db_urldecode(attrs)
     hstr = self.table_attrs.get("handler", "")
     self.plugin = build_handler(self.table_name, hstr, row["dest_table"])

def handler_allows_copy(table_attrs):
    """Decide if table is copyable based on attrs."""
    if not table_attrs:
        return True
    attrs = skytools.db_urldecode(table_attrs)
    hstr = attrs.get('handler', '')
    p = londiste.handler.build_handler('unused.string', hstr, None)
    return p.needs_table()

 def cmd_tables(self):
     """Show attached tables."""
     q = """select table_name, merge_state, table_attrs
     from londiste.get_table_list(%s) where local"""
     db = self.get_database('db')
     self.display_table(db, "Tables on node", q, [self.set_name],
                        fieldfmt = {'table_attrs': lambda f: '' if f is None
                                    else skytools.db_urldecode(f)})

 def __init__(self, row):
     self.table_name = row['table_name']
     self.dest_table = row['dest_table'] or row['table_name']
     self.merge_state = row['merge_state']
     attrs = row['table_attrs'] or ''
     self.table_attrs = skytools.db_urldecode(attrs)
     hstr = self.table_attrs.get('handler', '')
     self.plugin = build_handler(self.table_name, hstr, row['dest_table'])

 def process_event(self, ev, sql_queue_func, arg):
     """Filter event by hash in extra3, apply only local slots."""
     if ev.extra3:
         meta = skytools.db_urldecode(ev.extra3)
         slot = int(meta['hash']) & self.bubbles_max_slot
         if slot not in self.bubbles_local_slots:
             return
     BaseHandler.process_event(self, ev, sql_queue_func, arg)

 def process_event(self, ev, sql_queue_func, arg):
     """Filter event by hash in extra3, apply only if for local shard."""
     if ev.extra3 and self.hash_key is not None:
         meta = skytools.db_urldecode(ev.extra3)
         self.log.debug('shard.process_event: hash=%i, hash_mask=%i, shard_nr=%i',
                        int(meta['hash']), self.hash_mask, self.shard_nr)
         if (int(meta['hash']) & self.hash_mask) != self.shard_nr:
             self.log.debug('shard.process_event: not my event')
             return
     self._process_event(ev, sql_queue_func, arg)

 def process_event(self, ev, sql_queue_func, arg):
     """Filter event by hash in extra3, apply only local part."""
     if ev.extra3:
         meta = skytools.db_urldecode(ev.extra3)
         self.log.debug('part.process_event: hash=%d, max_part=%s, local_part=%d',
                        int(meta['hash']), self.max_part, self.local_part)
         if (int(meta['hash']) & self.max_part) != self.local_part:
             self.log.debug('part.process_event: not my event')
             return
     self.log.debug('part.process_event: my event, processing')
     TableHandler.process_event(self, ev, sql_queue_func, arg)

    def __init__(self, queue_name, nst):
        self.node_type = nst['node_type']
        self.node_name = nst['node_name']
        self.local_watermark = nst['local_watermark']
        self.global_watermark = nst['global_watermark']

        self.node_attrs = {}
        attrs = nst.get('node_attrs', '')
        if attrs:
            self.node_attrs = skytools.db_urldecode(attrs)

        ntype = nst['node_type']
        ctype = nst['combined_type']
        if ntype == 'root':
            self.global_wm_event = 1
            self.local_wm_publish = 0
        elif ntype == 'branch':
            self.target_queue = queue_name
            self.process_batch = 1
            self.process_events = 1
            self.copy_events = 1
            self.process_tick_event = 1
            self.keep_event_ids = 1
            self.create_tick = 1
            if 'sync_watermark' in self.node_attrs:
                slist = self.node_attrs['sync_watermark']
                self.sync_watermark = 1
                self.wm_sync_nodes = slist.split(',')
            else:
                self.process_global_wm = 1
        elif ntype == 'leaf' and not ctype:
            self.process_batch = 1
            self.process_events = 1
        elif ntype == 'leaf' and ctype:
            self.target_queue = nst['combined_queue']
            if ctype == 'root':
                self.process_batch = 1
                self.process_events = 1
                self.copy_events = 1
                self.filtered_copy = 1
                self.send_tick_event = 1
            elif ctype == 'branch':
                self.process_batch = 1
                self.wait_behind = 1
            else:
                raise Exception('invalid state 1')
        else:
            raise Exception('invalid state 2')
        if ctype and ntype != 'leaf':
            raise Exception('invalid state 3')

def _parse_handler(hstr):
    """Parse result of create_handler_string()."""
    args = {}
    name = hstr
    pos = hstr.find('(')
    if pos > 0:
        name = hstr[ : pos]
        if hstr[-1] != ')':
            raise Exception('invalid handler format: %s' % hstr)
        astr = hstr[pos + 1 : -1]
        if astr:
            astr = astr.replace(',', '&')
            args = skytools.db_urldecode(astr)
    return (name, args)

    def add_event(self, ev):
        """Store new event."""

        # op & data
        ev.op = ev.ev_type[0]
        ev.data = skytools.db_urldecode(ev.ev_data)

        # get pkey column names
        if self.pkey_str is None:
            if len(ev.ev_type) > 2:
                self.pkey_str = ev.ev_type.split(':')[1]
            else:
                self.pkey_str = ev.ev_extra2

            if self.pkey_str:
                self.pkey_list = self.pkey_str.split(',')

        # get pkey value
        if self.pkey_str:
            pk_data = []
            for k in self.pkey_list:
                pk_data.append(ev.data[k])
            ev.pk_data = tuple(pk_data)
        elif ev.op == 'I':
            # fake pkey, just to get them spread out
            ev.pk_data = ev.id
        else:
            raise Exception('non-pk tables not supported: %s' % self.name)

        # get full column list, detect added columns
        if not self.col_list:
            self.col_list = ev.data.keys()
        elif self.col_list != ev.data.keys():
            # ^ supposedly python guarantees same order in keys()

            # find new columns
            for c in ev.data.keys():
                if c not in self.col_list:
                    for oldev in self.ev_list:
                        oldev.data[c] = None
            self.col_list = ev.data.keys()

        # add to list
        self.ev_list.append(ev)

        # keep all versions of row data
        if ev.pk_data in self.pkey_map:
            self.pkey_map[ev.pk_data].append(ev)
        else:
            self.pkey_map[ev.pk_data] = [ev]