Add test for the Write a test for MySQL 5.7 column name support

Author: julien-duponchelle

README.md

@@ -364,6 +364,7 @@ Other contributors:
 - Oliver Seemann: Handle large json, github actions,
   Zero-pad fixed-length binary fields (https://github.com/oseemann)
 - Mahadir Ahmad: Handle null json payload (https://github.com/mahadirz)
+- Mehmet Kartalbas: Add MySQL 5.7 column name support (https://github.com/kartalbas)
 - Axel Viala: Removal of Python 2.7 (https://github.com/darnuria)
 - Etern: Add XAPrepareEvent, parse last_committed & sequence_number of GtidEvent (https://github.com/etern)
 - Jason Fulghum: typo in ident variable name (https://github.com/fulghum)

pymysqlreplication/tests/base.py

@@ -89,6 +89,12 @@ def isMySQL57(self):
         version = float(self.getMySQLVersion().rsplit(".", 1)[0])
         return version == 5.7
 
+    def isMySQL57AndMore(self):
+        if self.isMariaDB():
+            return False
+        version = float(self.getMySQLVersion().rsplit(".", 1)[0])
+        return version >= 5.7
+
     def isMySQL80AndMore(self):
         if self.isMariaDB():
             return False

pymysqlreplication/tests/test_basic.py

@@ -271,6 +271,75 @@ def test_write_row_event(self):
             self.assertEqual(event.rows[0]["values"]["data"], "Hello World")
             self.assertEqual(event.columns[1].name, "data")
 
+    def test_fetch_column_names_from_schema(self):
+        # This test is for MySQL 5.7+
+        if not self.isMySQL57AndMore():
+            self.skipTest("Test for MySQL 5.7+ where binlog_row_metadata can be MINIMAL")
+
+        self.execute("SET SESSION binlog_row_metadata = 'MINIMAL'")
+        query = "CREATE TABLE test_column_cache (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        self.execute("INSERT INTO test_column_cache (data) VALUES('Hello')")
+        self.execute("COMMIT")
+
+        # Test with use_column_name_cache = True
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            use_column_name_cache=True,
+            only_events=[WriteRowsEvent],
+        )
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, WriteRowsEvent)
+        self.assertEqual(event.table, "test_column_cache")
+        self.assertIn("id", event.rows[0]["values"])
+        self.assertIn("data", event.rows[0]["values"])
+        self.assertEqual(event.rows[0]["values"]["id"], 1)
+        self.assertEqual(event.rows[0]["values"]["data"], "Hello")
+
+        # Test with use_column_name_cache = False
+        self.stream.close()
+
+        # Clear cache before next run
+        from pymysqlreplication import row_event
+        row_event._COLUMN_NAME_CACHE.clear()
+
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1025, # different server_id to avoid caching issues
+            use_column_name_cache=False,
+            only_events=[WriteRowsEvent],
+        )
+
+        # Reset and replay events
+        self.resetBinLog()
+        self.execute("SET SESSION binlog_row_metadata = 'MINIMAL'")
+        self.execute("INSERT INTO test_column_cache (data) VALUES('World')")
+        self.execute("COMMIT")
+
+        # Skip RotateEvent and FormatDescriptionEvent
+        self.stream.fetchone()
+        self.stream.fetchone()
+        # Skip QueryEvent for BEGIN
+        if not self.isMariaDB():
+            self.stream.fetchone()
+        # Skip TableMapEvent
+        self.stream.fetchone()
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, WriteRowsEvent)
+        self.assertEqual(event.table, "test_column_cache")
+        # With cache disabled, we should not have column names
+        self.assertNotIn("id", event.rows[0]["values"])
+        self.assertNotIn("data", event.rows[0]["values"])
+
+        # cleanup
+        self.execute("SET SESSION binlog_row_metadata = 'FULL'")
+        row_event._COLUMN_NAME_CACHE.clear()
+
+
     def test_delete_row_event(self):
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)