Merge branch 'master' into mysql8-auth

Author: atercattus

canal/canal.go

@@ -450,6 +450,7 @@ func (c *Canal) prepareSyncer() error {
 TLSConfig: c.cfg.TLSConfig,
 Logger: c.cfg.Logger,
 Dialer: c.cfg.Dialer,
+Localhost: c.cfg.Localhost,
 }
 
 if strings.Contains(c.cfg.Addr, "/") {

canal/config.go

@@ -96,6 +96,9 @@ type Config struct {
 
 //Set Dialer
 Dialer client.Dialer
+
+//Set Localhost
+Localhost string
 }
 
 func NewConfigWithFile(name string) (*Config, error) {

client/resp.go

@@ -6,6 +6,7 @@ import (
 "crypto/x509"
 "encoding/binary"
 "encoding/pem"
+"fmt"
 
 "github.com/pingcap/errors"
 "github.com/siddontang/go/hack"
@@ -60,9 +61,9 @@ func (c *Conn) handleOKPacket(data []byte) (*Result, error) {
 // pos += 2
 }
 
-//new ok package will check CLIENT_SESSION_TRACK too, but I don't support it now.
+// new ok package will check CLIENT_SESSION_TRACK too, but I don't support it now.
 
-//skip info
+// skip info
 return r, nil
 }
 
@@ -75,7 +76,7 @@ func (c *Conn) handleErrorPacket(data []byte) error {
 pos += 2
 
 if c.capability&CLIENT_PROTOCOL_41 > 0 {
-//skip '#'
+// skip '#'
 pos++
 e.State = hack.String(data[pos : pos+5])
 pos += 5
@@ -89,11 +90,11 @@ func (c *Conn) handleErrorPacket(data []byte) error {
 func (c *Conn) handleAuthResult() error {
 data, switchToPlugin, err := c.readAuthResult()
 if err != nil {
-return err
+return fmt.Errorf("readAuthResult: %w", err)
 }
 // handle auth switch, only support 'sha256_password', and 'caching_sha2_password'
 if switchToPlugin != "" {
-//fmt.Printf("now switching auth plugin to '%s'\n", switchToPlugin)
+// fmt.Printf("now switching auth plugin to '%s'\n", switchToPlugin)
 if data == nil {
 data = c.salt
 } else {
@@ -168,7 +169,7 @@ func (c *Conn) handleAuthResult() error {
 func (c *Conn) readAuthResult() ([]byte, string, error) {
 data, err := c.ReadPacket()
 if err != nil {
-return nil, "", err
+return nil, "", fmt.Errorf("ReadPacket: %w", err)
 }
 
 // see: https://insidemysql.com/preparing-your-community-connector-for-mysql-8-part-2-sha256/
@@ -351,7 +352,7 @@ func (c *Conn) readResultColumns(result *Result) (err error) {
 if c.isEOFPacket(data) {
 if c.capability&CLIENT_PROTOCOL_41 > 0 {
 result.Warnings = binary.LittleEndian.Uint16(data[1:])
-//todo add strict_mode, warning will be treat as error
+// todo add strict_mode, warning will be treat as error
 result.Status = binary.LittleEndian.Uint16(data[3:])
 c.status = result.Status
 }
@@ -392,7 +393,7 @@ func (c *Conn) readResultRows(result *Result, isBinary bool) (err error) {
 if c.isEOFPacket(data) {
 if c.capability&CLIENT_PROTOCOL_41 > 0 {
 result.Warnings = binary.LittleEndian.Uint16(data[1:])
-//todo add strict_mode, warning will be treat as error
+// todo add strict_mode, warning will be treat as error
 result.Status = binary.LittleEndian.Uint16(data[3:])
 c.status = result.Status
 }

go.mod

@@ -17,6 +17,5 @@ require (
 github.com/siddontang/go v0.0.0-20180604090527-bdc77568d726
 github.com/siddontang/go-log v0.0.0-20180807004314-8d05993dda07
 github.com/stretchr/testify v1.8.0
-golang.org/x/mod v0.3.0
 golang.org/x/tools v0.0.0-20201125231158-b5590deeca9b // indirect
 )

go.sum

@@ -77,7 +77,6 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/exp v0.0.0-20181106170214-d68db9428509/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
-golang.org/x/mod v0.3.0 h1:RM4zey1++hCTbCVQfnWeKs9/IEsaBLA8vTkd0WVtmH4=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=

mysql/util_test.go

@@ -0,0 +1,30 @@
+package mysql
+
+import (
+"github.com/pingcap/check"
+)
+
+type utilTestSuite struct {
+}
+
+var _ = check.Suite(&utilTestSuite{})
+
+func (s *utilTestSuite) TestCompareServerVersions(c *check.C) {
+tests := []struct {
+A string
+B string
+Expect int
+}{
+{A: "1.2.3", B: "1.2.3", Expect: 0},
+{A: "5.6-999", B: "8.0", Expect: -1},
+{A: "8.0.32-0ubuntu0.20.04.2", B: "8.0.28", Expect: 1},
+}
+
+for _, test := range tests {
+comment := check.Commentf("%q vs. %q", test.A, test.B)
+
+got, err := CompareServerVersions(test.A, test.B)
+c.Assert(err, check.IsNil, comment)
+c.Assert(got, check.Equals, test.Expect, comment)
+}
+}

replication/json_binary.go

@@ -5,9 +5,10 @@ import (
 "fmt"
 "math"
 
-. "github.com/go-mysql-org/go-mysql/mysql"
 "github.com/pingcap/errors"
 "github.com/siddontang/go/hack"
+
+. "github.com/go-mysql-org/go-mysql/mysql"
 )
 
 const (
@@ -44,6 +45,60 @@ const (
 jsonbValueEntrySizeLarge = 1 + jsonbLargeOffsetSize
 )
 
+var (
+ErrCorruptedJSONDiff = fmt.Errorf("corrupted JSON diff") // ER_CORRUPTED_JSON_DIFF
+)
+
+type (
+// JsonDiffOperation is an enum that describes what kind of operation a JsonDiff object represents.
+// https://github.com/mysql/mysql-server/blob/8.0/sql/json_diff.h
+JsonDiffOperation byte
+)
+
+const (
+// The JSON value in the given path is replaced with a new value.
+//
+// It has the same effect as `JSON_REPLACE(col, path, value)`.
+JsonDiffOperationReplace = JsonDiffOperation(iota)
+
+// Add a new element at the given path.
+//
+// If the path specifies an array element, it has the same effect as `JSON_ARRAY_INSERT(col, path, value)`.
+//
+// If the path specifies an object member, it has the same effect as `JSON_INSERT(col, path, value)`.
+JsonDiffOperationInsert
+
+// The JSON value at the given path is removed from an array or object.
+//
+// It has the same effect as `JSON_REMOVE(col, path)`.
+JsonDiffOperationRemove
+)
+
+type (
+JsonDiff struct {
+Op JsonDiffOperation
+Path string
+Value string
+}
+)
+
+func (op JsonDiffOperation) String() string {
+switch op {
+case JsonDiffOperationReplace:
+return "Replace"
+case JsonDiffOperationInsert:
+return "Insert"
+case JsonDiffOperationRemove:
+return "Remove"
+default:
+return fmt.Sprintf("Unknown(%d)", op)
+}
+}
+
+func (jd *JsonDiff) String() string {
+return fmt.Sprintf("json_diff(op:%s path:%s value:%s)", jd.Op, jd.Path, jd.Value)
+}
+
 func jsonbGetOffsetSize(isSmall bool) int {
 if isSmall {
 return jsonbSmallOffsetSize
@@ -71,11 +126,6 @@ func jsonbGetValueEntrySize(isSmall bool) int {
 // decodeJsonBinary decodes the JSON binary encoding data and returns
 // the common JSON encoding data.
 func (e *RowsEvent) decodeJsonBinary(data []byte) ([]byte, error) {
-// Sometimes, we can insert a NULL JSON even we set the JSON field as NOT NULL.
-// If we meet this case, we can return an empty slice.
-if len(data) == 0 {
-return []byte{}, nil
-}
 d := jsonBinaryDecoder{
 useDecimal: e.useDecimal,
 ignoreDecodeErr: e.ignoreJSONDecodeErr,
@@ -491,3 +541,43 @@ func (d *jsonBinaryDecoder) decodeVariableLength(data []byte) (int, int) {
 
 return 0, 0
 }
+
+func (e *RowsEvent) decodeJsonPartialBinary(data []byte) (*JsonDiff, error) {
+// see Json_diff_vector::read_binary() in mysql-server/sql/json_diff.cc
+operationNumber := JsonDiffOperation(data[0])
+switch operationNumber {
+case JsonDiffOperationReplace:
+case JsonDiffOperationInsert:
+case JsonDiffOperationRemove:
+default:
+return nil, ErrCorruptedJSONDiff
+}
+data = data[1:]
+
+pathLength, _, n := LengthEncodedInt(data)
+data = data[n:]
+
+path := data[:pathLength]
+data = data[pathLength:]
+
+diff := &JsonDiff{
+Op: operationNumber,
+Path: string(path),
+// Value will be filled below
+}
+
+if operationNumber == JsonDiffOperationRemove {
+return diff, nil
+}
+
+valueLength, _, n := LengthEncodedInt(data)
+data = data[n:]
+
+d, err := e.decodeJsonBinary(data[:valueLength])
+if err != nil {
+return nil, fmt.Errorf("cannot read json diff for field %q: %w", path, err)
+}
+diff.Value = string(d)
+
+return diff, nil
+}

replication/parser.go

@@ -269,7 +269,8 @@ func (p *BinlogParser) parseEvent(h *EventHeader, data []byte, rawData []byte) (
 UPDATE_ROWS_EVENTv1,
 WRITE_ROWS_EVENTv2,
 UPDATE_ROWS_EVENTv2,
-DELETE_ROWS_EVENTv2:
+DELETE_ROWS_EVENTv2,
+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{}
@@ -381,14 +382,17 @@ func (p *BinlogParser) verifyCrc32Checksum(rawData []byte) error {
 
 func (p *BinlogParser) newRowsEvent(h *EventHeader) *RowsEvent {
 e := &RowsEvent{}
-if p.format.EventTypeHeaderLengths[h.EventType-1] == 6 {
+
+postHeaderLen := p.format.EventTypeHeaderLengths[h.EventType-1]
+if postHeaderLen == 6 {
 e.tableIDSize = 4
 } else {
 e.tableIDSize = 6
 }
 
 e.needBitmap2 = false
 e.tables = p.tables
+e.eventType = h.EventType
 e.parseTime = p.parseTime
 e.timestampStringLocation = p.timestampStringLocation
 e.useDecimal = p.useDecimal
@@ -415,6 +419,9 @@ func (p *BinlogParser) newRowsEvent(h *EventHeader) *RowsEvent {
 e.needBitmap2 = true
 case DELETE_ROWS_EVENTv2:
 e.Version = 2
+case PARTIAL_UPDATE_ROWS_EVENT:
+e.Version = 2
+e.needBitmap2 = true
 }
 
 return e

replication/parser_test.go

@@ -14,7 +14,7 @@ func (t *testSyncerSuite) TestIndexOutOfRange(c *C) {
 ServerVersion: []uint8{0x35, 0x2e, 0x36, 0x2e, 0x32, 0x30, 0x2d, 0x6c, 0x6f, 0x67, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
 CreateTimestamp: 0x0,
 EventHeaderLength: 0x13,
-EventTypeHeaderLengths: []uint8{0x38, 0xd, 0x0, 0x8, 0x0, 0x12, 0x0, 0x4, 0x4, 0x4, 0x4, 0x12, 0x0, 0x0, 0x5c, 0x0, 0x4, 0x1a, 0x8, 0x0, 0x0, 0x0, 0x8, 0x8, 0x8, 0x2, 0x0, 0x0, 0x0, 0xa, 0xa, 0xa, 0x19, 0x19, 0x0},
+EventTypeHeaderLengths: []uint8{0x38, 0xd, 0x0, 0x8, 0x0, 0x12, 0x0, 0x4, 0x4, 0x4, 0x4, 0x12, 0x0, 0x0, 0x5c, 0x0, 0x4, 0x1a, 0x8, 0x0, 0x0, 0x0, 0x8, 0x8, 0x8, 0x2, 0x0, 0x0, 0x0, 0xa, 0xa, 0xa, 0x19, 0x19, 0x0, 0x12, 0x34, 0x0, 0xa, 0x28, 0x0},
 ChecksumAlgorithm: 0x1,
 }
 
@@ -50,7 +50,7 @@ func (t *testSyncerSuite) TestParseEvent(c *C) {
 ServerVersion: []uint8{0x35, 0x2e, 0x36, 0x2e, 0x32, 0x30, 0x2d, 0x6c, 0x6f, 0x67, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
 CreateTimestamp: 0x0,
 EventHeaderLength: 0x13,
-EventTypeHeaderLengths: []uint8{0x38, 0xd, 0x0, 0x8, 0x0, 0x12, 0x0, 0x4, 0x4, 0x4, 0x4, 0x12, 0x0, 0x0, 0x5c, 0x0, 0x4, 0x1a, 0x8, 0x0, 0x0, 0x0, 0x8, 0x8, 0x8, 0x2, 0x0, 0x0, 0x0, 0xa, 0xa, 0xa, 0x19, 0x19, 0x0},
+EventTypeHeaderLengths: []uint8{0x38, 0xd, 0x0, 0x8, 0x0, 0x12, 0x0, 0x4, 0x4, 0x4, 0x4, 0x12, 0x0, 0x0, 0x5c, 0x0, 0x4, 0x1a, 0x8, 0x0, 0x0, 0x0, 0x8, 0x8, 0x8, 0x2, 0x0, 0x0, 0x0, 0xa, 0xa, 0xa, 0x19, 0x19, 0x0, 0x12, 0x34, 0x0, 0xa, 0x28, 0x0},
 ChecksumAlgorithm: 0x0,
 }
 testCases := []struct {
@@ -108,3 +108,34 @@ func (t *testSyncerSuite) TestRowsEventDecodeFunc(c *C) {
 c.Assert(e.Header.EventSize, Equals, tc.eventSize)
 }
 }
+
+func (t *testSyncerSuite) TestRowsEventDecodeImageWithEmptyJSON(c *C) {
+data := []byte("\x01\a\x00\xf6+\x0f\x00\xeb\xafP\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x99\xac\xfa\xbeÙ\xaf\xab1\x184\x11\x00\x00")
+
+bitmap := []byte{255}
+
+table := TableMapEvent{
+ColumnType: []byte{3, 3, 245, 245, 245, 18, 18, 3},
+ColumnMeta: []uint16{0, 0, 4, 4, 4, 0, 0, 0},
+}
+
+e := RowsEvent{
+eventType: PARTIAL_UPDATE_ROWS_EVENT,
+Table: &table,
+ColumnCount: uint64(len(table.ColumnType)),
+}
+n, err := e.decodeImage(data, bitmap, EnumRowImageTypeUpdateAI)
+c.Assert(err, IsNil)
+c.Assert(n, Equals, len(data))
+
+c.Assert(len(e.Rows), Equals, 1)
+c.Assert(len(e.Rows[0]), Equals, len(table.ColumnType))
+
+row := e.Rows[0]
+c.Assert(row[0], Equals, int32(994294))
+c.Assert(row[1], Equals, int32(38842347))
+c.Assert(row[2], DeepEquals, []byte{}) // empty json
+c.Assert(row[3], DeepEquals, []byte{}) // empty json
+c.Assert(row[4], DeepEquals, []byte{}) // empty json
+c.Assert(row[7], Equals, int32(4404))
+}