Added support to insert/update and CSV

Author: alexon1234

.gitignore

@@ -46,4 +46,5 @@ _build
 
 # Tinibird
 bl-*
-out-*
+out-*
+.e

clickhouse_mysql/clioptions.py

@@ -93,6 +93,10 @@ class CLIOptions(Options):
         #
         # general app section
         #
+
+        'tb_host': 'https://ui.tinybird.co',
+        'tb_token': None,
+
         'config_file': '/etc/clickhouse-mysql/clickhouse-mysql.conf',
         'log_file': None,
         'log_level': None,
@@ -171,6 +175,20 @@ def options(self):
         #
         # general app section
         #
+        argparser.add_argument(
+            '--tb-host',
+            type=str,
+            default=self.default_options['tb_host'],
+            help='Tinybird host'
+        )
+
+        argparser.add_argument(
+            '--tb-token',
+            type=str,
+            default=self.default_options['tb_token'],
+            help='Tinybird host'
+        )
+
         argparser.add_argument(
             '--config-file',
             type=str,
@@ -508,6 +526,11 @@ def options(self):
             #
             # general app section
             #
+
+            'tb_host': args.tb_host,
+            'tb_token': args.tb_token,
+
+
             'config_file': args.config_file,
             'log_file': args.log_file,
             'log_level': args.log_level,

clickhouse_mysql/config.py

@@ -6,7 +6,7 @@
 
 from clickhouse_mysql.writer.chwriter import CHWriter
 from clickhouse_mysql.writer.csvwriter import CSVWriter
-from clickhouse_mysql.writer.chcsvwriter import CHCSVWriter
+from clickhouse_mysql.writer.tbcsvwriter import TBCSVWriter
 from clickhouse_mysql.writer.poolwriter import PoolWriter
 from clickhouse_mysql.writer.processwriter import ProcessWriter
 from clickhouse_mysql.objectbuilder import ObjectBuilder
@@ -61,6 +61,10 @@ def __init__(self):
             #
             #
             #
+            'tinybird': {
+                'host': self.options['tb_host'],
+                'token': self.options['tb_token'],
+            },
             'app': {
                 'config_file': self.options['config_file'],
                 'log_file': self.options['log_file'],
@@ -359,8 +363,11 @@ def writer_builder_csvpool(self):
                 'dst_table': self.config['writer']['file']['dst_table'],
                 'dst_table_prefix': self.config['writer']['file']['dst_table_prefix'],
                 'next_writer_builder': ObjectBuilder(
-                    class_name=CHCSVWriter,
-                    constructor_params=self.config['writer']['clickhouse']
+                    class_name=TBCSVWriter,
+                    constructor_params={
+                        'tb_host': self.config['tinybird']['host'],
+                        'tb_token': self.config['tinybird']['token']
+                    }
                 ),
                 'converter_builder': self.converter_builder(CONVERTER_CSV),
             })

clickhouse_mysql/event/event.py

@@ -64,7 +64,11 @@ def __next__(self):
 
         if self.pymysqlreplication_event is not None:
             # in native replication event actual data are in row['values'] dict item
-            return item['values']
+            if 'after_values' in item:
+                return item['after_values']
+            else:
+                return item['values']
+            
         else:
             # local-kept data
             return item

clickhouse_mysql/reader/mysqlreader.py

@@ -316,7 +316,7 @@ def process_update_rows_event(self, mysql_event):
                 return
 
         # statistics
-        #self.stat_write_rows_event_calc_rows_num_min_max(rows_num_per_event=len(mysql_event.rows))
+        self.stat_write_rows_event_calc_rows_num_min_max(rows_num_per_event=len(mysql_event.rows))
 
         if self.subscribers('UpdateRowsEvent'):
             # dispatch event to subscribers
@@ -330,7 +330,7 @@ def process_update_rows_event(self, mysql_event):
             event.table = mysql_event.table
             event.pymysqlreplication_event = mysql_event
 
-            #self.process_first_event(event=event)
+            self.process_first_event(event=event)
             self.notify('UpdateRowsEvent', event=event)
 
         # self.stat_write_rows_event_finalyse()