Fix an issue in parsing default column charset

When parsing the default charset optional metadata, when handling the
column index/collation number pairs, the column index is in the range
of the number of columns with charset (not the number of all the
columns).

Author: YAtOff

pymysqlreplication/row_event.py

@@ -1005,14 +1005,17 @@ def _parsed_column_charset_by_default_charset(
         column_type_detect_function,
     ):
         column_charset = []
+        position = 0
         for i in range(self.column_count):
             column_type = self.columns[i].type
             if not column_type_detect_function(column_type, dbms=self.dbms):
                 continue
-            elif i not in column_charset_collation.keys():
-                column_charset.append(default_charset_collation)
             else:
-                column_charset.append(column_charset_collation[i])
+                if position not in column_charset_collation.keys():
+                    column_charset.append(default_charset_collation)
+                else:
+                    column_charset.append(column_charset_collation[position])
+                position += 1
 
         return column_charset
 

pymysqlreplication/tests/test_basic.py

@@ -1,6 +1,7 @@
 import io
 import time
 import unittest
+from random import randbytes, shuffle
 
 from pymysqlreplication.json_binary import JsonDiff, JsonDiffOperation
 from pymysqlreplication.tests import base
@@ -420,6 +421,44 @@ def test_minimal_image_update_row_event(self):
             self.assertEqual(event.rows[0]["after_values"]["id"], None)
             self.assertEqual(event.rows[0]["after_values"]["data"], "World")
 
+    def test_charset_parsing(self):
+        char_columns = {
+            f"c_utf8_{i}": ("VARCHAR(255) NOT NULL", "Hello")
+            for i in range(3)
+        }
+        char_columns["c_binary"] = ("LONGBLOB NOT NULL", randbytes(1024))
+        int_columns = {f"i_{i}": ("INTEGER NOT NULL", i) for i in range(3)}
+        columns = list({**char_columns, **int_columns}.items())
+        shuffle(columns)
+        column_types = ", ".join(f"{name} {type}" for name, (type, _) in columns)
+        column_names = ", ".join(name for name, _ in columns)
+        column_value_placeholders = ", ".join("%s" for _ in range(len(columns)))
+        column_values = [value for _, (_, value) in columns]
+        query = f"CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, {column_types}, PRIMARY KEY (id)) DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci;"
+        self.execute(query)
+        query = f"INSERT INTO test ({column_names}) VALUES({column_value_placeholders})"
+        self.execute_with_args(query, column_values)
+        self.execute("COMMIT")
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+        # QueryEvent for the Create Table
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+        # QueryEvent for the BEGIN
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, TableMapEvent)
+        if event.table_map[event.table_id].column_name_flag:
+            columns = {c.name: c for c in event.columns}
+            for name, column in columns.items():
+                if name.startswith("c_utf8_"):
+                    assert column.character_set_name == "utf8"
+                    assert column.collation_name == "utf8mb3_unicode_ci"
+            assert columns["c_binary"].character_set_name == "binary"
+            assert columns["c_binary"].collation_name == "binary"
+
+
     def test_log_pos(self):
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)