Merge pull request #16 from tinybirdco/fix-handling-null-values

Added changes to handle properly NULL and empty string values

Author: ygnuss

clickhouse_mysql/writer/csvwriter.py

@@ -136,7 +136,7 @@ def insert(self, event_or_events):
             if self.dst_table is None:
                 self.dst_table = event.table
 
-            self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.QUOTE_ALL)
+            self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.QUOTE_NONNUMERIC)
             if not self.header_written:
                 self.writer.writeheader()
 
@@ -190,7 +190,7 @@ def delete_row(self, event_or_events):
             if self.dst_table is None:
                 self.dst_table = event.table
 
-            self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.QUOTE_ALL)
+            self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.QUOTE_NONNUMERIC)
             if not self.header_written:
                 self.writer.writeheader()
 
@@ -253,7 +253,7 @@ def update(self, event_or_events):
             if self.dst_table is None:
                 self.dst_table = event.table
 
-            self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.csv.QUOTE_ALL)
+            self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.QUOTE_NONNUMERIC)
             if not self.header_written:
                 self.writer.writeheader()
 
@@ -266,28 +266,34 @@ def update(self, event_or_events):
             self.generate_row(event_converted)
 
 
+    def convert_null_values(self, row):
+        """ We need to mark those fields that are null to be able to distinguish between NULL and empty strings """
+        for key in list(row.keys()):
+            if row[key] is None:
+                row[key] = "NULL"
+
     def generate_row(self, event):
         """ When using mempool or csvpool events are cached so you can receive different kind of events in the same list. These events should be handled in a different way """
 
         if isinstance(event.pymysqlreplication_event, WriteRowsEvent):
             for row in event:
-                # row['tb_upd'] = event.ts.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
                 row['tb_upd'] = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
                 row['operation'] = 0
+                self.convert_null_values(row)
                 self.writer.writerow(self.convert(row))
         elif isinstance(event.pymysqlreplication_event, DeleteRowsEvent):
             for row in event:
-                # row['tb_upd'] = event.ts.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
                 row['tb_upd'] = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
                 row['operation'] = 2
+                self.convert_null_values(row)
                 self.writer.writerow(self.convert(row))
         elif isinstance(event.pymysqlreplication_event, UpdateRowsEvent):
             for row in event.pymysqlreplication_event.rows:
-                # row['after_values']['tb_upd'] = event.ts.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
                 row['after_values']['tb_upd'] = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
                 row['after_values']['operation'] = 1
+                self.convert_null_values(row['after_values'])
                 self.writer.writerow(self.convert(row['after_values']))
-
+    
 
     def push(self):
         if not self.next_writer_builder or not self.fieldnames:

clickhouse_mysql/writer/tbcsvwriter.py

@@ -3,6 +3,7 @@
 
 import logging
 import time
+import subprocess
 
 from clickhouse_mysql.writer.writer import Writer
 
@@ -50,6 +51,21 @@ def __init__(
         self.dst_table_prefix = dst_table_prefix
         self.dst_distribute = dst_distribute
 
+
+    def format_null_values(self, csv_file):
+        """ We need to replace NULL values by \\N (CH null value) and do not quote this field. 
+        It is the only way to distinguish between NULL and empty strings. With this we will have: 
+          - xx,\\N,yy --> For null values
+          - xx,'',yy --> For empty strings  """
+        
+        with open(csv_file, 'r') as file:
+            data = file.read()
+            data = data.replace('"NULL"', '\\N')    
+        
+        with open(csv_file, 'w') as file:
+            file.write(data)
+
+
     def uploadCSV(self, table, filename, tries=1):
         limit_of_retries = 3
         params = {
@@ -58,6 +74,9 @@ def uploadCSV(self, table, filename, tries=1):
         }
 
         try:
+            # Add replace NULL values by \N
+            self.format_null_values(filename)
+
             with open(filename, 'rb') as f:
                 m = MultipartEncoder(fields={'csv': ('csv', f, 'text/csv')})
                 url = f"{self.tb_host}/v0/datasources"