diff --git a/go.mod b/go.mod index af30097c3..71ef0ac5d 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.25.12 require ( github.com/DataDog/datadog-go/v5 v5.8.3 github.com/go-ini/ini v1.67.0 - github.com/go-mysql-org/go-mysql v1.15.0 + github.com/go-mysql-org/go-mysql v1.16.0 github.com/go-sql-driver/mysql v1.8.1 github.com/google/uuid v1.6.0 github.com/hashicorp/go-version v1.7.0 diff --git a/go.sum b/go.sum index 709d499e2..8726540b3 100644 --- a/go.sum +++ b/go.sum @@ -49,8 +49,8 @@ github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= -github.com/go-mysql-org/go-mysql v1.15.0 h1:bZeRUc9yNVbFEyote79Q4j8SV+q8Ls32AYXRl2QjUoc= -github.com/go-mysql-org/go-mysql v1.15.0/go.mod h1:VjBTZTTDKL8OMXUAhNbg3VHaVVq9HOXJEBLpAKBFIfE= +github.com/go-mysql-org/go-mysql v1.16.0 h1:odv4Ygtc1WHJv3uUF2aoJdE1RS7tA0sD3ET91ZAWQIg= +github.com/go-mysql-org/go-mysql v1.16.0/go.mod h1:VjBTZTTDKL8OMXUAhNbg3VHaVVq9HOXJEBLpAKBFIfE= github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y= diff --git a/vendor/github.com/go-mysql-org/go-mysql/client/conn.go b/vendor/github.com/go-mysql-org/go-mysql/client/conn.go index 8f717ff08..dd29da444 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/client/conn.go +++ b/vendor/github.com/go-mysql-org/go-mysql/client/conn.go @@ -5,6 +5,8 @@ import ( "context" "crypto/tls" "fmt" + "io" + "log" "maps" "math/bits" "net" @@ -391,6 +393,184 @@ func (c *Conn) ExecuteSelectStreaming(command string, result *mysql.Result, perR return c.readResultStreaming(false, result, perRowCallback, perResultCallback) } +// prepareLocalInfileReader fully validates that r can be read without error before any data is +// sent upstream, per the LOCAL INFILE protocol requirement that on error the server must see +// only the terminal empty packet, never a truncated file. If r implements io.Seeker (as returned +// by callers per the documented example: *bytes.Reader, *os.File) it is drained and seeked back +// to the start so large files are not fully buffered in memory; otherwise the content is read +// fully into memory with io.ReadAll. +func prepareLocalInfileReader(r io.Reader) (io.Reader, error) { + if r == nil { + return bytes.NewReader(nil), nil + } + if seeker, ok := r.(io.Seeker); ok { + if _, err := io.Copy(io.Discard, r); err != nil { + return nil, err + } + if _, err := seeker.Seek(0, io.SeekStart); err != nil { + return nil, err + } + return r, nil + } + content, err := io.ReadAll(r) + if err != nil { + return nil, err + } + return bytes.NewReader(content), nil +} + +func (c *Conn) writeLocalInfileTerminator() error { + return c.WritePacket(make([]byte, 4)) +} + +// streamLocalInfileChunks sends already-validated file bytes as LOCAL INFILE data packets, +// followed by the terminal empty packet. +func (c *Conn) streamLocalInfileChunks(r io.Reader) error { + buf := make([]byte, 4+defaultBufferSize) + sentData := false + for { + n, err := r.Read(buf[4:]) + if n > 0 { + if werr := c.WritePacket(buf[:4+n]); werr != nil { + if tErr := c.writeLocalInfileTerminator(); tErr != nil { + log.Printf("go-mysql: failed to send LOCAL INFILE terminator after write error: %v", tErr) + } + return errors.Trace(werr) + } + sentData = true + } + if err == io.EOF { + break + } + if err != nil { + if sentData { + if tErr := c.writeLocalInfileTerminator(); tErr != nil { + log.Printf("go-mysql: failed to send LOCAL INFILE terminator after read error: %v", tErr) + } + } + return errors.Trace(err) + } + } + if err := c.writeLocalInfileTerminator(); err != nil { + return errors.Trace(err) + } + return nil +} + +// sendLocalInfileContentAndAwaitResult validates and sends LOCAL INFILE content, then reads the +// server's OK or ERR. When relayErr is non-nil, or when reader validation fails, only the empty +// terminator packet is sent (no file data). +func (c *Conn) sendLocalInfileContentAndAwaitResult(reader io.Reader, relayErr error) (*mysql.Result, error) { + content := io.Reader(bytes.NewReader(nil)) + if relayErr == nil { + validated, err := prepareLocalInfileReader(reader) + if err != nil { + relayErr = err + } else { + content = validated + } + } + streamErr := error(nil) + if err := c.streamLocalInfileChunks(content); err != nil { + streamErr = err + } + if relayErr == nil && streamErr != nil { + relayErr = streamErr + } + resp, err := c.ReadPacket() + if err != nil { + return nil, errors.Trace(err) + } + if len(resp) == 0 { + return nil, errors.New("unexpected empty packet after LOCAL INFILE") + } + var result *mysql.Result + var respErr error + switch resp[0] { + case mysql.OK_HEADER: + result, respErr = c.handleOKPacket(resp) + case mysql.ERR_HEADER: + respErr = c.handleErrorPacket(resp) + default: + respErr = errors.Errorf("unexpected packet after LOCAL INFILE: 0x%x", resp[0]) + } + if relayErr != nil { + if respErr != nil { + return nil, errors.Errorf("local infile relay failed: %v; server response: %v", relayErr, respErr) + } + return nil, errors.Trace(relayErr) + } + return result, errors.Trace(respErr) +} + +// ExecQueryRelayLocalInfile sends COM_QUERY and handles the LOCAL INFILE protocol when the server +// responds with a 0xfb packet. relayFile receives the filename bytes from that request (without the +// 0xfb header) and must return a reader over the complete file content (or nil for an empty file). +// The library validates the reader, sends data packets and the terminal empty packet, then returns +// the final OK or ERR. +// +// If relayFile returns an error, or the reader cannot be fully read, no file data is sent upstream; +// only the empty terminator packet is sent so the connection can be reused. +// +// Notes: +// - This function is intended for LOAD DATA LOCAL INFILE queries. +// - relayFile is only invoked when the server responds with LocalInFile_HEADER (0xfb). +// - Direct OK or ERR (no 0xfb) occurs when the server rejects the statement before requesting a +// file — e.g. local_infile disabled, missing privileges, or a syntax error. Handling these +// responses keeps the connection usable, consistent with readResultStreaming and +// ExecuteMultiple. +// +// This is the proxy-friendly counterpart to the local-file reading in client/auth.go. It does not +// access the filesystem; ownership of the file transfer is delegated entirely to the caller. +// +// Example (MySQL proxy relaying LOAD DATA LOCAL INFILE from an application client to upstream): +// +// result, err := upstream.ExecQueryRelayLocalInfile(query, func(filename []byte) (io.Reader, error) { +// if err := downstream.WritePacket(wrapPacket(append([]byte{mysql.LocalInFile_HEADER}, filename...))); err != nil { +// return nil, err +// } +// var buf bytes.Buffer +// for { +// pkt, err := downstream.ReadPacket() +// if err != nil { +// return nil, err +// } +// if len(pkt) == 0 { +// break // empty packet = end of file from client +// } +// if _, err := buf.Write(pkt); err != nil { +// return nil, err +// } +// } +// return bytes.NewReader(buf.Bytes()), nil +// }) +func (c *Conn) ExecQueryRelayLocalInfile(query string, relayFile func(filename []byte) (io.Reader, error)) (*mysql.Result, error) { + if err := c.execSend(query); err != nil { + return nil, errors.Trace(err) + } + data, err := c.ReadPacket() + if err != nil { + return nil, errors.Trace(err) + } + if len(data) == 0 { + return nil, errors.New("unexpected empty packet from server") + } + switch data[0] { + case mysql.OK_HEADER: + return c.handleOKPacket(data) + case mysql.ERR_HEADER: + return nil, c.handleErrorPacket(data) + case mysql.LocalInFile_HEADER: + reader, relayErr := relayFile(data[1:]) + if closer, ok := reader.(io.Closer); ok && reader != nil { + defer closer.Close() + } + return c.sendLocalInfileContentAndAwaitResult(reader, relayErr) + default: + return nil, errors.Errorf("unexpected response to COM_QUERY (expected OK, ERR, or LOCAL INFILE): 0x%x", data[0]) + } +} + func (c *Conn) Begin() error { _, err := c.exec("BEGIN") return errors.Trace(err) diff --git a/vendor/github.com/go-mysql-org/go-mysql/client/resp.go b/vendor/github.com/go-mysql-org/go-mysql/client/resp.go index b5fe09d91..b4a329c88 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/client/resp.go +++ b/vendor/github.com/go-mysql-org/go-mysql/client/resp.go @@ -46,29 +46,29 @@ func (c *Conn) handleOKPacket(data []byte) (*mysql.Result, error) { pos += 2 } - if (c.capability&mysql.CLIENT_SESSION_TRACK > 0) && - (c.status&mysql.SERVER_SESSION_STATE_CHANGED > 0) { - var err error - - // Example status message: - // "Records: 3 Duplicates: 0 Warnings: 0" - statusMessageLength := int(data[pos]) - pos++ + if c.capability&mysql.CLIENT_SESSION_TRACK > 0 { + // info string is always present when CLIENT_SESSION_TRACK is negotiated. + // Example: "Records: 3 Duplicates: 0 Warnings: 0" + statusMessageLength, _, n := mysql.LengthEncodedInt(data[pos:]) + pos += n if statusMessageLength > 0 { - r.StatusMessage = utils.ByteSliceToString(data[pos : pos+statusMessageLength]) - pos += statusMessageLength + r.StatusMessage = utils.ByteSliceToString(data[pos : pos+int(statusMessageLength)]) + pos += int(statusMessageLength) } - sessionTrackingChangeLength := int(data[pos]) - pos++ - dataLength := len(data[pos:]) - if dataLength != sessionTrackingChangeLength { - return nil, fmt.Errorf("incorrect data length for session tracking data: expected %d but got %d", - sessionTrackingChangeLength, dataLength) - } - r.SessionTracking, err = decodeSessionTracking(data[pos:]) - if err != nil { - return nil, err + if c.status&mysql.SERVER_SESSION_STATE_CHANGED > 0 { + sessionTrackingChangeLength, _, n := mysql.LengthEncodedInt(data[pos:]) + pos += n + dataLength := len(data[pos:]) + if dataLength != int(sessionTrackingChangeLength) { + return nil, fmt.Errorf("incorrect data length for session tracking data: expected %d but got %d", + sessionTrackingChangeLength, dataLength) + } + var err error + r.SessionTracking, err = decodeSessionTracking(data[pos:]) + if err != nil { + return nil, err + } } } @@ -82,26 +82,32 @@ func decodeSessionTracking(data []byte) (s *mysql.SessionTrackingInfo, err error for pos < len(data) { sessionTrackingChangeType := data[pos] pos++ // session tracking type - pos++ // length of session tracking data, unused + _, _, n := mysql.LengthEncodedInt(data[pos:]) + pos += n // length of session tracking data, unused switch sessionTrackingChangeType { case mysql.SESSION_TRACK_SYSTEM_VARIABLES: if s.Variables == nil { s.Variables = make(map[string]string, 1) } - varNameLength := data[pos] - pos++ - varName := utils.ByteSliceToString(data[pos : pos+int(varNameLength)]) - pos += int(varNameLength) - varValueLength := data[pos] - pos++ - s.Variables[varName] = utils.ByteSliceToString(data[pos : pos+int(varValueLength)]) - pos += int(varValueLength) + varName, _, n, err := mysql.LengthEncodedString(data[pos:]) + if err != nil { + return nil, err + } + pos += n + varValue, _, n, err := mysql.LengthEncodedString(data[pos:]) + if err != nil { + return nil, err + } + pos += n + s.Variables[utils.ByteSliceToString(varName)] = utils.ByteSliceToString(varValue) case mysql.SESSION_TRACK_SCHEMA: - schemaInfoLength := data[pos] - pos++ - s.Schema = utils.ByteSliceToString(data[pos : pos+int(schemaInfoLength)]) - pos += int(schemaInfoLength) + schema, _, n, err := mysql.LengthEncodedString(data[pos:]) + if err != nil { + return nil, err + } + pos += n + s.Schema = utils.ByteSliceToString(schema) case mysql.SESSION_TRACK_STATE_CHANGE: s.State = string(data[pos]) pos++ @@ -111,22 +117,26 @@ func decodeSessionTracking(data []byte) (s *mysql.SessionTrackingInfo, err error return nil, fmt.Errorf("unexpected GTID format %d", gtidFormat) } pos++ - gtidLength := data[pos] - pos++ - s.GTID = utils.ByteSliceToString(data[pos : pos+int(gtidLength)]) - pos += int(gtidLength) + gtid, _, n, err := mysql.LengthEncodedString(data[pos:]) + if err != nil { + return nil, err + } + pos += n + s.GTID = utils.ByteSliceToString(gtid) case mysql.SESSION_TRACK_TRANSACTION_CHARACTERISTICS: - characteristicsLength := data[pos] - pos++ - if characteristicsLength > 0 { - s.Characteristics = utils.ByteSliceToString(data[pos : pos+int(characteristicsLength)]) - pos += int(characteristicsLength) + chars, _, n, err := mysql.LengthEncodedString(data[pos:]) + if err != nil { + return nil, err } + pos += n + s.Characteristics = utils.ByteSliceToString(chars) case mysql.SESSION_TRACK_TRANSACTION_STATE: - transactionStateLength := data[pos] - pos++ - s.TransactionState = utils.ByteSliceToString(data[pos : pos+int(transactionStateLength)]) - pos += int(transactionStateLength) + txState, _, n, err := mysql.LengthEncodedString(data[pos:]) + if err != nil { + return nil, err + } + pos += n + s.TransactionState = utils.ByteSliceToString(txState) default: return nil, fmt.Errorf("got unknown change type %v", sessionTrackingChangeType) } diff --git a/vendor/github.com/go-mysql-org/go-mysql/client/stmt.go b/vendor/github.com/go-mysql-org/go-mysql/client/stmt.go index c7a89981f..7443dcd9b 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/client/stmt.go +++ b/vendor/github.com/go-mysql-org/go-mysql/client/stmt.go @@ -49,6 +49,49 @@ func (s *Stmt) ExecuteSelectStreaming(result *mysql.Result, perRowCb SelectPerRo return s.conn.readResultStreaming(true, result, perRowCb, perResCb) } +// StmtProcedureMultiResultForward is called once per logical result returned by +// COM_STMT_EXECUTE. When err is non-nil, res is nil. Implementations typically +// forward each result to a proxy client via server.Conn.WriteValue. +// A nil return from forward means the result or error was handled successfully. +type StmtProcedureMultiResultForward func(res *mysql.Result, err error) error + +// ExecuteProcedureMultiResults runs COM_STMT_EXECUTE and reads every response +// until SERVER_MORE_RESULTS_EXISTS is clear (CALL / stored procedures with +// multiple result sets). forward is called for each result in arrival order; +// the loop continues draining the server response even after forward returns +// non-nil, matching ExecuteMultiple semantics so that unread packets do not +// corrupt the connection state. +// +// The first non-nil error returned by forward is saved and returned after all +// results have been drained. If forward handles an error and returns nil, that +// error is not propagated. +func (s *Stmt) ExecuteProcedureMultiResults(forward StmtProcedureMultiResultForward, args ...any) error { + if forward == nil { + return errors.New("forward callback cannot be nil") + } + if err := s.write(args...); err != nil { + return errors.Trace(err) + } + var forwardErr error + for { + res, err := s.conn.readResult(true) + if forwardErr == nil { + if err != nil { + forwardErr = forward(nil, err) + } else if res != nil { + forwardErr = forward(res, nil) + } + } + if err != nil { + break + } + if res == nil || res.Status&mysql.SERVER_MORE_RESULTS_EXISTS == 0 { + break + } + } + return forwardErr +} + func (s *Stmt) Close() error { if err := s.conn.writeCommandUint32(mysql.COM_STMT_CLOSE, s.ID); err != nil { return errors.Trace(err) diff --git a/vendor/github.com/go-mysql-org/go-mysql/compress/zlib.go b/vendor/github.com/go-mysql-org/go-mysql/compress/zlib.go index 474c02b63..c831353a8 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/compress/zlib.go +++ b/vendor/github.com/go-mysql-org/go-mysql/compress/zlib.go @@ -57,7 +57,7 @@ func GetPooledZlibReader(src io.Reader) (io.ReadCloser, error) { if r := zlibReaderPool.Get(); r != nil { rc = r.(io.ReadCloser) - if rc.(zlib.Resetter).Reset(src, nil) != nil { + if err = rc.(zlib.Resetter).Reset(src, nil); err != nil { return nil, err } } else { diff --git a/vendor/github.com/go-mysql-org/go-mysql/mysql/result.go b/vendor/github.com/go-mysql-org/go-mysql/mysql/result.go index 06a069991..aaa5c22f7 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/mysql/result.go +++ b/vendor/github.com/go-mysql-org/go-mysql/mysql/result.go @@ -1,5 +1,7 @@ package mysql +import "sort" + // Result should be created by NewResultWithoutRows or NewResult. The zero value // of Result is invalid. type Result struct { @@ -26,6 +28,107 @@ type SessionTrackingInfo struct { Characteristics string } +// AppendOKSessionTrackSuffix appends the OK-packet session tracking suffix when +// CLIENT_SESSION_TRACK is negotiated: [statusMessageLen][statusMessage], and +// when SERVER_SESSION_STATE_CHANGED is set in r.Status, also +// [sessionTrackBlockLen][sessionTrackBlock]. +func AppendOKSessionTrackSuffix(data []byte, r *Result) []byte { + if r == nil { + return data + } + + statusMessage := r.StatusMessage + data = append(data, PutLengthEncodedInt(uint64(len(statusMessage)))...) + if len(statusMessage) > 0 { + data = append(data, statusMessage...) + } + + if r.Status&SERVER_SESSION_STATE_CHANGED == 0 { + return data + } + + block := encodeSessionTracking(r.SessionTracking) + data = append(data, PutLengthEncodedInt(uint64(len(block)))...) + if len(block) > 0 { + data = append(data, block...) + } + + return data +} + +// EncodeSessionTracking serializes a SessionTrackingInfo into the +// session-state-changes block used in OK packets. +func EncodeSessionTracking(s *SessionTrackingInfo) []byte { + return encodeSessionTracking(s) +} + +func encodeSessionTracking(s *SessionTrackingInfo) []byte { + if s == nil { + return nil + } + + var data []byte + + if len(s.Variables) > 0 { + names := make([]string, 0, len(s.Variables)) + for name := range s.Variables { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + value := s.Variables[name] + var payload []byte + payload = appendLenEncString(payload, name) + payload = appendLenEncString(payload, value) + data = appendSessionTrackEntry(data, SESSION_TRACK_SYSTEM_VARIABLES, payload) + } + } + + if s.Schema != "" { + var payload []byte + payload = appendLenEncString(payload, s.Schema) + data = appendSessionTrackEntry(data, SESSION_TRACK_SCHEMA, payload) + } + + if s.State != "" { + data = appendSessionTrackEntry(data, SESSION_TRACK_STATE_CHANGE, []byte(s.State[:1])) + } + + if s.GTID != "" { + var payload []byte + payload = append(payload, 0x00) + payload = appendLenEncString(payload, s.GTID) + data = appendSessionTrackEntry(data, SESSION_TRACK_GTIDS, payload) + } + + if s.Characteristics != "" { + var payload []byte + payload = appendLenEncString(payload, s.Characteristics) + data = appendSessionTrackEntry(data, SESSION_TRACK_TRANSACTION_CHARACTERISTICS, payload) + } + + if s.TransactionState != "" { + var payload []byte + payload = appendLenEncString(payload, s.TransactionState) + data = appendSessionTrackEntry(data, SESSION_TRACK_TRANSACTION_STATE, payload) + } + + return data +} + +func appendLenEncString(data []byte, s string) []byte { + data = append(data, PutLengthEncodedInt(uint64(len(s)))...) + data = append(data, s...) + return data +} + +func appendSessionTrackEntry(data []byte, trackType byte, payload []byte) []byte { + data = append(data, trackType) + data = append(data, PutLengthEncodedInt(uint64(len(payload)))...) + data = append(data, payload...) + return data +} + func NewResult(resultset *Resultset) *Result { return &Result{ Resultset: resultset, diff --git a/vendor/github.com/go-mysql-org/go-mysql/packet/conn.go b/vendor/github.com/go-mysql-org/go-mysql/packet/conn.go index 8a2eae6d8..0dd6353d6 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/packet/conn.go +++ b/vendor/github.com/go-mysql-org/go-mysql/packet/conn.go @@ -50,8 +50,6 @@ type Conn struct { compressedHeader [7]byte compressedReader io.Reader - - compressedReaderActive bool } func NewConn(conn net.Conn) *Conn { @@ -106,20 +104,16 @@ func (c *Conn) ReadPacketReuseMem(dst []byte) ([]byte, error) { utils.BytesBufferPut(buf) }() - if c.Compression != mysql.MYSQL_COMPRESS_NONE { - // it's possible that we're using compression but the server response with a compressed - // packet with uncompressed length of 0. In this case we leave compressedReader nil. The - // compressedReaderActive flag is important to track the state of the reader, allowing - // for the compressedReader to be reset after a packet write. Without this flag, when a - // compressed packet with uncompressed length of 0 is read, the compressedReader would - // be nil, and we'd incorrectly attempt to read the next packet as compressed. - if !c.compressedReaderActive { - var err error - c.compressedReader, err = c.newCompressedPacketReader() - if err != nil { - return nil, err - } - c.compressedReaderActive = true + // compressedReader is reset to nil after each WritePacket, so a nil reader here means + // we're at the start of a new compressed frame and need to read its header. While a + // frame still has buffered packets to hand out, we reuse the existing reader rather + // than consuming another frame header. (newCompressedPacketReader never returns a nil + // reader, so its nilness fully tracks whether a frame read is in progress.) + if c.Compression != mysql.MYSQL_COMPRESS_NONE && c.compressedReader == nil { + var err error + c.compressedReader, err = c.newCompressedPacketReader() + if err != nil { + return nil, err } } @@ -168,8 +162,12 @@ func (c *Conn) newCompressedPacketReader() (io.Reader, error) { compressedLength := int(uint32(c.compressedHeader[0]) | uint32(c.compressedHeader[1])<<8 | uint32(c.compressedHeader[2])<<16) uncompressedLength := int(uint32(c.compressedHeader[4]) | uint32(c.compressedHeader[5])<<8 | uint32(c.compressedHeader[6])<<16) + + // Always bound reads to this frame's payload (compressedLength bytes on the wire). + // copyN relies on hitting EOF at the frame boundary to advance CompressedSequence + // and move on to the next compressed packet. + limitedReader := io.LimitReader(c.reader, int64(compressedLength)) if uncompressedLength > 0 { - limitedReader := io.LimitReader(c.reader, int64(compressedLength)) switch c.Compression { case mysql.MYSQL_COMPRESS_ZLIB: return compress.GetPooledZlibReader(limitedReader) @@ -178,7 +176,12 @@ func (c *Conn) newCompressedPacketReader() (io.Reader, error) { } } - return nil, nil + // uncompressedLength == 0 means the payload was sent verbatim (compression wasn't + // worthwhile for this chunk). It must still be bounded to the frame: returning the + // raw, unbounded connection here lets a packet that spans into the following frame + // read straight through that frame's header, desyncing the compressed stream and + // surfacing later as "invalid compressed sequence" / "zlib: invalid header". + return limitedReader, nil } func (c *Conn) currentPacketReader() io.Reader { @@ -327,7 +330,6 @@ func (c *Conn) WritePacket(data []byte) error { return errors.Wrapf(mysql.ErrBadConn, "Write failed. only %v bytes written, while %v expected", n, len(data)) } - c.compressedReaderActive = false if c.compressedReader != nil { if _, ok := c.compressedReader.(io.ReadCloser); ok { _ = c.compressedReader.(io.ReadCloser).Close() diff --git a/vendor/github.com/go-mysql-org/go-mysql/replication/binlogsyncer.go b/vendor/github.com/go-mysql-org/go-mysql/replication/binlogsyncer.go index 6cdf4e586..b500dd414 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/replication/binlogsyncer.go +++ b/vendor/github.com/go-mysql-org/go-mysql/replication/binlogsyncer.go @@ -79,6 +79,37 @@ type BinlogSyncerConfig struct { // FloatWithTrailingZero structure for floats. UseFloatWithTrailingZero bool + // RenderJSONAsMySQLText, when true, makes the JSONB decoder emit text + // that is faithful to each value's original JSONB type tag where the + // JSON text grammar can express it. The default decode->json.Marshal + // path is lossy for DOUBLE and (less importantly) NEWDECIMAL, which + // matters when replaying the output back into a MySQL JSON column. + // + // Per-tag behaviour: + // - JSONB_DOUBLE 1.0 renders as "1.0" (not "1"), so MySQL re-stores + // it as JSONB_DOUBLE rather than JSONB_INT. + // - JSONB_OPAQUE NEWDECIMAL renders as an unquoted number rather + // than a quoted string. Note that MySQL's JSON text grammar has + // no syntax for a decimal literal: re-inserting the text creates + // a JSON DOUBLE, not the original JSONB_OPAQUE NEWDECIMAL. The + // numeric value is preserved, the opaque type tag is not. + // - JSONB_OPAQUE DATE renders as "YYYY-MM-DD" (not the legacy + // "YYYY-MM-DD 00:00:00.000000"). + // - JSONB_OPAQUE values of unrecognised inner types render as + // "base64:typeN:" (matching mysqld) instead of the raw + // payload bytes. + // - Object key order is preserved from the JSONB stream + // (length-then-bytes) instead of the lexicographic order Go maps + // produce, matching MySQL's own text output. + // + // Other notes: + // - UseDecimal and UseFloatWithTrailingZero have no effect on JSON + // columns when this is enabled (the renderer always emits + // MySQL-style text). + // - Only applies to JSON columns; non-JSON DECIMAL/DATE/etc. columns + // are unaffected. + RenderJSONAsMySQLText bool + // RecvBufferSize sets the size in bytes of the operating system's receive buffer associated with the connection. RecvBufferSize int @@ -211,6 +242,7 @@ func NewBinlogSyncer(cfg BinlogSyncerConfig) *BinlogSyncer { b.parser.SetTimestampStringLocation(b.cfg.TimestampStringLocation) b.parser.SetUseDecimal(b.cfg.UseDecimal) b.parser.SetUseFloatWithTrailingZero(b.cfg.UseFloatWithTrailingZero) + b.parser.SetRenderJSONAsMySQLText(b.cfg.RenderJSONAsMySQLText) b.parser.SetVerifyChecksum(b.cfg.VerifyChecksum) b.parser.SetPayloadDecoderConcurrency(cfg.PayloadDecoderConcurrency) b.parser.SetRowsEventDecodeFunc(b.cfg.RowsEventDecodeFunc) @@ -393,18 +425,24 @@ func (b *BinlogSyncer) enableSemiSync() error { return nil } - r, err := b.c.Execute("SHOW VARIABLES LIKE 'rpl_semi_sync_master_enabled';") + // MySQL 8.0.26 renamed rpl_semi_sync_master_enabled to + // rpl_semi_sync_source_enabled (keeping the old name as an alias) and + // 8.4.0 removed the alias, so accept either spelling. + r, err := b.c.Execute("SHOW VARIABLES WHERE Variable_name IN ('rpl_semi_sync_master_enabled', 'rpl_semi_sync_source_enabled')") if err != nil { return errors.Trace(err) } s, _ := r.GetString(0, 1) if s != "ON" { - b.cfg.Logger.Error("master does not support semi synchronous replication, use no semi-sync") + b.cfg.Logger.Error("source does not support semi synchronous replication, use no semi-sync") b.cfg.SemiSyncEnabled = false return nil } - _, err = b.c.Execute(`SET @rpl_semi_sync_slave = 1;`) + // MySQL 8.0.26 also renamed the @rpl_semi_sync_slave session variable + // to @rpl_semi_sync_replica. These are user variables, so setting both + // is harmless on either server version. + _, err = b.c.Execute(`SET @rpl_semi_sync_slave = 1, @rpl_semi_sync_replica = 1;`) if err != nil { return errors.Trace(err) } @@ -969,6 +1007,21 @@ func (b *BinlogSyncer) handleEventAndACK(s *BinlogStreamer, e *BinlogEvent, need if !b.cfg.DiscardGTIDSet { event.GSet = b.getCurrentGtidSet() } + + case *TransactionPayloadEvent: + // XID/Query decoded from compressed payload need GTID set attached, + // same as their uncompressed counterparts above; GTID event precedes + // payload uncompressed, so currGset already covers this transaction + if !b.cfg.DiscardGTIDSet { + for _, inner := range event.Events { + switch innerEvent := inner.Event.(type) { + case *XIDEvent: + innerEvent.GSet = b.getCurrentGtidSet() + case *QueryEvent: + innerEvent.GSet = b.getCurrentGtidSet() + } + } + } } // Use SynchronousEventHandler if it's set diff --git a/vendor/github.com/go-mysql-org/go-mysql/replication/json_binary.go b/vendor/github.com/go-mysql-org/go-mysql/replication/json_binary.go index 1dd3b6f7e..619118330 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/replication/json_binary.go +++ b/vendor/github.com/go-mysql-org/go-mysql/replication/json_binary.go @@ -1,6 +1,7 @@ package replication import ( + "encoding/base64" "fmt" "math" "strconv" @@ -136,21 +137,34 @@ func jsonbGetValueEntrySize(isSmall bool) int { return jsonbValueEntrySizeLarge } -// decodeJSONBinary decodes the JSON binary encoding data and returns -// the common JSON encoding data. +// decodeJSONBinary decodes the JSON binary encoding data and returns the +// common JSON encoding data. When RenderJSONAsMySQLText is set on the +// parent RowsEvent the decoder wraps leaf values in MySQL-text marshalers +// (see json_mysql_text.go) so json.Marshal produces MySQL's textual JSON +// form, faithful to each JSONB value's original type tag where the JSON +// text grammar can express it. NEWDECIMAL is the one tag that cannot be +// preserved on text round-trip (no decimal literal in JSON); see +// BinlogSyncerConfig.RenderJSONAsMySQLText for the full caveat list. func (e *RowsEvent) decodeJSONBinary(data []byte) ([]byte, error) { d := jsonBinaryDecoder{ useDecimal: e.useDecimal, useFloatWithTrailingZero: e.useFloatWithTrailingZero, ignoreDecodeErr: e.ignoreJSONDecodeErr, + mysqlTextMode: e.renderJSONAsMySQLText, } if d.isDataShort(data, 1) { + if d.ignoreDecodeErr { + return []byte("null"), nil + } return nil, d.err } v := d.decodeValue(data[0], data[1:]) if d.err != nil { + if d.ignoreDecodeErr { + return []byte("null"), nil + } return nil, d.err } @@ -161,6 +175,7 @@ type jsonBinaryDecoder struct { useDecimal bool useFloatWithTrailingZero bool ignoreDecodeErr bool + mysqlTextMode bool err error } @@ -193,12 +208,19 @@ func (d *jsonBinaryDecoder) decodeValue(tp byte, data []byte) any { case JSONB_UINT64: return d.decodeUint64(data) case JSONB_DOUBLE: + if d.mysqlTextMode { + return jsonMySQLDouble(d.decodeDouble(data)) + } if d.useFloatWithTrailingZero { return d.decodeDoubleWithTrailingZero(data) } return d.decodeDouble(data) case JSONB_STRING: - return d.decodeString(data) + s := d.decodeString(data) + if d.mysqlTextMode { + return jsonString(s) + } + return s case JSONB_OPAQUE: return d.decodeOpaque(data) default: @@ -301,6 +323,13 @@ func (d *jsonBinaryDecoder) decodeObjectOrArray(data []byte, isSmall bool, isObj return values } + if d.mysqlTextMode { + // Preserve JSONB key order (length-then-bytes, which is what MySQL + // emits) instead of going through map[string]any, which json.Marshal + // would sort lexicographically. + return jsonObject{keys: keys, values: values} + } + m := make(map[string]any, count) for i := range count { m[keys[i]] = values[i] @@ -459,20 +488,47 @@ func (d *jsonBinaryDecoder) decodeOpaque(data []byte) any { return d.decodeDecimal(data) case mysql.MYSQL_TYPE_TIME: return d.decodeTime(data) - case mysql.MYSQL_TYPE_DATE, mysql.MYSQL_TYPE_DATETIME, mysql.MYSQL_TYPE_TIMESTAMP: - return d.decodeDateTime(data) + case mysql.MYSQL_TYPE_DATE: + // Historically dates have been decoded the same as datetime (including the time portion). + // This is mostly harmless, but in text-mode we want to ensure that + // the time portion is omitted. + return d.decodeDateTime(data, d.mysqlTextMode) + case mysql.MYSQL_TYPE_DATETIME, mysql.MYSQL_TYPE_TIMESTAMP: + return d.decodeDateTime(data, false) default: + if d.mysqlTextMode { + return "base64:type" + strconv.Itoa(int(tp)) + ":" + base64.StdEncoding.EncodeToString(data) + } return utils.ByteSliceToString(data) } } func (d *jsonBinaryDecoder) decodeDecimal(data []byte) any { + if d.isDataShort(data, 2) { + return nil + } precision := int(data[0]) scale := int(data[1]) - v, _, err := decodeDecimal(data[2:], precision, scale, d.useDecimal) - d.err = err - + // MySQL renders JSON DECIMAL values unquoted; force the string form + // (useDecimal=false) so we can wrap it as a jsonRawNumber. + useDecimal := d.useDecimal + if d.mysqlTextMode { + useDecimal = false + } + v, _, err := decodeDecimal(data[2:], precision, scale, useDecimal) + if err != nil { + d.err = err + return nil + } + if d.mysqlTextMode { + s, ok := v.(string) + if !ok { + d.err = errors.Errorf("decimal decode returned %T, want string", v) + return nil + } + return jsonRawNumber(s) + } return v } @@ -498,9 +554,12 @@ func (d *jsonBinaryDecoder) decodeTime(data []byte) any { return fmt.Sprintf("%s%02d:%02d:%02d.%06d", sign, hour, minute, sec, frac) } -func (d *jsonBinaryDecoder) decodeDateTime(data []byte) any { +func (d *jsonBinaryDecoder) decodeDateTime(data []byte, isDate bool) any { v := d.decodeInt64(data) if v == 0 { + if isDate { + return "0000-00-00" + } return "0000-00-00 00:00:00" } @@ -522,6 +581,9 @@ func (d *jsonBinaryDecoder) decodeDateTime(data []byte) any { second := hms % (1 << 6) frac := v % (1 << 24) + if isDate { + return fmt.Sprintf("%04d-%02d-%02d", year, month, day) + } return fmt.Sprintf("%04d-%02d-%02d %02d:%02d:%02d.%06d", year, month, day, hour, minute, second, frac) } diff --git a/vendor/github.com/go-mysql-org/go-mysql/replication/json_mysql_text.go b/vendor/github.com/go-mysql-org/go-mysql/replication/json_mysql_text.go new file mode 100644 index 000000000..9983ac47d --- /dev/null +++ b/vendor/github.com/go-mysql-org/go-mysql/replication/json_mysql_text.go @@ -0,0 +1,147 @@ +package replication + +import ( + "bytes" + "math" + "strconv" + + "github.com/goccy/go-json" +) + +// This file holds the MySQL-text marshalers used by jsonBinaryDecoder +// when its mysqlTextMode flag is set. Wrapping the leaf decode returns +// in these types lets the existing json.Marshal pass produce JSON text +// that is faithful to each JSONB value's original type tag where the +// JSON text grammar can express it (DOUBLE 1.0 stays "1.0"; NEWDECIMAL +// stays unquoted; etc.) and preserves the JSONB key order. +// +// Caveats: +// - The output is type-faithful, not byte-identical to MySQL's +// "SELECT json_col" form. Inter-token whitespace is compact (no +// space after ',' or ':') and floating-point text differs in some +// exponent/precision corner cases (see jsonMySQLDouble). +// - NEWDECIMAL is the one tag that cannot be preserved on text +// round-trip: MySQL's JSON text grammar has no decimal literal, so +// re-inserting the unquoted number yields a JSON DOUBLE, not the +// original JSONB_OPAQUE NEWDECIMAL. The numeric value still +// round-trips; only the opaque type tag is lost. All other tags +// covered here do reproduce the original JSONB binary on re-insert. + +// jsonString carries a JSONB string payload as raw bytes so MarshalJSON +// can pass non-ASCII bytes through verbatim. MySQL JSON is byte- +// transparent, so bytes >= 0x20 (other than '"' and '\\') are written +// without UTF-8 validation -- unlike the default encoding/json path which +// replaces invalid UTF-8 with U+FFFD. +type jsonString string + +func (s jsonString) MarshalJSON() ([]byte, error) { + buf := bytes.NewBuffer(make([]byte, 0, len(s)+2)) + buf.WriteByte('"') + writeJSONString(buf, []byte(s)) + buf.WriteByte('"') + return buf.Bytes(), nil +} + +// jsonRawNumber emits its bytes unquoted. Used for JSONB OPAQUE +// NEWDECIMAL values: MySQL renders these as plain numbers in JSON text, +// not as quoted strings. +type jsonRawNumber string + +func (n jsonRawNumber) MarshalJSON() ([]byte, error) { + return []byte(n), nil +} + +// jsonMySQLDouble formats a float64 close to the way MySQL does in JSON +// text: whole-number doubles keep a trailing ".0" so MySQL re-stores +// them as JSON DOUBLE rather than JSON INTEGER, and non-integer values +// use the shortest round-trippable form. We can't reuse +// FloatWithTrailingZero here because it formats non-integers with 'f' +// (always plain decimal); MySQL uses scientific notation for some +// magnitudes, which 'g' matches more closely. +// +// The output is NOT guaranteed to be byte-identical to MySQL's +// my_gcvt-formatted text: Go's 'g' verb emits exponents as e.g. +// "1.5e-05" where MySQL emits "1.5e-5", and the integer/scientific +// crossover threshold differs. Binary round-trip through a MySQL JSON +// column is unaffected (the same float64 produces the same JSONB +// DOUBLE bytes); only the visible text form may differ. +type jsonMySQLDouble float64 + +func (f jsonMySQLDouble) MarshalJSON() ([]byte, error) { + return []byte(formatMySQLDouble(float64(f))), nil +} + +// jsonObject preserves JSONB key order (length-then-bytes, which is what +// MySQL emits) instead of going through map[string]any, which json.Marshal +// would sort lexicographically. +type jsonObject struct { + keys []string + values []any +} + +func (o jsonObject) MarshalJSON() ([]byte, error) { + var buf bytes.Buffer + buf.WriteByte('{') + for i, k := range o.keys { + if i > 0 { + buf.WriteByte(',') + } + buf.WriteByte('"') + writeJSONString(&buf, []byte(k)) + buf.WriteString(`":`) + vb, err := json.Marshal(o.values[i]) + if err != nil { + return nil, err + } + buf.Write(vb) + } + buf.WriteByte('}') + return buf.Bytes(), nil +} + +func formatMySQLDouble(f float64) string { + if math.IsNaN(f) || math.IsInf(f, 0) { + // MySQL refuses to store NaN/Inf in JSON; emit a safe fallback + // rather than corrupt the surrounding document. + return "null" + } + if f == math.Trunc(f) { + return strconv.FormatFloat(f, 'f', 1, 64) + } + return strconv.FormatFloat(f, 'g', -1, 64) +} + +// writeJSONString writes s as the contents of a JSON string (no +// surrounding quotes), with byte-transparent semantics: bytes >= 0x20 +// other than '"' and '\\' are written verbatim, including high-bit bytes +// that may not form valid UTF-8. +func writeJSONString(buf *bytes.Buffer, s []byte) { + const hexdigits = "0123456789abcdef" + for i := range len(s) { + c := s[i] + if c < 0x20 || c == '"' || c == '\\' { + switch c { + case '"': + buf.WriteString(`\"`) + case '\\': + buf.WriteString(`\\`) + case '\b': + buf.WriteString(`\b`) + case '\f': + buf.WriteString(`\f`) + case '\n': + buf.WriteString(`\n`) + case '\r': + buf.WriteString(`\r`) + case '\t': + buf.WriteString(`\t`) + default: + buf.WriteString(`\u00`) + buf.WriteByte(hexdigits[c>>4]) + buf.WriteByte(hexdigits[c&0xF]) + } + continue + } + buf.WriteByte(c) + } +} diff --git a/vendor/github.com/go-mysql-org/go-mysql/replication/parser.go b/vendor/github.com/go-mysql-org/go-mysql/replication/parser.go index 804966c74..79deae535 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/replication/parser.go +++ b/vendor/github.com/go-mysql-org/go-mysql/replication/parser.go @@ -37,6 +37,7 @@ type BinlogParser struct { useDecimal bool useFloatWithTrailingZero bool + renderJSONAsMySQLText bool ignoreJSONDecodeErr bool verifyChecksum bool @@ -205,6 +206,12 @@ func (p *BinlogParser) SetUseFloatWithTrailingZero(useFloatWithTrailingZero bool p.useFloatWithTrailingZero = useFloatWithTrailingZero } +// SetRenderJSONAsMySQLText toggles MySQL-text JSON rendering for RowsEvents. +// See BinlogSyncerConfig.RenderJSONAsMySQLText for the full rationale. +func (p *BinlogParser) SetRenderJSONAsMySQLText(renderJSONAsMySQLText bool) { + p.renderJSONAsMySQLText = renderJSONAsMySQLText +} + func (p *BinlogParser) SetIgnoreJSONDecodeError(ignoreJSONDecodeErr bool) { p.ignoreJSONDecodeErr = ignoreJSONDecodeErr } @@ -229,6 +236,30 @@ func (p *BinlogParser) SetTableMapOptionalMetaDecodeFunc(tableMapOptionalMetaDec p.tableMapOptionalMetaDecodeFunc = tableMapOptionalMetaDecondeFunc } +// cloneForPayloadDecode returns a new BinlogParser that inherits the +// caller's user-settable decode options (UseDecimal, RenderJSONAsMySQLText, +// ParseTime, etc.) but with checksum verification disabled and a fresh +// tables map. It is used to parse the events nested inside a +// TRANSACTION_PAYLOAD_EVENT, so that those rows decode with the same +// options as uncompressed rows. +func (p *BinlogParser) cloneForPayloadDecode() *BinlogParser { + inner := NewBinlogParser() + inner.flavor = p.flavor + inner.rawMode = p.rawMode + inner.parseTime = p.parseTime + inner.timestampStringLocation = p.timestampStringLocation + inner.useDecimal = p.useDecimal + inner.useFloatWithTrailingZero = p.useFloatWithTrailingZero + inner.renderJSONAsMySQLText = p.renderJSONAsMySQLText + inner.ignoreJSONDecodeErr = p.ignoreJSONDecodeErr + // verifyChecksum is intentionally left at the zero value: nested + // events do not carry their own checksum trailers. + inner.payloadDecoderConcurrency = p.payloadDecoderConcurrency + inner.rowsEventDecodeFunc = p.rowsEventDecodeFunc + inner.tableMapOptionalMetaDecodeFunc = p.tableMapOptionalMetaDecodeFunc + return inner +} + func (p *BinlogParser) parseHeader(data []byte) (*EventHeader, error) { h := new(EventHeader) err := h.Decode(data) @@ -356,6 +387,10 @@ func (p *BinlogParser) parseEvent(h *EventHeader, data []byte, rawData []byte) ( p.tables[te.TableID] = te } + if tpe, ok := e.(*TransactionPayloadEvent); ok { + tpe.stampInnerEventPositions(h) + } + if re, ok := e.(*RowsEvent); ok { if (re.Flags & RowsEventStmtEndFlag) > 0 { // Refer https://github.com/alibaba/canal/blob/38cc81b7dab29b51371096fb6763ca3a8432ffee/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java#L176 @@ -432,6 +467,7 @@ func (p *BinlogParser) newRowsEvent(h *EventHeader) *RowsEvent { e.timestampStringLocation = p.timestampStringLocation e.useDecimal = p.useDecimal e.useFloatWithTrailingZero = p.useFloatWithTrailingZero + e.renderJSONAsMySQLText = p.renderJSONAsMySQLText e.ignoreJSONDecodeErr = p.ignoreJSONDecodeErr switch h.EventType { @@ -477,6 +513,7 @@ func (p *BinlogParser) newTransactionPayloadEvent() *TransactionPayloadEvent { e := &TransactionPayloadEvent{} e.format = *p.format e.concurrency = p.payloadDecoderConcurrency + e.parent = p return e } diff --git a/vendor/github.com/go-mysql-org/go-mysql/replication/row_event.go b/vendor/github.com/go-mysql-org/go-mysql/replication/row_event.go index eaef3edd5..c47bf0cce 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/replication/row_event.go +++ b/vendor/github.com/go-mysql-org/go-mysql/replication/row_event.go @@ -953,6 +953,7 @@ type RowsEvent struct { timestampStringLocation *time.Location useDecimal bool useFloatWithTrailingZero bool + renderJSONAsMySQLText bool ignoreJSONDecodeErr bool } diff --git a/vendor/github.com/go-mysql-org/go-mysql/replication/transaction_payload_event.go b/vendor/github.com/go-mysql-org/go-mysql/replication/transaction_payload_event.go index aab9d6a0f..cd5da5a10 100644 --- a/vendor/github.com/go-mysql-org/go-mysql/replication/transaction_payload_event.go +++ b/vendor/github.com/go-mysql-org/go-mysql/replication/transaction_payload_event.go @@ -29,8 +29,15 @@ const ( ) type TransactionPayloadEvent struct { - format FormatDescriptionEvent - concurrency int + format FormatDescriptionEvent + concurrency int + // parent is the BinlogParser that produced this event. The inner + // parser used to decode the decompressed payload inherits its + // user-settable options (UseDecimal, RenderJSONAsMySQLText, ...) so + // that compressed and uncompressed rows decode identically. nil when + // the event is constructed outside of BinlogParser, in which case + // decodePayload falls back to default parser options. + parent *BinlogParser Size uint64 UncompressedSize uint64 CompressionType uint64 @@ -99,6 +106,23 @@ func (e *TransactionPayloadEvent) decodeFields(data []byte) error { return nil } +// stampInnerEventPositions fixes LogPos of decoded inner events. MySQL copies +// them from transaction cache before position fixup, so they arrive with +// LogPos=0. Stamp checkpoint-equivalent positions: payload start for +// mid-transaction events (resume there replays whole transaction, matching +// uncompressed semantics), payload end for final event (XID/COMMIT, resume +// there skips transaction) +func (e *TransactionPayloadEvent) stampInnerEventPositions(h *EventHeader) { + if len(e.Events) == 0 || h.LogPos < h.EventSize { + return + } + start := h.LogPos - h.EventSize + for _, inner := range e.Events { + inner.Header.LogPos = start + } + e.Events[len(e.Events)-1].Header.LogPos = h.LogPos +} + func (e *TransactionPayloadEvent) decodePayload() error { if e.CompressionType != ZSTD { return fmt.Errorf("TransactionPayloadEvent has compression type %d (%s)", @@ -117,10 +141,18 @@ func (e *TransactionPayloadEvent) decodePayload() error { } // The uncompressed data needs to be split up into individual events for Parse() - // to work on them. We can't use e.parser directly as we need to disable checksums - // but we still need the initialization from the FormatDescriptionEvent. We can't - // modify e.parser as it is used elsewhere. - parser := NewBinlogParser() + // to work on them. We can't use the parent parser directly as we need to disable + // checksums but we still need the initialization from the FormatDescriptionEvent. + // We can't modify the parent parser as it is used elsewhere. We do however want + // to inherit user-settable decode options (UseDecimal, RenderJSONAsMySQLText, + // IgnoreJSONDecodeError, ...) so that rows inside a compressed payload decode + // with the same semantics as uncompressed rows. + var parser *BinlogParser + if e.parent != nil { + parser = e.parent.cloneForPayloadDecode() + } else { + parser = NewBinlogParser() + } parser.format = &FormatDescriptionEvent{ Version: e.format.Version, ServerVersion: e.format.ServerVersion, diff --git a/vendor/modules.txt b/vendor/modules.txt index b91f4fb0c..64d11a761 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -77,7 +77,7 @@ github.com/go-logr/logr/funcr # github.com/go-logr/stdr v1.2.2 ## explicit; go 1.16 github.com/go-logr/stdr -# github.com/go-mysql-org/go-mysql v1.15.0 +# github.com/go-mysql-org/go-mysql v1.16.0 ## explicit; go 1.25.0 github.com/go-mysql-org/go-mysql/client github.com/go-mysql-org/go-mysql/compress