wip

Author: serprex

canal/expr.go

@@ -43,7 +43,7 @@ func (ve *valueExpr) GetString() string                             { return ""
 func (ve *valueExpr) GetProjectionOffset() int                      { return 0 }
 func (ve *valueExpr) SetProjectionOffset(offset int)                {}
 func (ve *valueExpr) Restore(ctx *format.RestoreCtx) error          { return nil }
-func (ve *valueExpr) Accept(v ast.Visitor) (node ast.Node, ok bool) { return }
+func (ve *valueExpr) Accept(v ast.Visitor) (node ast.Node, ok bool) { return node, ok }
 func (ve *valueExpr) Text() string                                  { return "" }
 func (ve *valueExpr) SetText(enc charset.Encoding, text string)     {}
 func (ve *valueExpr) Format(w io.Writer)                            {}

canal/sync.go

@@ -249,7 +249,7 @@ func (c *Canal) updateTable(header *replication.EventHeader, db, table string) (
 	if err = c.eventHandler.OnTableChanged(header, db, table); err != nil && errors.Cause(err) != schema.ErrTableNotExist {
 		return errors.Trace(err)
 	}
-	return
+	return err
 }
 
 func (c *Canal) updateReplicationDelay(ev *replication.BinlogEvent) {

client/auth.go

@@ -220,7 +220,7 @@ func (c *Conn) writeAuthHandshake() error {
 		c.ccaps&mysql.CLIENT_MULTI_STATEMENTS | c.ccaps&mysql.CLIENT_MULTI_RESULTS |
 		c.ccaps&mysql.CLIENT_PS_MULTI_RESULTS | c.ccaps&mysql.CLIENT_CONNECT_ATTRS |
 		c.ccaps&mysql.CLIENT_COMPRESS | c.ccaps&mysql.CLIENT_ZSTD_COMPRESSION_ALGORITHM |
-		c.ccaps&mysql.CLIENT_LOCAL_FILES
+		c.ccaps&mysql.CLIENT_LOCAL_FILES | c.ccaps&mysql.CLIENT_DEPRECATE_EOF
 
 	capability &^= c.clientExplicitOffCaps
 

client/resp.go

@@ -20,12 +20,12 @@ func (c *Conn) readUntilEOF() (err error) {
 	for {
 		data, err = c.ReadPacket()
 		if err != nil {
-			return
+			return err
 		}
 
 		// EOF Packet
 		if c.isEOFPacket(data) {
-			return
+			return err
 		}
 	}
 }
@@ -336,33 +336,16 @@ func (c *Conn) readResultsetStreaming(data []byte, binary bool, result *mysql.Re
 }
 
 func (c *Conn) readResultColumns(result *mysql.Result) (err error) {
-	i := 0
 	var data []byte
 
-	for {
+	for i := range len(result.Fields) {
 		rawPkgLen := len(result.RawPkg)
 		result.RawPkg, err = c.ReadPacketReuseMem(result.RawPkg)
 		if err != nil {
 			return err
 		}
 		data = result.RawPkg[rawPkgLen:]
 
-		// EOF Packet
-		if c.isEOFPacket(data) {
-			if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
-				result.Warnings = binary.LittleEndian.Uint16(data[1:])
-				// todo add strict_mode, warning will be treat as error
-				result.Status = binary.LittleEndian.Uint16(data[3:])
-				c.status = result.Status
-			}
-
-			if i != len(result.Fields) {
-				err = mysql.ErrMalformPacket
-			}
-
-			return err
-		}
-
 		if result.Fields[i] == nil {
 			result.Fields[i] = &mysql.Field{}
 		}
@@ -372,8 +355,30 @@ func (c *Conn) readResultColumns(result *mysql.Result) (err error) {
 		}
 
 		result.FieldNames[utils.ByteSliceToString(result.Fields[i].Name)] = i
+	}
+
+	if !c.HasCapability(mysql.CLIENT_DEPRECATE_EOF) {
+		// EOF Packet
+		rawPkgLen := len(result.RawPkg)
+		result.RawPkg, err = c.ReadPacketReuseMem(result.RawPkg)
+		if err != nil {
+			return err
+		}
+		data = result.RawPkg[rawPkgLen:]
 
-		i++
+		if c.isEOFPacket(data) {
+			if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
+				result.Warnings = binary.LittleEndian.Uint16(data[1:])
+				// todo add strict_mode, warning will be treat as error
+				result.Status = binary.LittleEndian.Uint16(data[3:])
+				c.status = result.Status
+			}
+			return nil
+		} else {
+			return mysql.ErrMalformPacket
+		}
+	} else {
+		return nil
 	}
 }
 
@@ -388,15 +393,21 @@ func (c *Conn) readResultRows(result *mysql.Result, isBinary bool) (err error) {
 		}
 		data = result.RawPkg[rawPkgLen:]
 
-		// EOF Packet
-		if c.isEOFPacket(data) {
-			if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
+		if data[0] == mysql.EOF_HEADER && len(data) <= 0xffffff {
+			if c.HasCapability(mysql.CLIENT_DEPRECATE_EOF) {
+				// Treat like OK
+				affectedRows, _, n := mysql.LengthEncodedInt(data[1:])
+				insertId, _, m := mysql.LengthEncodedInt(data[1+n:])
+				result.Status = binary.LittleEndian.Uint16(data[1+n+m:])
+				result.AffectedRows = affectedRows
+				result.InsertId = insertId
+				c.status = result.Status
+			} else if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
 				result.Warnings = binary.LittleEndian.Uint16(data[1:])
 				// todo add strict_mode, warning will be treat as error
 				result.Status = binary.LittleEndian.Uint16(data[3:])
 				c.status = result.Status
 			}
-
 			break
 		}
 

mysql/error.go

@@ -62,5 +62,5 @@ func ErrorCode(errMsg string) (code int) {
 	var tmpStr string
 	// golang scanf doesn't support %*,so I used a temporary variable
 	_, _ = fmt.Sscanf(errMsg, "%s%d", &tmpStr, &code)
-	return
+	return code
 }

mysql/mysql_gtid.go

@@ -42,14 +42,14 @@ func parseInterval(str string) (i Interval, err error) {
 	}
 
 	if err != nil {
-		return
+		return i, err
 	}
 
 	if i.Stop <= i.Start {
 		err = errors.Errorf("invalid interval format, must n[-n] and the end must >= start")
 	}
 
-	return
+	return i, err
 }
 
 func (i Interval) String() string {

mysql/resultset_helper.go

@@ -143,7 +143,7 @@ func fieldType(value interface{}) (typ uint8, err error) {
 	default:
 		err = errors.Errorf("unsupport type %T for resultset", value)
 	}
-	return
+	return typ, err
 }
 
 func formatField(field *Field, value interface{}) error {

replication/event.go

@@ -254,10 +254,10 @@ func decodeSid(data []byte) (format GtidFormat, sidnr uint64) {
 		masked := make([]byte, 8)
 		copy(masked, data[1:7])
 		sidnr = binary.LittleEndian.Uint64(masked)
-		return
+		return format, sidnr
 	}
 	sidnr = binary.LittleEndian.Uint64(data[:8])
-	return
+	return format, sidnr
 }
 
 func (e *PreviousGTIDsEvent) Decode(data []byte) error {

replication/row_event.go

@@ -344,18 +344,18 @@ func (e *TableMapEvent) decodeIntSeq(v []byte) (ret []uint64, err error) {
 		p += n
 		ret = append(ret, i)
 	}
-	return
+	return ret, err
 }
 
 func (e *TableMapEvent) decodeDefaultCharset(v []byte) (ret []uint64, err error) {
 	ret, err = e.decodeIntSeq(v)
 	if err != nil {
-		return
+		return ret, err
 	}
 	if len(ret)%2 != 1 {
 		return nil, errors.Errorf("Expect odd item in DefaultCharset but got %d", len(ret))
 	}
-	return
+	return ret, err
 }
 
 func (e *TableMapEvent) decodeColumnNames(v []byte) error {
@@ -390,7 +390,7 @@ func (e *TableMapEvent) decodeStrValue(v []byte) (ret [][][]byte, err error) {
 		}
 		ret = append(ret, vals)
 	}
-	return
+	return ret, err
 }
 
 func (e *TableMapEvent) decodeSimplePrimaryKey(v []byte) error {
@@ -561,7 +561,7 @@ func (e *TableMapEvent) Dump(w io.Writer) {
 // i must be in range [0, ColumnCount).
 func (e *TableMapEvent) Nullable(i int) (available, nullable bool) {
 	if len(e.NullBitmap) == 0 {
-		return
+		return available, nullable
 	}
 	return true, e.NullBitmap[i/8]&(1<<uint(i%8)) != 0
 }
@@ -1083,7 +1083,7 @@ func (e *RowsEvent) DecodeData(pos int, data []byte) (err2 error) {
 		data, err2 = mysql.DecompressMariadbData(data[pos:])
 		if err2 != nil {
 			//nolint:nakedret
-			return
+			return err2
 		}
 		pos = 0
 	}
@@ -1481,7 +1481,7 @@ func decodeString(data []byte, length int) (v string, n int) {
 		v = utils.ByteSliceToString(data[2:n])
 	}
 
-	return
+	return v, n
 }
 
 // ref: https://github.com/mysql/mysql-server/blob/a9b0c712de3509d8d08d3ba385d41a4df6348775/strings/decimal.c#L137
@@ -1502,7 +1502,7 @@ func decodeDecimalDecompressValue(compIndx int, data []byte, mask uint8) (size i
 	case 4:
 		value = uint32(data[3]^mask) | uint32(data[2]^mask)<<8 | uint32(data[1]^mask)<<16 | uint32(data[0]^mask)<<24
 	}
-	return
+	return size, value
 }
 
 var zeros = [digitsPerInteger]byte{48, 48, 48, 48, 48, 48, 48, 48, 48}
@@ -1625,7 +1625,7 @@ func decodeBit(data []byte, nbits int, length int) (value int64, err error) {
 			value = int64(data[0])
 		}
 	}
-	return
+	return value, err
 }
 
 func littleDecodeBit(data []byte, nbits int, length int) (value int64, err error) {
@@ -1657,7 +1657,7 @@ func littleDecodeBit(data []byte, nbits int, length int) (value int64, err error
 			value = int64(data[0])
 		}
 	}
-	return
+	return value, err
 }
 
 func decodeTimestamp2(data []byte, dec uint16, timestampStringLocation *time.Location) (interface{}, int, error) {
@@ -1859,7 +1859,7 @@ func decodeBlob(data []byte, meta uint16) (v []byte, n int, err error) {
 		err = fmt.Errorf("invalid blob packlen = %d", meta)
 	}
 
-	return
+	return v, n, err
 }
 
 func (e *RowsEvent) Dump(w io.Writer) {

serialization/serialization.go

@@ -58,7 +58,7 @@ func (f *Format) stringParts() (parts []string) {
 			parts = append(parts, fmt.Sprintf("  Value: %s", f.Type.String()))
 		}
 	}
-	return
+	return parts
 }
 
 // Field represents a `message_field`