Merge branch 'master' into master

Author: lance6716

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)

replication/backup_test.go

@@ -2,11 +2,14 @@ package replication
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"os"
 	"path"
 	"time"
 
+	"github.com/google/uuid"
+
 	"github.com/stretchr/testify/require"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
@@ -61,6 +64,16 @@ func (t *testSyncerSuite) TestSyncBackup() {
 	testBackup(t, true) // true indicates synchronous mode
 }
 
+// TestAsyncBackupWithGTID runs the backup process in asynchronous mode with GTID and verifies binlog file creation.
+func (t *testSyncerSuite) TestAsyncBackupWithGTID() {
+	testBackUpWithGTID(t, false) // false indicates asynchronous mode
+}
+
+// TestSyncBackupWithGTID runs the backup process in synchronous mode with GTID and verifies binlog file creation.
+func (t *testSyncerSuite) TestSyncBackupWithGTID() {
+	testBackUpWithGTID(t, true) // true indicates synchronous mode
+}
+
 // testBackup is a helper function that runs the backup process in the specified mode and checks if binlog files are written correctly.
 func testBackup(t *testSyncerSuite, isSynchronous bool) {
 	t.setupTest(mysql.MySQLFlavor)
@@ -111,6 +124,71 @@ func testBackup(t *testSyncerSuite, isSynchronous bool) {
 	}
 }
 
