Skip to content

Add GTIDEvent interface to get the SequenceNumber and LastCommitted #818

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Sep 13, 2023
Merged

Add GTIDEvent interface to get the SequenceNumber and LastCommitted #818

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2ed1ac5
add gtid event interface
cameron-p-m Sep 7, 2023
17dc7aa
Merge branch 'master' into gtid-interface
cameron-p-m Sep 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions canal/handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ type EventHandler interface {
OnDDL(header *replication.EventHeader, nextPos mysql.Position, queryEvent *replication.QueryEvent) error
OnRow(e *RowsEvent) error
OnXID(header *replication.EventHeader, nextPos mysql.Position) error
OnGTID(header *replication.EventHeader, gtid mysql.GTIDSet) error
OnGTID(header *replication.EventHeader, gtidEvent mysql.BinlogGTIDEvent) error
// OnPosSynced Use your own way to sync position. When force is true, sync position immediately.
OnPosSynced(header *replication.EventHeader, pos mysql.Position, set mysql.GTIDSet, force bool) error
String() string
Expand All @@ -34,7 +34,9 @@ func (h *DummyEventHandler) OnDDL(*replication.EventHeader, mysql.Position, *rep
}
func (h *DummyEventHandler) OnRow(*RowsEvent) error { return nil }
func (h *DummyEventHandler) OnXID(*replication.EventHeader, mysql.Position) error { return nil }
func (h *DummyEventHandler) OnGTID(*replication.EventHeader, mysql.GTIDSet) error { return nil }
func (h *DummyEventHandler) OnGTID(*replication.EventHeader, mysql.BinlogGTIDEvent) error {
return nil
}
func (h *DummyEventHandler) OnPosSynced(*replication.EventHeader, mysql.Position, mysql.GTIDSet, bool) error {
return nil
}
Expand Down
16 changes: 2 additions & 14 deletions canal/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,12 @@
package canal

import (
"fmt"
"sync/atomic"
"time"

"github.com/go-mysql-org/go-mysql/mysql"
"github.com/go-mysql-org/go-mysql/replication"
"github.com/go-mysql-org/go-mysql/schema"
"github.com/google/uuid"
"github.com/pingcap/errors"
"github.com/pingcap/tidb/parser/ast"
)
Expand Down Expand Up @@ -118,21 +116,11 @@ func (c *Canal) runSyncBinlog() error {
c.master.UpdateGTIDSet(e.GSet)
}
case *replication.MariadbGTIDEvent:
// try to save the GTID later
gtid, err := mysql.ParseMariadbGTIDSet(e.GTID.String())
if err != nil {
return errors.Trace(err)
}
if err := c.eventHandler.OnGTID(ev.Header, gtid); err != nil {
if err := c.eventHandler.OnGTID(ev.Header, e); err != nil {
return errors.Trace(err)
}
case *replication.GTIDEvent:
u, _ := uuid.FromBytes(e.SID)
gtid, err := mysql.ParseMysqlGTIDSet(fmt.Sprintf("%s:%d", u.String(), e.GNO))
if err != nil {
return errors.Trace(err)
}
if err := c.eventHandler.OnGTID(ev.Header, gtid); err != nil {
if err := c.eventHandler.OnGTID(ev.Header, e); err != nil {
return errors.Trace(err)
}
case *replication.QueryEvent:
Expand Down
4 changes: 4 additions & 0 deletions mysql/gtid.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,3 +29,7 @@ func ParseGTIDSet(flavor string, s string) (GTIDSet, error) {
return nil, errors.Errorf("invalid flavor %s", flavor)
}
}

type BinlogGTIDEvent interface {
GTIDNext() (GTIDSet, error)
}
12 changes: 12 additions & 0 deletions replication/event.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -462,6 +462,14 @@ func (e *GTIDEvent) Dump(w io.Writer) {
fmt.Fprintln(w)
}

func (e *GTIDEvent) GTIDNext() (GTIDSet, error) {
u, err := uuid.FromBytes(e.SID)
if err != nil {
return nil, err
}
return ParseMysqlGTIDSet(strings.Join([]string{u.String(), strconv.FormatInt(e.GNO, 10)}, ":"))
}

// ImmediateCommitTime returns the commit time of this trx on the immediate server
// or zero time if not available.
func (e *GTIDEvent) ImmediateCommitTime() time.Time {
Expand Down Expand Up @@ -625,6 +633,10 @@ func (e *MariadbGTIDEvent) Dump(w io.Writer) {
fmt.Fprintln(w)
}

func (e *MariadbGTIDEvent) GTIDNext() (GTIDSet, error) {
return ParseMariadbGTIDSet(e.GTID.String())
}

type MariadbGTIDListEvent struct {
GTIDs []MariadbGTID
}
Expand Down
24 changes: 23 additions & 1 deletion replication/event_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
package replication

import (
"fmt"
"testing"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

Expand Down Expand Up @@ -49,6 +51,9 @@ func TestMariadbGTIDEvent(t *testing.T) {
require.True(t, ev.IsStandalone())
require.True(t, ev.IsGroupCommit())
require.Equal(t, uint64(0x1716151413121110), ev.CommitID)
set, err := ev.GTIDNext()
require.NoError(t, err)
require.Equal(t, "70975786-0-578437695752307201", set.String())
}

func TestGTIDEventMysql8NewFields(t *testing.T) {
Expand All @@ -59,6 +64,9 @@ func TestGTIDEventMysql8NewFields(t *testing.T) {
expectTransactionLength uint64
expectImmediateServerVersion uint32
expectOriginalServerVersion uint32
expectGTID string
expectSequnceNumber int64
expectLastCommitted int64
}{
{
// master: mysql-5.7, slave: mysql-8.0
Expand All @@ -68,6 +76,9 @@ func TestGTIDEventMysql8NewFields(t *testing.T) {
expectTransactionLength: 965,
expectImmediateServerVersion: 80019,
expectOriginalServerVersion: 0,
expectGTID: "5aa72a7f-44a8-11ea-947f-0242ac190002:258",
expectSequnceNumber: 119,
expectLastCommitted: 118,
},
{
// mysql-5.7 only
Expand All @@ -77,6 +88,9 @@ func TestGTIDEventMysql8NewFields(t *testing.T) {
expectTransactionLength: 0,
expectImmediateServerVersion: 0,
expectOriginalServerVersion: 0,
expectGTID: "5aa72a7f-44a8-11ea-947f-0242ac190002:259",
expectSequnceNumber: 54,
expectLastCommitted: 53,
},
{
// mysql-8.0 only
Expand All @@ -86,10 +100,13 @@ func TestGTIDEventMysql8NewFields(t *testing.T) {
expectTransactionLength: 963,
expectImmediateServerVersion: 80019,
expectOriginalServerVersion: 80019,
expectGTID: "5ccc1033-44a8-11ea-bd59-0242ac190003:119",
expectSequnceNumber: 121,
expectLastCommitted: 120,
},
}

for _, tc := range testcases {
for i, tc := range testcases {
ev := new(GTIDEvent)
err := ev.Decode(tc.data)
require.NoError(t, err)
Expand All @@ -98,6 +115,11 @@ func TestGTIDEventMysql8NewFields(t *testing.T) {
require.Equal(t, tc.expectTransactionLength, ev.TransactionLength)
require.Equal(t, tc.expectImmediateServerVersion, ev.ImmediateServerVersion)
require.Equal(t, tc.expectOriginalServerVersion, ev.OriginalServerVersion)
set, err := ev.GTIDNext()
require.NoError(t, err)
assert.Equal(t, tc.expectGTID, set.String(), fmt.Sprintf("testcase: %d", i))
assert.Equal(t, tc.expectSequnceNumber, ev.SequenceNumber, fmt.Sprintf("testcase: %d", i))
assert.Equal(t, tc.expectLastCommitted, ev.LastCommitted, fmt.Sprintf("testcase: %d", i))
}
}

Expand Down