add binlog row minimal and noblob image support

pymysqlreplication/bitmap.py

@@ -0,0 +1,83 @@
+# -*- coding: utf-8 -*-
+
+import struct
+
+class Bitmap(object):
+    """ Use for current-present-bitmap1/2 in RowsEvent, copy from MySQL mysys/my_bitmap.c
+    See http://dev.mysql.com/doc/internals/en/rows-event.html
+
+        null bitmap length = (bits set in 'columns-present-bitmap1'+7)/8
+        null bitmap2 length = (bits set in 'columns-present-bitmap2'+7)/8
+
+    We can not calculate the length using column count directly in minimal or noblob binlog row image.  
+
+    """ 
+
+    bits2Nbits = [
+        0, 1, 1, 2, 1, 2, 2, 3, 1, 2, 2, 3, 2, 3, 3, 4,
+        1, 2, 2, 3, 2, 3, 3, 4, 2, 3, 3, 4, 3, 4, 4, 5,
+        1, 2, 2, 3, 2, 3, 3, 4, 2, 3, 3, 4, 3, 4, 4, 5,
+        2, 3, 3, 4, 3, 4, 4, 5, 3, 4, 4, 5, 4, 5, 5, 6,
+        1, 2, 2, 3, 2, 3, 3, 4, 2, 3, 3, 4, 3, 4, 4, 5,
+        2, 3, 3, 4, 3, 4, 4, 5, 3, 4, 4, 5, 4, 5, 5, 6,
+        2, 3, 3, 4, 3, 4, 4, 5, 3, 4, 4, 5, 4, 5, 5, 6,
+        3, 4, 4, 5, 4, 5, 5, 6, 4, 5, 5, 6, 5, 6, 6, 7,
+        1, 2, 2, 3, 2, 3, 3, 4, 2, 3, 3, 4, 3, 4, 4, 5,
+        2, 3, 3, 4, 3, 4, 4, 5, 3, 4, 4, 5, 4, 5, 5, 6,
+        2, 3, 3, 4, 3, 4, 4, 5, 3, 4, 4, 5, 4, 5, 5, 6,
+        3, 4, 4, 5, 4, 5, 5, 6, 4, 5, 5, 6, 5, 6, 6, 7,
+        2, 3, 3, 4, 3, 4, 4, 5, 3, 4, 4, 5, 4, 5, 5, 6,
+        3, 4, 4, 5, 4, 5, 5, 6, 4, 5, 5, 6, 5, 6, 6, 7,
+        3, 4, 4, 5, 4, 5, 5, 6, 4, 5, 5, 6, 5, 6, 6, 7,
+        4, 5, 5, 6, 5, 6, 6, 7, 5, 6, 6, 7, 6, 7, 7, 8,
+    ]
+
+    def __init__(self, bitmap, nbits):
+        self._nbits = int(nbits)   
+        self._bitmap = bytearray(b'\x00' * int((self._nbits + 31) / 32) * 4)
+        self._bitmap[0:len(bitmap)] = bitmap
+        self._lastWordMask = 0
+        self._createLastWordMask()
+
+    def _countBitsUint32(self, v):
+        s = bytearray(struct.pack("<I", v))
+        return self.bits2Nbits[s[0]] + self.bits2Nbits[s[1]] + self.bits2Nbits[s[2]] + self.bits2Nbits[s[3]]
+
+    def _createLastWordMask(self):
+        used = 1 + (self._nbits - 1) & 0x7
+        mask = (~(1 << used) - 1) & 0xFF
+
+        b = bytearray(0)
+        l = int((self._nbits + 7) / 8) & 0x3
+        if l == 1:
+            self._lastWordMask = ~0 & 0xFFFFFFFF
+            b = bytearray(struct.pack("<I", self._lastWordMask))
+            b[0] = mask
+        elif l == 2:
+            self._lastWordMask = ~0 & 0xFFFFFFFF
+            b = bytearray(struct.pack("<I", self._lastWordMask))
+            b[0] = 0
+            b[1] = mask
+        elif l == 3:
+            self._lastWordMask = 0
+            b = bytearray(struct.pack("<I", self._lastWordMask))
+            b[2] = mask
+            b[3] = 0xFF
+        elif l == 0:
+            self._lastWordMask = 0
+            b = bytearray(struct.pack("<I", self._lastWordMask))
+            b[3] = mask
+
+        self._lastWordMask = struct.unpack("<I", bytes(b))[0]
+
+    def bits_set(self):
+        res = 0
+        for i in range(0, int(len(self._bitmap) / 4) - 1):
+            res += self._countBitsUint32(struct.unpack("<I", bytes(self._bitmap[i*4:i*4 + 4]))[0])
+
+        res += self._countBitsUint32(struct.unpack("<I", bytes(self._bitmap[-4:]))[0] & ~self._lastWordMask)
+
+        return res
+
+    def is_set(self, bit):
+        return self._bitmap[int(bit / 8)] & (1 << (bit & 7))

