Add support for MARIADB_QUERY_COMPRESSED_EVENT

Author: monder

mysql/util.go

@@ -1,6 +1,8 @@
 package mysql
 
 import (
+"bytes"
+"compress/zlib"
 "crypto/rand"
 "crypto/rsa"
 "crypto/sha1"
@@ -92,6 +94,24 @@ func EncryptPassword(password string, seed []byte, pub *rsa.PublicKey) ([]byte,
 return rsa.EncryptOAEP(sha1v, rand.Reader, pub, plain, nil)
 }
 
+func DecompressMariadbData(data []byte) ([]byte, error) {
+// algorithm always 0=zlib
+// algorithm := (data[pos] & 0x07) >> 4
+headerSize := int(data[0] & 0x07)
+uncompressedDataSize := BFixedLengthInt(data[1 : 1+headerSize])
+uncompressedData := make([]byte, uncompressedDataSize)
+r, err := zlib.NewReader(bytes.NewReader(data[1+headerSize:]))
+if err != nil {
+return nil, err
+}
+defer r.Close()
+_, err = io.ReadFull(r, uncompressedData)
+if err != nil {
+return nil, err
+}
+return uncompressedData, nil
+}
+
 // AppendLengthEncodedInteger: encodes a uint64 value and appends it to the given bytes slice
 func AppendLengthEncodedInteger(b []byte, n uint64) []byte {
 switch {

replication/const.go

@@ -202,6 +202,10 @@ func (e EventType) String() string {
 return "TransactionPayloadEvent"
 case HEARTBEAT_LOG_EVENT_V2:
 return "HeartbeatLogEventV2"
+case MARIADB_START_ENCRYPTION_EVENT:
+return "MariadbStartEncryptionEvent"
+case MARIADB_QUERY_COMPRESSED_EVENT:
+return "MariadbQueryCompressedEvent"
 case MARIADB_WRITE_ROWS_COMPRESSED_EVENT_V1:
 return "MariadbWriteRowsCompressedEventV1"
 case MARIADB_UPDATE_ROWS_COMPRESSED_EVENT_V1:

replication/event.go

@@ -297,6 +297,9 @@ type QueryEvent struct {
 Schema []byte
 Query []byte
 
+// for mariadb QUERY_COMPRESSED_EVENT
+compressed bool
+
 // in fact QueryEvent dosen't have the GTIDSet information, just for beneficial to use
 GSet GTIDSet
 }
@@ -328,7 +331,15 @@ func (e *QueryEvent) Decode(data []byte) error {
 //skip 0x00
 pos++
 
-e.Query = data[pos:]
+if e.compressed {
+decompressedQuery, err := DecompressMariadbData(data[pos:])
+if err != nil {
+return err
+}
+e.Query = decompressedQuery
+} else {
+e.Query = data[pos:]
+}
 return nil
 }
 

replication/parser.go

@@ -249,6 +249,10 @@ func (p *BinlogParser) parseEvent(h *EventHeader, data []byte, rawData []byte) (
 switch h.EventType {
 case QUERY_EVENT:
 e = &QueryEvent{}
+case MARIADB_QUERY_COMPRESSED_EVENT:
+e = &QueryEvent{
+compressed: true,
+}
 case XID_EVENT:
 e = &XIDEvent{}
 case TABLE_MAP_EVENT:

replication/row_event.go

@@ -1,8 +1,6 @@
 package replication
 
 import (
-"bytes"
-"compress/zlib"
 "encoding/binary"
 "encoding/hex"
 "fmt"
@@ -1008,7 +1006,7 @@ func (e *RowsEvent) Decode(data []byte) error {
 return err
 }
 if e.compressed {
-uncompressedData, err := e.decompressData(pos, data)
+uncompressedData, err := DecompressMariadbData(data[pos:])
 if err != nil {
 return err
 }
@@ -1115,28 +1113,6 @@ 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