Merge pull request #112 from jasonzzz/master

add support for BeginLoadQueryEvent and ExecuteLoadQueryEvent, update tests

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -8,11 +8,13 @@
 from pymysql.util import int2byte
 
 from .packet import BinLogPacketWrapper
-from .constants.BINLOG import TABLE_MAP_EVENT, ROTATE_EVENT, STOP_EVENT
+from .constants.BINLOG import TABLE_MAP_EVENT, ROTATE_EVENT
 from .gtid import GtidSet
 from .event import (
     QueryEvent, RotateEvent, FormatDescriptionEvent,
-    XidEvent, GtidEvent, StopEvent, NotImplementedEvent)
+    XidEvent, GtidEvent, StopEvent,
+    BeginLoadQueryEvent, ExecuteLoadQueryEvent,
+    NotImplementedEvent)
 from .row_event import (
     UpdateRowsEvent, WriteRowsEvent, DeleteRowsEvent, TableMapEvent)
 
@@ -297,6 +299,8 @@ def _allowed_event_list(self, only_events, ignored_events,
                 FormatDescriptionEvent,
                 XidEvent,
                 GtidEvent,
+                BeginLoadQueryEvent,
+                ExecuteLoadQueryEvent,
                 UpdateRowsEvent,
                 WriteRowsEvent,
                 DeleteRowsEvent,

pymysqlreplication/event.py

@@ -147,6 +147,72 @@ def _dump(self):
         print("Query: %s" % (self.query))
 
 
+class BeginLoadQueryEvent(BinLogEvent):
+    """
+
+    Attributes:
+        file_id
+        block-data
+    """
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(BeginLoadQueryEvent, self).__init__(from_packet, event_size, table_map,
+                                                     ctl_connection, **kwargs)
+
+        # Payload
+        self.file_id = self.packet.read_uint32()
+        self.block_data = self.packet.read(event_size - 4)
+
+    def _dump(self):
+        super(BeginLoadQueryEvent, self)._dump()
+        print("File id: %d" % (self.file_id))
+        print("Block data: %s" % (self.block_data))
+
+
+class ExecuteLoadQueryEvent(BinLogEvent):
+    """
+
+    Attributes:
+        slave_proxy_id
+        execution_time
+        schema_length
+        error_code
+        status_vars_length
+
+        file_id
+        start_pos
+        end_pos
+        dup_handling_flags
+    """
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(ExecuteLoadQueryEvent, self).__init__(from_packet, event_size, table_map,
+                                                        ctl_connection, **kwargs)
+
+        # Post-header
+        self.slave_proxy_id = self.packet.read_uint32()
+        self.execution_time = self.packet.read_uint32()
+        self.schema_length = self.packet.read_uint8()
+        self.error_code = self.packet.read_uint16()
+        self.status_vars_length = self.packet.read_uint16()
+
+        # Payload
+        self.file_id = self.packet.read_uint32()
+        self.start_pos = self.packet.read_uint32()
+        self.end_pos = self.packet.read_uint32()
+        self.dup_handling_flags = self.packet.read_uint8()
+
+    def _dump(self):
+        super(ExecuteLoadQueryEvent, self)._dump()
+        print("Slave proxy id: %d" % (self.slave_proxy_id))
+        print("Execution time: %d" % (self.execution_time))
+        print("Schema length: %d" % (self.schema_length))
+        print("Error code: %d" % (self.error_code))
+        print("Status vars length: %d" % (self.status_vars_length))
+        print("File id: %d" % (self.file_id))
+        print("Start pos: %d" % (self.start_pos))
+        print("End pos: %d" % (self.end_pos))
+        print("Dup handling flags: %d" % (self.dup_handling_flags))
+
+
 class NotImplementedEvent(BinLogEvent):
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(NotImplementedEvent, self).__init__(

pymysqlreplication/packet.py

@@ -34,6 +34,8 @@ class BinLogPacketWrapper(object):
         constants.INTVAR_EVENT: event.NotImplementedEvent,
         constants.GTID_LOG_EVENT: event.GtidEvent,
         constants.STOP_EVENT: event.StopEvent,
+        constants.BEGIN_LOAD_QUERY_EVENT: event.BeginLoadQueryEvent,
+        constants.EXECUTE_LOAD_QUERY_EVENT: event.ExecuteLoadQueryEvent,
         # row_event
         constants.UPDATE_ROWS_EVENT_V1: row_event.UpdateRowsEvent,
         constants.WRITE_ROWS_EVENT_V1: row_event.WriteRowsEvent,

pymysqlreplication/tests/test_basic.py

@@ -14,9 +14,9 @@ def ignoredEvents(self):
         return [GtidEvent]
 
     def test_allowed_event_list(self):
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 11)
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 10)
-        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 10)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 13)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 12)
+        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 12)
         self.assertEqual(len(self.stream._allowed_event_list([RotateEvent], None, False)), 1)
 
     def test_read_query_event(self):
@@ -63,6 +63,32 @@ def test_reading_rotate_event(self):
         # Rotate event
         self.assertIsInstance(self.stream.fetchone(), RotateEvent)
 
+    """ `test_load_query_event` needs statement-based binlog
+    def test_load_query_event(self):
+        # prepare csv
+        with open("/tmp/test_load_query.csv", "w") as fp:
+            fp.write("1,aaa\n2,bbb\n3,ccc\n4,ddd\n")
+
+        query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "LOAD DATA INFILE '/tmp/test_load_query.csv' INTO TABLE test \
+                FIELDS TERMINATED BY ',' \
+                ENCLOSED BY '\"' \
+                LINES TERMINATED BY '\r\n'"
+        self.execute(query)
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+        # create table
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+        # begin
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+
+        self.assertIsInstance(self.stream.fetchone(), BeginLoadQueryEvent)
+        self.assertIsInstance(self.stream.fetchone(), ExecuteLoadQueryEvent)
+
+        self.assertIsInstance(self.stream.fetchone(), XidEvent)
+    """
 
     def test_connection_stream_lost_event(self):
         self.stream.close()