pymysqlreplication/row_event.py

@@ -11,6 +11,7 @@
 from .constants import BINLOG
 from .column import Column
 from .table import Table
+from .bitmap import Bitmap
 
 
 class RowsEvent(BinLogEvent):
@@ -60,18 +61,26 @@ def __is_null(self, null_bitmap, position):
             bit = ord(bit)
         return bit & (1 << (position % 8))
 
-    def _read_column_data(self, null_bitmap):
+    def _read_column_data(self, bitmap):
         """Use for WRITE, UPDATE and DELETE events.
         Return an array of column data
         """
         values = {}
 
+        null_bitmap = self.packet.read((bitmap.bits_set() + 7) / 8)
+
+        nullBitmapIndex = 0
         nb_columns = len(self.columns)
         for i in range(0, nb_columns):
             column = self.columns[i]
             name = self.table_map[self.table_id].columns[i].name
             unsigned = self.table_map[self.table_id].columns[i].unsigned
-            if self.__is_null(null_bitmap, i):
+
+            if bitmap.is_set(i) == 0:
+                values[name] = None
+                continue
+
+            if self.__is_null(null_bitmap, nullBitmapIndex):
                 values[name] = None
             elif column.type == FIELD_TYPE.TINY:
                 if unsigned:
@@ -153,6 +162,9 @@ def _read_column_data(self, null_bitmap):
             else:
                 raise NotImplementedError("Unknown MySQL column type: %d" %
                                           (column.type))
+
+            nullBitmapIndex += 1
+
         return values
 
     def __add_fsp_to_time(self, time, column):
@@ -394,8 +406,9 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
     def _fetch_one_row(self):
         row = {}
 
-        null_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
-        row["values"] = self._read_column_data(null_bitmap)
+        b = Bitmap(self.columns_present_bitmap, self.number_of_columns)
+        #null_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
+        row["values"] = self._read_column_data(b)
         return row
 
     def _dump(self):
@@ -423,8 +436,9 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
     def _fetch_one_row(self):
         row = {}
 
-        null_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
-        row["values"] = self._read_column_data(null_bitmap)
+        b = Bitmap(self.columns_present_bitmap, self.number_of_columns)
+#        null_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
+        row["values"] = self._read_column_data(b)
         return row
 
     def _dump(self):
@@ -459,12 +473,14 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
 
     def _fetch_one_row(self):
         row = {}
-        null_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
 
-        row["before_values"] = self._read_column_data(null_bitmap)
+        b = Bitmap(self.columns_present_bitmap, self.number_of_columns)
+        #null_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
+        row["before_values"] = self._read_column_data(b)
 
-        null_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
-        row["after_values"] = self._read_column_data(null_bitmap)
+        b = Bitmap(self.columns_present_bitmap2, self.number_of_columns)
+        #null_bitmap = self.packet.read((self.number_of_columns + 7) / 8)
+        row["after_values"] = self._read_column_data(b)
         return row
 
     def _dump(self):