Merge pull request #23 from esakik/development

RELEASE 2023-12-30

Author: esakik

beam_mysql/connector/io.py

@@ -118,7 +118,7 @@ def __init__(
 
     def start_bundle(self):
         self._build_value()
-        self._queries = []
+        self._columns_and_values = dict()
 
     def process(self, element: Dict, *args, **kwargs):
         columns = []
@@ -137,18 +137,25 @@ def process(self, element: Dict, *args, **kwargs):
             ]
         )
 
-        query = f"INSERT INTO {self._config['database']}.{self._table}({column_str}) VALUES({value_str});"
+        if column_str not in self._columns_and_values:
+            self._columns_and_values[column_str] = []
 
-        self._queries.append(query)
+        self._columns_and_values[column_str].append(f"({value_str})")
 
-        if len(self._queries) > self._batch_size:
-            self._client.record_loader("\n".join(self._queries))
-            self._queries.clear()
+        if len(self._columns_and_values) > self._batch_size:
+            for column_str in self._columns_and_values.keys():
+                if len(self._columns_and_values[column_str]) > 0:
+                    self._client.record_loader(self._build_query(column_str, self._columns_and_values[column_str]))
+                    self._columns_and_values[column_str].clear()
 
     def finish_bundle(self):
-        if len(self._queries):
-            self._client.record_loader("\n".join(self._queries))
-            self._queries.clear()
+        for column_str in self._columns_and_values.keys():
+            if len(self._columns_and_values[column_str]) > 0:
+                self._client.record_loader(self._build_query(column_str, self._columns_and_values[column_str]))
+                self._columns_and_values[column_str].clear()
+
+    def _build_query(self, column_str, values_str):
+        return f"INSERT INTO {self._config['database']}.{self._table}({column_str}) VALUES {','.join(values_str)};"
 
     def _build_value(self):
         for k, v in self._config.items():