Merge branch 'master' into v1_13_0

Author: lance6716

.github/workflows/ci.yml

@@ -10,8 +10,8 @@ jobs:
   test:
     strategy:
       matrix:
-        go: [ "1.24", "1.23", "1.22" ]
-        os: [ ubuntu-24.04, ubuntu-22.04 ]
+        go: [ "1.24", "1.23" ]
+        os: [ ubuntu-latest, ubuntu-24.04, ubuntu-22.04 ]
     name: Tests Go ${{ matrix.go }} on ${{ matrix.os }} # This name is used in main branch protection rules
     runs-on: ${{ matrix.os }}
 
@@ -56,8 +56,9 @@ jobs:
     strategy:
       matrix:
         mysql_version:
-          - 8.0.40
-          - 8.4.3
+          - 8.0.42
+          - 8.4.5
+          - 9.3.0
     name: Tests with MySQL ${{ matrix.mysql_version }}
     runs-on: ubuntu-latest
     services:

README.md

@@ -62,7 +62,7 @@ import (
 	"github.com/go-mysql-org/go-mysql/replication"
 	"os"
 )
-// Create a binlog syncer with a unique server id, the server id must be different from other MySQL's. 
+// Create a binlog syncer with a unique server id, the server id must be different from other MySQL's.
 // flavor is mysql or mariadb
 cfg := replication.BinlogSyncerConfig {
 	ServerID: 100,
@@ -133,13 +133,39 @@ Schema: test
 Query: DROP TABLE IF EXISTS `test_replication` /* generated by server */
 ```
 
-## Canal 
+### MariaDB 11.4+ compatibility
+
+MariaDB 11.4+ introduced an optimization where events written through transaction or statement cache have `LogPos=0` so they can be copied directly to the binlog without computing the real end position. This optimization improves performance but makes position tracking unreliable for replication clients that need to track LogPos of events inside transactions.
+
+To address this, a `FillZeroLogPos` configuration option is available:
+
+```go
+cfg := replication.BinlogSyncerConfig {
+	ServerID: 100,
+	Flavor:   "mariadb",
+	Host:     "127.0.0.1",
+	Port:     3306,
+	User:     "root",
+	Password: "",
+	// Enable dynamic LogPos calculation for MariaDB 11.4+
+	FillZeroLogPos: true,
+}
+```
+
+**Behavior:**
+- When `FillZeroLogPos` is `true` and flavor is `mariadb`, the library automatically:
+  - Adds `BINLOG_SEND_ANNOTATE_ROWS_EVENT` flag to binlog dump commands. This ensures correct position tracking by making the server send `ANNOTATE_ROWS_EVENT` events which are needed for accurate position calculation.
+  - Calculates LogPos dynamically for events with `LogPos=0` that are not artificial.
+- Only works with MariaDB flavor; has no effect with MySQL.
+- Should be set to `true` if tracking of LogPos inside transactions is required.
+
+## Canal
 
 Canal is a package that can sync your MySQL into everywhere, like Redis, Elasticsearch.
 
-First, canal will dump your MySQL data then sync changed data using binlog incrementally. 
+First, canal will dump your MySQL data then sync changed data using binlog incrementally.
 
-You must use ROW format for binlog, full binlog row image is preferred, because we may meet some errors when primary key changed in update for minimal or noblob row image. 
+You must use ROW format for binlog, full binlog row image is preferred, because we may meet some errors when primary key changed in update for minimal or noblob row image.
 
 A simple example:
 
@@ -188,9 +214,9 @@ You can see [go-mysql-elasticsearch](https://github.com/go-mysql-org/go-mysql-el
 
 ## Client
 
-Client package supports a simple MySQL connection driver which you can use it to communicate with MySQL server. 
+Client package supports a simple MySQL connection driver which you can use it to communicate with MySQL server.
 
-For an example see [`example_client_test.go`](client/example_client_test.go). You can run this testable example with 
+For an example see [`example_client_test.go`](client/example_client_test.go). You can run this testable example with
 `go test -v ./client -run Example`.
 
 Tested MySQL versions for the client include:
@@ -237,7 +263,7 @@ conn.Execute() / conn.Begin() / etc...
 
 ## Server
 
-Server package supplies a framework to implement a simple MySQL server which can handle the packets from the MySQL client. 
+Server package supplies a framework to implement a simple MySQL server which can handle the packets from the MySQL client.
 You can use it to build your own MySQL proxy. The server connection is compatible with MySQL 5.5, 5.6, 5.7, and 8.0 versions,
 so that most MySQL clients should be able to connect to the Server without modifications.
 
@@ -493,7 +519,7 @@ We pass all tests in https://github.com/bradfitz/go-sql-test using go-mysql driv
 
 Logging uses [log/slog](https://pkg.go.dev/log/slog) and by default is sent to standard out.
 
-For the old logging package `github.com/siddontang/go-log/log`, a converting package 
+For the old logging package `github.com/siddontang/go-log/log`, a converting package
 `https://github.com/serprex/slog-siddontang` is available.
 ## How to migrate to this repo
 To change the used package in your repo it's enough to add this `replace` directive to your `go.mod`:

canal/canal.go

@@ -51,7 +51,7 @@ type Canal struct {
 	includeTableRegex []*regexp.Regexp
 	excludeTableRegex []*regexp.Regexp
 
-	delay *uint32
+	delay atomic.Uint32
 
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -85,8 +85,6 @@ func NewCanal(cfg *Config) (*Canal, error) {
 	}
 	c.master = &masterInfo{logger: c.cfg.Logger}
 
-	c.delay = new(uint32)
-
 	var err error
 
 	if err = c.prepareDumper(); err != nil {
@@ -195,7 +193,7 @@ func (c *Canal) prepareDumper() error {
 }
 
 func (c *Canal) GetDelay() uint32 {
-	return atomic.LoadUint32(c.delay)
+	return c.delay.Load()
 }
 
 // Run will first try to dump all data from MySQL master `mysqldump`,
@@ -468,6 +466,8 @@ func (c *Canal) prepareSyncer() error {
 		Dialer:                  c.cfg.Dialer,
 		Localhost:               c.cfg.Localhost,
 		EventCacheCount:         c.cfg.EventCacheCount,
+		FillZeroLogPos:          c.cfg.FillZeroLogPos,
+
 		RowsEventDecodeFunc: func(event *replication.RowsEvent, data []byte) error {
 			pos, err := event.DecodeHeader(data)
 			if err != nil {

canal/config.go

@@ -113,6 +113,12 @@ type Config struct {
 	// the default value is 10240.
 	// if you table contain large columns, you can decrease this value to avoid OOM.
 	EventCacheCount int
+
+	// FillZeroLogPos enables dynamic LogPos calculation for MariaDB.
+	// When enabled, automatically adds BINLOG_SEND_ANNOTATE_ROWS_EVENT flag
+	// to ensure correct position calculation in MariaDB 11.4+.
+	// Only works with MariaDB flavor.
+	FillZeroLogPos bool
 }
 
 func NewConfigWithFile(name string) (*Config, error) {

canal/sync.go

@@ -2,7 +2,6 @@ package canal
 
 import (
 	"log/slog"
-	"sync/atomic"
 	"time"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
@@ -259,7 +258,7 @@ func (c *Canal) updateReplicationDelay(ev *replication.BinlogEvent) {
 	if now >= ev.Header.Timestamp {
 		newDelay = now - ev.Header.Timestamp
 	}
-	atomic.StoreUint32(c.delay, newDelay)
+	c.delay.Store(newDelay)
 }
 
 func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {

client/auth.go

@@ -222,6 +222,8 @@ func (c *Conn) writeAuthHandshake() error {
 		c.ccaps&mysql.CLIENT_COMPRESS | c.ccaps&mysql.CLIENT_ZSTD_COMPRESSION_ALGORITHM |
 		c.ccaps&mysql.CLIENT_LOCAL_FILES
 
+	capability &^= c.clientExplicitOffCaps
+
 	// To enable TLS / SSL
 	if c.tlsConfig != nil {
 		capability |= mysql.CLIENT_SSL

client/conn.go

@@ -45,6 +45,9 @@ type Conn struct {
 	capability uint32
 	// client-set capabilities only
 	ccaps uint32
+	// Capability flags explicitly disabled by the client via UnsetCapability()
+	// These flags are removed from the final advertised capability set during handshake.
+	clientExplicitOffCaps uint32
 
 	attributes map[string]string
 
@@ -234,14 +237,17 @@ func (c *Conn) Ping() error {
 	return nil
 }
 
-// SetCapability enables the use of a specific capability
+// SetCapability marks the specified flag as explicitly enabled by the client.
 func (c *Conn) SetCapability(cap uint32) {
 	c.ccaps |= cap
+	c.clientExplicitOffCaps &^= cap
 }
 
-// UnsetCapability disables the use of a specific capability
+// UnsetCapability marks the specified flag as explicitly disabled by the client.
+// This disables the flag even if the server supports it.
 func (c *Conn) UnsetCapability(cap uint32) {
-	c.ccaps &= ^cap
+	c.ccaps &^= cap
+	c.clientExplicitOffCaps |= cap
 }
 
 // HasCapability returns true if the connection has the specific capability

replication/backup.go

@@ -27,6 +27,20 @@ func (b *BinlogSyncer) StartBackup(backupDir string, p mysql.Position, timeout t
 	}
 }
 
+func (b *BinlogSyncer) StartBackupGTID(backupDir string, gset mysql.GTIDSet, timeout time.Duration) error {
+	err := os.MkdirAll(backupDir, 0o755)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if b.cfg.SynchronousEventHandler == nil {
+		return b.StartBackupWithHandlerAndGTID(gset, timeout, func(filename string) (io.WriteCloser, error) {
+			return os.OpenFile(path.Join(backupDir, filename), os.O_CREATE|os.O_WRONLY, 0o644)
+		})
+	} else {
+		return b.StartSynchronousBackupWithGTID(gset, timeout)
+	}
+}
+
 // StartBackupWithHandler starts the backup process for the binary log using the specified position and handler.
 // The process will continue until the timeout is reached or an error occurs.
 // This method should not be used together with SynchronousEventHandler.
@@ -54,52 +68,72 @@ func (b *BinlogSyncer) StartBackupWithHandler(p mysql.Position, timeout time.Dur
 	backupHandler := &BackupEventHandler{
 		handler: handler,
 	}
-
 	s, err := b.StartSync(p)
 	if err != nil {
 		return errors.Trace(err)
 	}
+	return processWithHandler(b, s, backupHandler, timeout)
+}
 
-	defer func() {
-		if backupHandler.w != nil {
-			closeErr := backupHandler.w.Close()
-			if retErr == nil {
-				retErr = closeErr
-			}
-		}
-	}()
+// StartBackupWithHandlerAndGTID starts the backup process for the binary log using the specified GTID set and handler.
+//   - gset: The GTID set from which to begin the backup.
+//   - timeout: The maximum duration to wait for new binlog events before stopping the backup process.
+//     If set to 0, a default very long timeout (30 days) is used instead.
+//   - handler: A function that takes a binlog filename and returns an WriteCloser for writing raw events to.
+func (b *BinlogSyncer) StartBackupWithHandlerAndGTID(gset mysql.GTIDSet, timeout time.Duration,
+	handler func(binlogFilename string) (io.WriteCloser, error),
+) (retErr error) {
+	if timeout == 0 {
+		// a very long timeout here
+		timeout = 30 * 3600 * 24 * time.Second
+	}
+	if b.cfg.SynchronousEventHandler != nil {
+		return errors.New("StartBackupWithHandlerAndGTID cannot be used when SynchronousEventHandler is set. Use StartSynchronousBackupWithGTID instead.")
+	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
+	// Force use raw mode
+	b.parser.SetRawMode(true)
 
-	for {
-		select {
-		case <-ctx.Done():
-			return nil
-		case <-b.ctx.Done():
-			return nil
-		case err := <-s.ech:
-			return errors.Trace(err)
-		case e := <-s.ch:
-			err = backupHandler.HandleEvent(e)
-			if err != nil {
-				return errors.Trace(err)
-			}
-		}
+	// Set up the backup event handler
+	backupHandler := &BackupEventHandler{
+		handler: handler,
+	}
+
+	s, err := b.StartSyncGTID(gset)
+	if err != nil {
+		return errors.Trace(err)
 	}
+	return processWithHandler(b, s, backupHandler, timeout)
 }
 
 // StartSynchronousBackup starts the backup process using the SynchronousEventHandler in the BinlogSyncerConfig.
 func (b *BinlogSyncer) StartSynchronousBackup(p mysql.Position, timeout time.Duration) error {
 	if b.cfg.SynchronousEventHandler == nil {
 		return errors.New("SynchronousEventHandler must be set in BinlogSyncerConfig to use StartSynchronousBackup")
 	}
-
 	s, err := b.StartSync(p)
 	if err != nil {
 		return errors.Trace(err)
 	}
 
+	return process(b, s, timeout)
+}
+
+// StartSynchronousBackupWithGTID starts the backup process using the SynchronousEventHandler in the BinlogSyncerConfig with a specified GTID set.
+func (b *BinlogSyncer) StartSynchronousBackupWithGTID(gset mysql.GTIDSet, timeout time.Duration) error {
+	if b.cfg.SynchronousEventHandler == nil {
+		return errors.New("SynchronousEventHandler must be set in BinlogSyncerConfig to use StartSynchronousBackupWithGTID")
+	}
+
+	s, err := b.StartSyncGTID(gset)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	return process(b, s, timeout)
+}
+
+func process(b *BinlogSyncer, s *BinlogStreamer, timeout time.Duration) error {
 	var ctx context.Context
 	var cancel context.CancelFunc
 
@@ -123,6 +157,35 @@ func (b *BinlogSyncer) StartSynchronousBackup(p mysql.Position, timeout time.Dur
 	}
 }
 
+func processWithHandler(b *BinlogSyncer, s *BinlogStreamer, backupHandler *BackupEventHandler, timeout time.Duration) (retErr error) {
+	defer func() {
+		if backupHandler.w != nil {
+			closeErr := backupHandler.w.Close()
+			if retErr == nil {
+				retErr = closeErr
+			}
+		}
+	}()
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-b.ctx.Done():
+			return nil
+		case err := <-s.ech:
+			return errors.Trace(err)
+		case e := <-s.ch:
+			err := backupHandler.HandleEvent(e)
+			if err != nil {
+				return errors.Trace(err)
+			}
+		}
+	}
+}
+
 // BackupEventHandler handles writing events for backup
 type BackupEventHandler struct {
 	handler func(binlogFilename string) (io.WriteCloser, error)