londiste: added ignore_truncate handler option

Truncate events can be ignored on per handler basis.

diff --git a/python/londiste/handler.py b/python/londiste/handler.py
index 8ce45cff3ad12158782d071a7e34baedf3fea7f1..18a81ae0ff5393d09783702a7adcdb92bc481a4e 100644 (file)
--- a/python/londiste/handler.py
+++ b/python/londiste/handler.py
@@ -176,6 +176,7 @@ class TableHandler(BaseHandler):
     Parameters:
       encoding=ENC - Validate and fix incoming data from encoding.
                      Only 'utf8' is supported at the moment.
+      ignore_truncate=BOOL - Ignore truncate event. Default: 0; Values: 0,1.
     """
     handler_name = 'londiste'
 
@@ -196,6 +197,11 @@ class TableHandler(BaseHandler):
         else:
             self.encoding_validator = None
 
+    def get_config (self):
+        conf = BaseHandler.get_config(self)
+        conf.ignore_truncate = self.get_arg('ignore_truncate', [0, 1], 0)
+        return conf
+
     def process_event(self, ev, sql_queue_func, arg):
         row = self.parse_row_data(ev)
         if len(ev.type) == 1:

diff --git a/python/londiste/handlers/dispatch.py b/python/londiste/handlers/dispatch.py
index 2e1c3143cfecee04185432fbc87faaee567bbf0f..d139428c91d53f17213c7dd31e4e3f5d57797c7e 100644 (file)
--- a/python/londiste/handlers/dispatch.py
+++ b/python/londiste/handlers/dispatch.py
@@ -142,6 +142,10 @@ ignore_old_events:
     * 0 - handle all events in the same way (default)
     * 1 - ignore events coming for obsolete partitions
 
+ignore_truncate:
+    * 0 - process truncate event (default)
+    * 1 - ignore truncate event
+
 encoding:
     name of destination encoding. handler replaces all invalid encoding symbols
     and logs them as warnings

diff --git a/python/londiste/handlers/shard.py b/python/londiste/handlers/shard.py
index 329c6cf4e2cb77948e795909934b247870d432d4..69b83cdbaef3a9693e4d78685f030b9573156948 100644 (file)
--- a/python/londiste/handlers/shard.py
+++ b/python/londiste/handlers/shard.py
@@ -6,6 +6,7 @@ Parameters:
   hashfunc=NAME: function to use for hashing (default: partconf.get_hash_raw)
   hashexpr=EXPR: full expression to use for hashing (deprecated)
   encoding=ENC: validate and fix incoming data (only utf8 supported atm)
+  ignore_truncate=BOOL: ignore truncate event, default: 0, values: 0,1
 
 On root node:
 * Hash of key field will be added to ev_extra3.

diff --git a/python/londiste/playback.py b/python/londiste/playback.py
index bc20920cdb37aa91ab47ef60d04867bbf0eafc79..f21d7e582ff949153aeeedec2d37e0b228e10fd4 100644 (file)
--- a/python/londiste/playback.py
+++ b/python/londiste/playback.py
@@ -651,7 +651,7 @@ class Replicator(CascadedWorker):
         self.current_event = None
 
     def handle_data_event(self, ev, dst_curs):
-        """handle one truncate event"""
+        """handle one data event"""
         t = self.get_table_by_name(ev.extra1)
         if not t or not t.interesting(ev, self.cur_tick, self.copy_thread, self.copy_table_name):
             self.stat_increase('ignored_events')
@@ -675,6 +675,16 @@ class Replicator(CascadedWorker):
 
         fqname = skytools.quote_fqident(t.dest_table)
 
+        try:
+            p = self.used_plugins[ev.extra1]
+        except KeyError:
+            p = t.get_plugin()
+            self.used_plugins[ev.extra1] = p
+
+        if p.conf.get('ignore_truncate'):
+            self.log.info("ignoring truncate for %s", fqname)
+            return
+
         #
         # Always use CASCADE, because without it the
         # operation cannot work with FKeys, on both