+func testBackUpWithGTID(t *testSyncerSuite, isSynchronous bool) {
+	t.setupTest(mysql.MySQLFlavor)
+	t.b.cfg.SemiSyncEnabled = false // Ensure semi-sync is disabled
+
+	binlogDir := "./var"
+	os.RemoveAll(binlogDir)
+	timeout := 3 * time.Second
+
+	if isSynchronous {
+		// Set up a BackupEventHandler for synchronous mode
+		backupHandler := NewBackupEventHandler(
+			func(filename string) (io.WriteCloser, error) {
+				return os.OpenFile(path.Join(binlogDir, filename), os.O_CREATE|os.O_WRONLY, 0o644)
+			},
+		)
+		t.b.cfg.SynchronousEventHandler = backupHandler
+	} else {
+		// Ensure SynchronousEventHandler is nil for asynchronous mode
+		t.b.cfg.SynchronousEventHandler = nil
+	}
+
+	r, err := t.c.Execute("SELECT @@gtid_mode")
+	require.NoError(t.T(), err)
+	modeOn, _ := r.GetString(0, 0)
+	if modeOn != "ON" {
+		t.T().Skip("GTID mode is not ON")
+	}
+
+	r, err = t.c.Execute("SHOW GLOBAL VARIABLES LIKE 'SERVER_UUID'")
+	require.NoError(t.T(), err)
+
+	var masterUuid uuid.UUID
+	if s, _ := r.GetString(0, 1); len(s) > 0 && s != "NONE" {
+		masterUuid, err = uuid.Parse(s)
+		require.NoError(t.T(), err)
+	}
+
+	set, _ := mysql.ParseMysqlGTIDSet(fmt.Sprintf("%s:%d-%d", masterUuid.String(), 1, 2))
+	done := make(chan bool)
+
+	// Start the backup process in a goroutine
+	go func() {
+		err := t.b.StartBackupGTID(binlogDir, set, timeout)
+		require.NoError(t.T(), err)
+		done <- true
+	}()
+
+	failTimeout := 2 * timeout
+	ctx, cancel := context.WithTimeout(context.Background(), failTimeout)
+	defer cancel()
+
+	// Wait for the backup to complete or timeout
+	select {
+	case <-done:
+		files, err := os.ReadDir(binlogDir)
+		require.NoError(t.T(), err, "Failed to read binlog directory")
+		require.Greater(t.T(), len(files), 0, "Binlog files were not written to the directory")
+		mode := modeLabel(isSynchronous)
+		t.T().Logf("Backup completed successfully in %s mode using GTID with %d binlog file(s).", mode, len(files))
+	case <-ctx.Done():
+		mode := modeLabel(isSynchronous)
+		t.T().Fatalf("Timeout error during backup in %s mode.", mode)
+	}
+}
+
 func modeLabel(isSynchronous bool) string {
 	if isSynchronous {
 		return "synchronous"

replication/row_event.go

@@ -1317,7 +1317,7 @@ func (e *RowsEvent) decodeValue(data []byte, tp byte, meta uint16, isPartial boo
 		n = 4
 		t := binary.LittleEndian.Uint32(data)
 		if t == 0 {
-			v = formatZeroTime(0, 0)
+			v = "0000-00-00 00:00:00"
 		} else {
 			v = e.parseFracTime(fracTime{
 				Time:                    time.Unix(int64(t), 0),
@@ -1332,26 +1332,37 @@ func (e *RowsEvent) decodeValue(data []byte, tp byte, meta uint16, isPartial boo
 		n = 8
 		i64 := binary.LittleEndian.Uint64(data)
 		if i64 == 0 {
-			v = formatZeroTime(0, 0)
+			v = "0000-00-00 00:00:00"
 		} else {
 			d := i64 / 1000000
 			t := i64 % 1000000
-			v = e.parseFracTime(fracTime{
-				Time: time.Date(
-					int(d/10000),
-					time.Month((d%10000)/100),
-					int(d%100),
-					int(t/10000),
-					int((t%10000)/100),
-					int(t%100),
-					0,
-					time.UTC,
-				),
-				Dec: 0,
-			})
+			years := int(d / 10000)
+			months := int(d%10000) / 100
+			days := int(d % 100)
+			hours := int(t / 10000)
+			minutes := int(t%10000) / 100
+			seconds := int(t % 100)
+			if !e.parseTime || months == 0 || days == 0 {
+				v = fmt.Sprintf("%04d-%02d-%02d %02d:%02d:%02d",
+					years, months, days, hours, minutes, seconds)
+			} else {
+				v = e.parseFracTime(fracTime{
+					Time: time.Date(
+						years,
+						time.Month(months),
+						days,
+						hours,
+						minutes,
+						seconds,
+						0,
+						time.UTC,
+					),
+					Dec: 0,
+				})
+			}
 		}
 	case mysql.MYSQL_TYPE_DATETIME2:
-		v, n, err = decodeDatetime2(data, meta)
+		v, n, err = decodeDatetime2(data, meta, e.parseTime)
 		v = e.parseFracTime(v)
 	case mysql.MYSQL_TYPE_TIME:
 		n = 3
@@ -1676,7 +1687,7 @@ func decodeTimestamp2(data []byte, dec uint16, timestampStringLocation *time.Loc
 
 const DATETIMEF_INT_OFS int64 = 0x8000000000
 
-func decodeDatetime2(data []byte, dec uint16) (interface{}, int, error) {
+func decodeDatetime2(data []byte, dec uint16, parseTime bool) (interface{}, int, error) {
 	// get datetime binary length
 	n := int(5 + (dec+1)/2)
 
@@ -1726,8 +1737,8 @@ func decodeDatetime2(data []byte, dec uint16) (interface{}, int, error) {
 	// minute = 0 = 0b000000
 	// second = 0 = 0b000000
 	// integer value = 0b1100100000010110000100000000000000000 = 107420450816
-	if intPart < 107420450816 {
-		return formatBeforeUnixZeroTime(year, month, day, hour, minute, second, int(frac), int(dec)), n, nil
+	if !parseTime || intPart < 107420450816 || month == 0 || day == 0 {
+		return formatDatetime(year, month, day, hour, minute, second, int(frac), int(dec)), n, nil
 	}
 
 	return fracTime{

replication/row_event_test.go

@@ -675,6 +675,8 @@ func TestDecodeDatetime2(t *testing.T) {
 	}{
 		{[]byte("\xfe\xf3\xff\x7e\xfb"), 0, true, "9999-12-31 23:59:59"},
 		{[]byte("\x99\x9a\xb8\xf7\xaa"), 0, true, "2016-10-28 15:30:42"},
+		{[]byte("\x99\x98\x38\xf7\xaa"), 0, false, "2016-00-28 15:30:42"},
+		{[]byte("\x99\x9a\x80\xf7\xaa"), 0, false, "2016-10-00 15:30:42"},
 		{[]byte("\x99\x02\xc2\x00\x00"), 0, true, "1970-01-01 00:00:00"},
 		{[]byte("\x80\x00\x00\x00\x00"), 0, false, "0000-00-00 00:00:00"},
 		{[]byte("\x80\x00\x02\xf1\x05"), 0, false, "0000-00-01 15:04:05"},
@@ -684,17 +686,20 @@ func TestDecodeDatetime2(t *testing.T) {
 		{[]byte("\x80\x03\x82\x00\x00\x01\xe2\x40"), uint16(6), false, "0001-01-01 00:00:00.123456"},
 	}
 	for _, tc := range testcases {
-		value, _, err := decodeDatetime2(tc.data, tc.dec)
-		require.NoError(t, err)
-		switch v := value.(type) {
-		case fracTime:
-			require.True(t, tc.getFracTime)
-			require.Equal(t, tc.expected, v.String())
-		case string:
-			require.False(t, tc.getFracTime)
-			require.Equal(t, tc.expected, v)
-		default:
-			require.FailNow(t, "invalid value type: %T", value)
+		for _, parseTime := range []bool{true, false} {
+			value, _, err := decodeDatetime2(tc.data, tc.dec, parseTime)
+			require.NoError(t, err)
+			switch v := value.(type) {
+			case fracTime:
+				require.True(t, parseTime)
+				require.True(t, tc.getFracTime)
+				require.Equal(t, tc.expected, v.String())
+			case string:
+				require.False(t, parseTime && tc.getFracTime)
+				require.Equal(t, tc.expected, v)
+			default:
+				require.FailNow(t, "invalid value type: %T", value)
+			}
 		}
 	}
 }

replication/time.go

@@ -44,7 +44,7 @@ func formatZeroTime(frac int, dec int) string {
 	return s[0 : len(s)-(6-dec)]
 }
 
-func formatBeforeUnixZeroTime(year, month, day, hour, minute, second, frac, dec int) string {
+func formatDatetime(year, month, day, hour, minute, second, frac, dec int) string {
 	if dec == 0 {
 		return fmt.Sprintf("%04d-%02d-%02d %02d:%02d:%02d", year, month, day, hour, minute, second)
 	}