Add support for MariaDB compressed binlog events

Author: monder

canal/sync.go

@@ -253,11 +253,11 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 }
 var action string
 switch e.Header.EventType {
-case replication.WRITE_ROWS_EVENTv1, replication.WRITE_ROWS_EVENTv2:
+case replication.WRITE_ROWS_EVENTv1, replication.WRITE_ROWS_EVENTv2, replication.MARIADB_WRITE_ROWS_COMPRESSED_EVENT_V1:
 action = InsertAction
-case replication.DELETE_ROWS_EVENTv1, replication.DELETE_ROWS_EVENTv2:
+case replication.DELETE_ROWS_EVENTv1, replication.DELETE_ROWS_EVENTv2, replication.MARIADB_DELETE_ROWS_COMPRESSED_EVENT_V1:
 action = DeleteAction
-case replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2:
+case replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2, replication.MARIADB_UPDATE_ROWS_COMPRESSED_EVENT_V1:
 action = UpdateAction
 default:
 return errors.Errorf("%s not supported now", e.Header.EventType)

replication/const.go

@@ -101,6 +101,11 @@ const (
 MARIADB_BINLOG_CHECKPOINT_EVENT
 MARIADB_GTID_EVENT
 MARIADB_GTID_LIST_EVENT
+MARIADB_START_ENCRYPTION_EVENT
+MARIADB_QUERY_COMPRESSED_EVENT
+MARIADB_WRITE_ROWS_COMPRESSED_EVENT_V1
+MARIADB_UPDATE_ROWS_COMPRESSED_EVENT_V1
+MARIADB_DELETE_ROWS_COMPRESSED_EVENT_V1
 )
 
 func (e EventType) String() string {
@@ -197,6 +202,12 @@ func (e EventType) String() string {
 return "TransactionPayloadEvent"
 case HEARTBEAT_LOG_EVENT_V2:
 return "HeartbeatLogEventV2"
+case MARIADB_WRITE_ROWS_COMPRESSED_EVENT_V1:
+return "MariadbWriteRowsCompressedEventV1"
+case MARIADB_UPDATE_ROWS_COMPRESSED_EVENT_V1:
+return "MariadbUpdateRowsCompressedEventV1"
+case MARIADB_DELETE_ROWS_COMPRESSED_EVENT_V1:
+return "MariadbDeleteRowsCompressedEventV1"
 
 default:
 return "UnknownEvent"

replication/parser.go

@@ -270,7 +270,11 @@ func (p *BinlogParser) parseEvent(h *EventHeader, data []byte, rawData []byte) (
 WRITE_ROWS_EVENTv2,
 UPDATE_ROWS_EVENTv2,
 DELETE_ROWS_EVENTv2,
+MARIADB_WRITE_ROWS_COMPRESSED_EVENT_V1,
+MARIADB_UPDATE_ROWS_COMPRESSED_EVENT_V1,
+MARIADB_DELETE_ROWS_COMPRESSED_EVENT_V1,
 PARTIAL_UPDATE_ROWS_EVENT: // Extension of UPDATE_ROWS_EVENT, allowing partial values according to binlog_row_value_options
+
 e = p.newRowsEvent(h)
 case ROWS_QUERY_EVENT:
 e = &RowsQueryEvent{}
@@ -412,6 +416,16 @@ func (p *BinlogParser) newRowsEvent(h *EventHeader) *RowsEvent {
 case UPDATE_ROWS_EVENTv1:
 e.Version = 1
 e.needBitmap2 = true
+case MARIADB_WRITE_ROWS_COMPRESSED_EVENT_V1:
+e.Version = 1
+e.compressed = true
+case MARIADB_DELETE_ROWS_COMPRESSED_EVENT_V1:
+e.Version = 1
+e.compressed = true
+case MARIADB_UPDATE_ROWS_COMPRESSED_EVENT_V1:
+e.Version = 1
+e.compressed = true
+e.needBitmap2 = true
 case WRITE_ROWS_EVENTv2:
 e.Version = 2
 case UPDATE_ROWS_EVENTv2:

replication/row_event.go

@@ -1,6 +1,8 @@
 package replication
 
 import (
+"bytes"
+"compress/zlib"
 "encoding/binary"
 "encoding/hex"
 "fmt"
@@ -830,6 +832,9 @@ type RowsEvent struct {
 tables map[uint64]*TableMapEvent
 needBitmap2 bool
 
+// for mariadb *_COMPRESSED_EVENT_V1
+compressed bool
+
 eventType EventType
 
 Table *TableMapEvent
@@ -970,9 +975,9 @@ func (e *RowsEvent) DecodeData(pos int, data []byte) (err2 error) {
 
 var rowImageType EnumRowImageType
 switch e.eventType {
-case WRITE_ROWS_EVENTv0, WRITE_ROWS_EVENTv1, WRITE_ROWS_EVENTv2:
+case WRITE_ROWS_EVENTv0, WRITE_ROWS_EVENTv1, WRITE_ROWS_EVENTv2, MARIADB_WRITE_ROWS_COMPRESSED_EVENT_V1:
 rowImageType = EnumRowImageTypeWriteAI
-case DELETE_ROWS_EVENTv0, DELETE_ROWS_EVENTv1, DELETE_ROWS_EVENTv2:
+case DELETE_ROWS_EVENTv0, DELETE_ROWS_EVENTv1, DELETE_ROWS_EVENTv2, MARIADB_DELETE_ROWS_COMPRESSED_EVENT_V1:
 rowImageType = EnumRowImageTypeDeleteBI
 default:
 rowImageType = EnumRowImageTypeUpdateBI
@@ -1002,6 +1007,13 @@ func (e *RowsEvent) Decode(data []byte) error {
 if err != nil {
 return err
 }
+if e.compressed {
+uncompressedData, err := e.decompressData(pos, data)
+if err != nil {
+return err
+}
+return e.DecodeData(0, uncompressedData)
+}
 return e.DecodeData(pos, data)
 }
 
@@ -1103,6 +1115,28 @@ func (e *RowsEvent) parseFracTime(t interface{}) interface{} {
 return v.Time
 }
 
+func (e *RowsEvent) decompressData(pos int, data []byte) ([]byte, error) {
+// algorithm always 0=zlib
+// algorithm := (data[pos] & 0x07) >> 4
+headerSize := int(data[pos] & 0x07)
+pos++
+
+uncompressedDataSize := BFixedLengthInt(data[pos : pos+headerSize])
+
+pos += headerSize
+uncompressedData := make([]byte, uncompressedDataSize)
+r, err := zlib.NewReader(bytes.NewReader(data[pos:]))
+if err != nil {
+return nil, err
+}
+defer r.Close()
+_, err = io.ReadFull(r, uncompressedData)
+if err != nil {
+return nil, err
+}
+return uncompressedData, nil
+}
+
 // see mysql sql/log_event.cc log_event_print_value
 func (e *RowsEvent) decodeValue(data []byte, tp byte, meta uint16, isPartial bool) (v interface{}, n int, err error) {
 var length = 0