Merge remote-tracking branch 'origin/feature/optional-meta-data' into feature/optional-meta-data

Author: sean-k1

README.md

@@ -56,11 +56,14 @@ Limitations
 
 https://python-mysql-replication.readthedocs.org/en/latest/limitations.html
 
-Featured Books
+Featured
 =============
 
 [Data Pipelines Pocket Reference](https://www.oreilly.com/library/view/data-pipelines-pocket/9781492087823/) (by James Densmore, O'Reilly): Introduced and exemplified in Chapter 4: Data Ingestion: Extracting Data.
 
+[Streaming Changes in a Database with Amazon Kinesis](https://aws.amazon.com/blogs/database/streaming-changes-in-a-database-with-amazon-kinesis/) (by Emmanuel Espina, Amazon Web Services)
+
+
 Projects using this library
 ===========================
 
@@ -84,6 +87,7 @@ Projects using this library
 * MySQL to Kafka (https://github.com/scottpersinger/mysql-to-kafka/)
 * Aventri MySQL Monitor (https://github.com/aventri/mysql-monitor)
 * BitSwanPump: A real-time stream processor  (https://github.com/LibertyAces/BitSwanPump)
+* clickhouse-mysql-data-reader: https://github.com/Altinity/clickhouse-mysql-data-reader
 
 MySQL server settings
 =========================

docker-compose-test.yml

@@ -1,26 +1,56 @@
-version: '3.2'
+version: '3.4'
+
+x-mysql: &mysql
+  environment:
+    MYSQL_ALLOW_EMPTY_PASSWORD: true
+  command: >
+    mysqld
+    --log-bin=mysql-bin.log
+    --server-id 1
+    --binlog-format=row
+    --gtid_mode=on
+    --enforce-gtid-consistency=on
+
+x-mariadb: &mariadb
+  environment:
+    MARIADB_ALLOW_EMPTY_ROOT_PASSWORD: 1
+  command: >
+    --server-id=1
+    --default-authentication-plugin=mysql_native_password
+    --log-bin=master-bin
+    --binlog-format=row
+
 services:
-  percona-5.7:
-    platform: linux/amd64
+  percona-5.7-ctl:
+    <<: *mysql
     image: percona:5.7
-    environment:
-      MYSQL_ALLOW_EMPTY_PASSWORD: true
-      MYSQL_DATABASE: pymysqlreplication_test
     ports:
-      - 3306:3306
-    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates
-    restart: always
+      - "3307:3306"
     networks:
       - default
 
-  percona-5.7-ctl:
+  percona-5.7:
+    <<: *mysql
     image: percona:5.7
-    environment:
-      MYSQL_ALLOW_EMPTY_PASSWORD: true
-      MYSQL_DATABASE: pymysqlreplication_test
     ports:
-      - 3307:3307
-    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3307
+      - "3306:3306"
+    networks:
+      - default
+
+  mariadb-10.6:
+    <<: *mariadb
+    image: mariadb:10.6
+    ports:
+      - "3308:3306"
+    volumes:
+      - type: bind
+        source: ./.mariadb
+        target: /opt/key_file
+      - type: bind
+        source: ./.mariadb/my.cnf
+        target: /etc/mysql/my.cnf
+    networks:
+      - default
 
   pymysqlreplication:
     build:
@@ -30,6 +60,9 @@ services:
         BASE_IMAGE: python:3.11-alpine
         MYSQL_5_7: percona-5.7
         MYSQL_5_7_CTL: percona-5.7-ctl
+        MYSQL_5_7_CTL_PORT: 3306
+        MARIADB_10_6: mariadb-10.6
+        MARIADB_10_6_PORT: 3306
 
     command:
       - /bin/sh
@@ -39,7 +72,7 @@ services:
         
         while :
         do
-          if mysql -h percona-5.7 --user=root --execute "USE pymysqlreplication_test;" 2>&1 >/dev/null && mysql -h percona-5.7-ctl --port=3307  --user=root --execute "USE pymysqlreplication_test;" 2>&1 >/dev/null; then
+          if mysql -h percona-5.7 --user=root --execute "SELECT version();" 2>&1 >/dev/null && mysql -h percona-5.7-ctl --user=root --execute "SELECT version();" 2>&1 >/dev/null; then
             break
           fi
           sleep 1
@@ -56,4 +89,5 @@ services:
       - percona-5.7-ctl
 
 networks:
-  default: {}
+  default:
+    driver: bridge

docker-compose.yml

@@ -1,37 +1,47 @@
-version: '3.2'
+version: '3.4'
+
+x-mysql: &mysql
+  environment:
+    MYSQL_ALLOW_EMPTY_PASSWORD: true
+  command: >
+    mysqld
+    --log-bin=mysql-bin.log
+    --server-id 1
+    --binlog-format=row
+    --gtid_mode=on
+    --enforce-gtid-consistency=on
+
+x-mariadb: &mariadb
+  environment:
+    MARIADB_ALLOW_EMPTY_ROOT_PASSWORD: 1
+  command: >
+    --log-bin=master-bin
+    --server-id=1
+    --default-authentication-plugin=mysql_native_password
+    --binlog-format=row
+
 services:
   percona-5.7:
+    <<: *mysql
     image: percona:5.7
-    environment:
-      MYSQL_ALLOW_EMPTY_PASSWORD: true
     ports:
-      - 3306:3306
-    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates
+      - "3306:3306"
 
   percona-5.7-ctl:
+    <<: *mysql
     image: percona:5.7
-    environment:
-      MYSQL_ALLOW_EMPTY_PASSWORD: true
     ports:
-      - 3307:3307
-    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3307
+      - "3307:3306"
 
   mariadb-10.6:
+    <<: *mariadb
     image: mariadb:10.6
-    environment:
-      MARIADB_ALLOW_EMPTY_ROOT_PASSWORD: 1
     ports:
       - "3308:3306"
-    command: |
-      --server-id=1
-      --default-authentication-plugin=mysql_native_password
-      --log-bin=master-bin
-      --binlog-format=row
-      --log-slave-updates=on
     volumes:
       - type: bind
         source: ./.mariadb
         target: /opt/key_file
       - type: bind
         source: ./.mariadb/my.cnf
-        target: /etc/mysql/my.cnf  
+        target: /etc/mysql/my.cnf

examples/mariadb_gtid/read_event.py

@@ -1,7 +1,7 @@
 import pymysql
 
 from pymysqlreplication import BinLogStreamReader, gtid
-from pymysqlreplication.event import GtidEvent, RotateEvent, MariadbGtidEvent, QueryEvent,MariadbAnnotateRowsEvent
+from pymysqlreplication.event import GtidEvent, RotateEvent, MariadbGtidEvent, QueryEvent,MariadbAnnotateRowsEvent, MariadbBinLogCheckPointEvent
 from pymysqlreplication.row_event import WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent
 
 MARIADB_SETTINGS = {
@@ -62,6 +62,7 @@ def query_server_id(self):
         blocking=False,
         only_events=[
             MariadbGtidEvent,
+            MariadbBinLogCheckPointEvent,
             RotateEvent,
             WriteRowsEvent,
             UpdateRowsEvent,

pymysqlreplication/binlogstream.py

@@ -13,7 +13,8 @@
     XidEvent, GtidEvent, StopEvent, XAPrepareEvent,
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
     HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent,
-    MariadbAnnotateRowsEvent, RandEvent, MariadbStartEncryptionEvent, RowsQueryLogEvent)
+    MariadbAnnotateRowsEvent, RandEvent, MariadbStartEncryptionEvent, RowsQueryLogEvent,
+    MariadbGtidListEvent, MariadbBinLogCheckPointEvent)
 from .exceptions import BinLogNotEnabled
 from .gtid import GtidSet
 from .packet import BinLogPacketWrapper
@@ -627,6 +628,8 @@ def _allowed_event_list(self, only_events, ignored_events,
                 MariadbAnnotateRowsEvent,
                 RandEvent,
                 MariadbStartEncryptionEvent,
+                MariadbGtidListEvent,
+                MariadbBinLogCheckPointEvent
             ))
         if ignored_events is not None:
             for e in ignored_events:
@@ -654,9 +657,8 @@ def __get_table_information(self, schema, table):
                         information_schema.columns
                     WHERE
                         table_schema = %s AND table_name = %s
-                    ORDER BY ORDINAL_POSITION
                     """, (schema, table))
-                result = cur.fetchall()
+                result = sorted(cur.fetchall(), key=lambda x: x['ORDINAL_POSITION'])
                 cur.close()
 
                 return result

pymysqlreplication/constants/CHARSET.py

@@ -54,7 +54,7 @@ def by_name(self, name, dbms='mysql'):
 charset_by_id = charsets.by_id
 
 with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'charset_list.csv'), 'r') as f:
-    f.read()  # pass header
+    f.readline()  # pass header
     for line in f:
         lines = line.split(',')
         if len(lines) != 5:

pymysqlreplication/event.py

@@ -40,7 +40,7 @@ def _read_table_id(self):
 
     def dump(self):
         print("=== %s ===" % (self.__class__.__name__))
-        print("Date: %s" % (datetime.datetime.fromtimestamp(self.timestamp)
+        print("Date: %s" % (datetime.datetime.utcfromtimestamp(self.timestamp)
                             .isoformat()))
         print("Log position: %d" % self.packet.log_pos)
         print("Event size: %d" % (self.event_size))
@@ -111,6 +111,25 @@ def _dump(self):
         print("Flags:", self.flags)
         print('GTID:', self.gtid)
 
+class MariadbBinLogCheckPointEvent(BinLogEvent):
+    """
+    Represents a checkpoint in a binlog event in MariaDB.
+
+    More details are available in the MariaDB Knowledge Base:
+    https://mariadb.com/kb/en/binlog_checkpoint_event/
+
+    :ivar filename_length:  int - The length of the filename.
+    :ivar filename: str - The name of the file saved at the checkpoint.
+    """
+
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(MariadbBinLogCheckPointEvent, self).__init__(from_packet, event_size, table_map, ctl_connection,
+                                                           **kwargs)
+        filename_length = self.packet.read_uint32()
+        self.filename = self.packet.read(filename_length).decode()
+
+    def _dump(self):
+        print('Filename:', self.filename)
 
 class MariadbAnnotateRowsEvent(BinLogEvent):
     """
@@ -127,7 +146,41 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
 
     def _dump(self):
         super()._dump()
-        print("SQL statement :", self.sql_statement)   
+        print("SQL statement :", self.sql_statement)
+
+class MariadbGtidListEvent(BinLogEvent):
+    """
+    GTID List event
+    https://mariadb.com/kb/en/gtid_list_event/
+
+    Attributes:
+        gtid_length: Number of GTIDs
+        gtid_list: list of 'MariadbGtidObejct'
+
+        'MariadbGtidObejct' Attributes:
+            domain_id: Replication Domain ID
+            server_id: Server_ID
+            gtid_seq_no: GTID sequence
+            gtid: 'domain_id'+ 'server_id' + 'gtid_seq_no'
+    """
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+
+        super(MariadbGtidListEvent, self).__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
+
+        class MariadbGtidObejct(BinLogEvent):
+            """
+            Information class of elements in GTID list
+            """
+            def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+                super(MariadbGtidObejct, self).__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
+                self.domain_id = self.packet.read_uint32()
+                self.server_id = self.packet.read_uint32()
+                self.gtid_seq_no = self.packet.read_uint64()
+                self.gtid = "%d-%d-%d" % (self.domain_id, self.server_id, self.gtid_seq_no)
+
+
+        self.gtid_length = self.packet.read_uint32()
+        self.gtid_list = [MariadbGtidObejct(from_packet, event_size, table_map, ctl_connection, **kwargs) for i in range(self.gtid_length)]
 
 
 class RotateEvent(BinLogEvent):

pymysqlreplication/packet.py

@@ -88,8 +88,9 @@ class BinLogPacketWrapper(object):
         # MariaDB GTID
         constants.MARIADB_ANNOTATE_ROWS_EVENT: event.MariadbAnnotateRowsEvent,
         constants.MARIADB_BINLOG_CHECKPOINT_EVENT: event.NotImplementedEvent,
+        constants.MARIADB_BINLOG_CHECKPOINT_EVENT: event.MariadbBinLogCheckPointEvent,
         constants.MARIADB_GTID_EVENT: event.MariadbGtidEvent,
-        constants.MARIADB_GTID_GTID_LIST_EVENT: event.NotImplementedEvent,
+        constants.MARIADB_GTID_GTID_LIST_EVENT: event.MariadbGtidListEvent,
         constants.MARIADB_START_ENCRYPTION_EVENT: event.MariadbStartEncryptionEvent
     }