From 143dbf148073abecf86b3ed6e6f8fb2195c67238 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Tue, 21 Jul 2026 19:53:34 +0200 Subject: [PATCH 1/3] Add GTID binlog coordinate mode (representation + config + reads) Extend BinlogCoordinate with a GTID representation and add the BinlogCoordinateMode feature flag, without switching binlog streaming. - BinlogCoordinate gains a GTID form (NewGTIDCoordinate, IsGTID, ParsedGTIDSet, GTID-aware IsZero/String, JSON). GTID coordinates use set containment (Contains); Compare panics for GTID since GTID sets have no total order. - Config gains BinlogCoordinateMode ("file_position" default, "gtid" experimental) with validation. - Add DB helpers: ReadExecutedGTIDSet, ReadCurrentGTIDCoordinate, and CheckServerGTIDModeEnabled for @@GLOBAL.GTID_EXECUTED / gtid_mode. - Unit tests for GTID coordinate semantics, JSON round-trip, and mode validation. Streaming still runs on file/position; this only adds the GTID representation, feature flag, and validation groundwork. --- binlog_coordinate.go | 96 +++++++++++++++++++++++++++++-- config.go | 25 ++++++++ test/go/binlog_coordinate_test.go | 84 +++++++++++++++++++++++++++ test/go/config_test.go | 19 ++++++ utils.go | 48 ++++++++++++++++ 5 files changed, 267 insertions(+), 5 deletions(-) diff --git a/binlog_coordinate.go b/binlog_coordinate.go index 007280640..84bf486b1 100644 --- a/binlog_coordinate.go +++ b/binlog_coordinate.go @@ -28,14 +28,19 @@ const ( // BinlogCoordinate is a representation-agnostic replication coordinate. // -// Today it only wraps a file/position (mysql.Position). It exists so that the -// rest of Ghostferry can be migrated to talk in terms of "a coordinate" rather -// than "a file and a position", which is a prerequisite for adding a GTID mode -// behind a feature flag without a second invasive refactor. +// It can express either a file/position (mysql.Position) or a GTID set. It +// exists so that the rest of Ghostferry can be migrated to talk in terms of "a +// coordinate" rather than "a file and a position", which is a prerequisite for +// adding a GTID mode behind a feature flag without a second invasive refactor. // // The zero value is a zero file/position coordinate, matching the previous // behavior where an empty mysql.Position was used as the "no coordinate" // sentinel. +// +// GTID coordinates are stored canonically as the GTID set string. The parsed +// mysql.GTIDSet is derived on demand; it is intentionally not stored so that +// the value type stays trivially copyable and comparable-by-value-free (GTID +// sets are mutable and must be cloned before mutation). type BinlogCoordinate struct { // Type selects the active representation. An empty Type is treated as // BinlogCoordinateFilePosition for backwards compatibility. @@ -43,6 +48,13 @@ type BinlogCoordinate struct { // FilePosition holds the coordinate when Type is BinlogCoordinateFilePosition. FilePosition mysql.Position + + // GTIDSet holds the coordinate as a canonical MySQL GTID set string when + // Type is BinlogCoordinateGTID, e.g. + // "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-57". An empty string means "no + // GTIDs" which is distinct from a nil/unset coordinate; callers that need + // that distinction should check Type and IsZero together. + GTIDSet string } // NewFilePositionCoordinate wraps a mysql.Position in a BinlogCoordinate. @@ -53,6 +65,16 @@ func NewFilePositionCoordinate(pos mysql.Position) BinlogCoordinate { } } +// NewGTIDCoordinate builds a GTID coordinate from a canonical GTID set string. +// The string is not validated here; use ParsedGTIDSet or the DB read helpers +// when a parsed/validated set is required. +func NewGTIDCoordinate(gtidSet string) BinlogCoordinate { + return BinlogCoordinate{ + Type: BinlogCoordinateGTID, + GTIDSet: gtidSet, + } +} + // resolvedType returns the effective type, treating the empty string as // file/position for backwards compatibility with older serialized state and // zero-valued coordinates. @@ -68,6 +90,11 @@ func (c BinlogCoordinate) IsFilePosition() bool { return c.resolvedType() == BinlogCoordinateFilePosition } +// IsGTID reports whether this coordinate is a GTID coordinate. +func (c BinlogCoordinate) IsGTID() bool { + return c.resolvedType() == BinlogCoordinateGTID +} + // Position returns the underlying mysql.Position. // // It is valid to call this only for file/position coordinates. This accessor @@ -78,12 +105,31 @@ func (c BinlogCoordinate) Position() mysql.Position { return c.FilePosition } +// ParsedGTIDSet parses and returns the underlying GTID set. +// +// It is valid to call this only for GTID coordinates. A fresh mysql.GTIDSet is +// returned on each call so callers may mutate it freely without affecting the +// coordinate. +func (c BinlogCoordinate) ParsedGTIDSet() (mysql.GTIDSet, error) { + if c.resolvedType() != BinlogCoordinateGTID { + return nil, fmt.Errorf("ParsedGTIDSet called on non-GTID coordinate of type %q", c.resolvedType()) + } + return mysql.ParseMysqlGTIDSet(c.GTIDSet) +} + // IsZero reports whether the coordinate carries no meaningful position. This // mirrors the previous use of an empty mysql.Position as the "unset" sentinel. +// +// For GTID coordinates, an empty GTID set string is considered zero. Note that +// an empty GTID set is semantically "earlier than everything"; callers that +// must distinguish "no usable coordinate" from "genuinely empty GTID set" +// should track that distinction separately (e.g. a nil coordinate pointer). func (c BinlogCoordinate) IsZero() bool { switch c.resolvedType() { case BinlogCoordinateFilePosition: return c.FilePosition == (mysql.Position{}) + case BinlogCoordinateGTID: + return c.GTIDSet == "" default: return false } @@ -96,6 +142,11 @@ func (c BinlogCoordinate) IsZero() bool { // programmer error and panics, because there is no meaningful total ordering // across representations. Callers that may receive mixed types should branch on // the coordinate type first. +// +// GTID coordinates do not form a total order (a set can contain another, be +// contained by it, both, or neither). Compare therefore does not support GTID +// coordinates; use Contains for GTID reachability checks instead. This mirrors +// the reality that "which GTID set is further ahead" is not well defined. func (c BinlogCoordinate) Compare(other BinlogCoordinate) int { if c.resolvedType() != other.resolvedType() { panic(fmt.Sprintf( @@ -108,8 +159,35 @@ func (c BinlogCoordinate) Compare(other BinlogCoordinate) int { case BinlogCoordinateFilePosition: return c.FilePosition.Compare(other.FilePosition) default: - panic(fmt.Sprintf("comparison not implemented for binlog coordinate type %q", c.resolvedType())) + panic(fmt.Sprintf("Compare not supported for binlog coordinate type %q; use Contains for GTID", c.resolvedType())) + } +} + +// Contains reports whether this GTID coordinate's set fully contains the other +// GTID coordinate's set. This is the correct "have we reached/passed" check for +// GTID based stop and catchup conditions. +// +// Both coordinates must be GTID coordinates. Calling Contains on file/position +// coordinates returns an error, since containment is not the file/position +// reachability model (Compare is). +func (c BinlogCoordinate) Contains(other BinlogCoordinate) (bool, error) { + if c.resolvedType() != BinlogCoordinateGTID || other.resolvedType() != BinlogCoordinateGTID { + return false, fmt.Errorf( + "Contains is only defined for GTID coordinates, got %q and %q", + c.resolvedType(), other.resolvedType(), + ) + } + + mine, err := c.ParsedGTIDSet() + if err != nil { + return false, fmt.Errorf("parsing GTID set %q: %w", c.GTIDSet, err) } + theirs, err := other.ParsedGTIDSet() + if err != nil { + return false, fmt.Errorf("parsing GTID set %q: %w", other.GTIDSet, err) + } + + return mine.Contain(theirs), nil } // String returns a human readable representation for logs and status output. @@ -117,6 +195,8 @@ func (c BinlogCoordinate) String() string { switch c.resolvedType() { case BinlogCoordinateFilePosition: return fmt.Sprintf("%s:%d", c.FilePosition.Name, c.FilePosition.Pos) + case BinlogCoordinateGTID: + return fmt.Sprintf("gtid:%s", c.GTIDSet) default: return fmt.Sprintf("", c.resolvedType()) } @@ -129,6 +209,7 @@ func (c BinlogCoordinate) String() string { type serializedBinlogCoordinate struct { Type BinlogCoordinateType `json:"Type"` FilePosition *mysql.Position `json:"FilePosition,omitempty"` + GTIDSet string `json:"GTIDSet,omitempty"` } // MarshalJSON implements json.Marshaler. @@ -139,6 +220,8 @@ func (c BinlogCoordinate) MarshalJSON() ([]byte, error) { case BinlogCoordinateFilePosition: pos := c.FilePosition out.FilePosition = &pos + case BinlogCoordinateGTID: + out.GTIDSet = c.GTIDSet default: return nil, fmt.Errorf("cannot marshal binlog coordinate of type %q", c.resolvedType()) } @@ -162,6 +245,9 @@ func (c *BinlogCoordinate) UnmarshalJSON(data []byte) error { c.FilePosition = mysql.Position{} } return nil + case BinlogCoordinateGTID: + c.GTIDSet = typed.GTIDSet + return nil default: return fmt.Errorf("cannot unmarshal binlog coordinate of type %q", typed.Type) } diff --git a/config.go b/config.go index d696b79ab..afbcc41e0 100644 --- a/config.go +++ b/config.go @@ -684,6 +684,22 @@ type Config struct { // Make sure you have binlog_row_image=FULL when turning on this SkipBinlogRowImageCheck bool + // BinlogCoordinateMode selects how Ghostferry expresses and persists binlog + // coordinates. + // + // Valid values: + // - "" or "file_position": the classic (file, position) coordinate. This + // is the default and current behavior. + // - "gtid": GTID-set based coordinates (experimental, MySQL only). At this + // stage GTID mode only affects coordinate representation, validation, and + // the ability to read the current GTID set; binlog streaming still uses + // file/position until a later stage wires GTID streaming. + // + // Prefer BinlogCoordinateMode over any future boolean flag: it keeps the + // file/position and GTID paths cleanly separated and leaves room for + // additional coordinate sources. + BinlogCoordinateMode BinlogCoordinateType + // This config is necessary for inline verification for a special case of // Ghostferry: // @@ -853,6 +869,15 @@ func (c *Config) ValidateConfig() error { return fmt.Errorf("StateToResumeFrom version mismatch: resume = %s, current = %s", c.StateToResumeFrom.GhostferryVersion, VersionString) } + switch c.BinlogCoordinateMode { + case "": + c.BinlogCoordinateMode = BinlogCoordinateFilePosition + case BinlogCoordinateFilePosition, BinlogCoordinateGTID: + // valid + default: + return fmt.Errorf("invalid BinlogCoordinateMode %q, must be %q or %q", c.BinlogCoordinateMode, BinlogCoordinateFilePosition, BinlogCoordinateGTID) + } + if c.VerifierType == VerifierTypeIterative { if err := c.IterativeVerifierConfig.Validate(); err != nil { return fmt.Errorf("IterativeVerifierConfig invalid: %v", err) diff --git a/test/go/binlog_coordinate_test.go b/test/go/binlog_coordinate_test.go index 808e50fad..599954fc0 100644 --- a/test/go/binlog_coordinate_test.go +++ b/test/go/binlog_coordinate_test.go @@ -76,3 +76,87 @@ func TestBinlogCoordinate_UnmarshalTypedForm(t *testing.T) { assert.Equal(t, "mysql-bin.000050", decoded.Position().Name) assert.Equal(t, uint32(700), decoded.Position().Pos) } + +const ( + testGTIDSetA = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-57" + testGTIDSetB = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-100" +) + +func TestBinlogCoordinate_GTIDBasics(t *testing.T) { + coord := ghostferry.NewGTIDCoordinate(testGTIDSetA) + + assert.True(t, coord.IsGTID()) + assert.False(t, coord.IsFilePosition()) + assert.False(t, coord.IsZero()) + assert.Equal(t, "gtid:"+testGTIDSetA, coord.String()) + + set, err := coord.ParsedGTIDSet() + require.NoError(t, err) + assert.Equal(t, testGTIDSetA, set.String()) +} + +func TestBinlogCoordinate_GTIDEmptyIsZero(t *testing.T) { + coord := ghostferry.NewGTIDCoordinate("") + + assert.True(t, coord.IsGTID()) + assert.True(t, coord.IsZero()) +} + +func TestBinlogCoordinate_GTIDContains(t *testing.T) { + larger := ghostferry.NewGTIDCoordinate(testGTIDSetB) + smaller := ghostferry.NewGTIDCoordinate(testGTIDSetA) + + contains, err := larger.Contains(smaller) + require.NoError(t, err) + assert.True(t, contains) + + contains, err = smaller.Contains(larger) + require.NoError(t, err) + assert.False(t, contains) +} + +func TestBinlogCoordinate_ContainsRejectsFilePosition(t *testing.T) { + gtid := ghostferry.NewGTIDCoordinate(testGTIDSetA) + filePos := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4}) + + _, err := gtid.Contains(filePos) + assert.Error(t, err) +} + +func TestBinlogCoordinate_ParsedGTIDSetRejectsFilePosition(t *testing.T) { + filePos := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4}) + + _, err := filePos.ParsedGTIDSet() + assert.Error(t, err) +} + +func TestBinlogCoordinate_GTIDJSONRoundTrip(t *testing.T) { + coord := ghostferry.NewGTIDCoordinate(testGTIDSetA) + + data, err := json.Marshal(coord) + require.NoError(t, err) + + var decoded ghostferry.BinlogCoordinate + require.NoError(t, json.Unmarshal(data, &decoded)) + + assert.True(t, decoded.IsGTID()) + assert.Equal(t, testGTIDSetA, decoded.GTIDSet) +} + +func TestBinlogCoordinate_CompareRejectsMixedTypes(t *testing.T) { + gtid := ghostferry.NewGTIDCoordinate(testGTIDSetA) + filePos := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4}) + + assert.Panics(t, func() { + gtid.Compare(filePos) + }) +} + +func TestBinlogCoordinate_CompareUnsupportedForGTID(t *testing.T) { + a := ghostferry.NewGTIDCoordinate(testGTIDSetA) + b := ghostferry.NewGTIDCoordinate(testGTIDSetB) + + assert.Panics(t, func() { + a.Compare(b) + }) +} diff --git a/test/go/config_test.go b/test/go/config_test.go index 702e9be53..50b8d0f13 100644 --- a/test/go/config_test.go +++ b/test/go/config_test.go @@ -201,6 +201,25 @@ func (this *ConfigTestSuite) TestDefaultMarginalia() { this.Require().Equal(ghostferry.DefaultMarginalia, this.config.Target.Marginalia) } +func (this *ConfigTestSuite) TestBinlogCoordinateModeDefaultsToFilePosition() { + err := this.config.ValidateConfig() + this.Require().Nil(err) + this.Require().Equal(ghostferry.BinlogCoordinateFilePosition, this.config.BinlogCoordinateMode) +} + +func (this *ConfigTestSuite) TestBinlogCoordinateModeGTIDIsValid() { + this.config.BinlogCoordinateMode = ghostferry.BinlogCoordinateGTID + err := this.config.ValidateConfig() + this.Require().Nil(err) + this.Require().Equal(ghostferry.BinlogCoordinateGTID, this.config.BinlogCoordinateMode) +} + +func (this *ConfigTestSuite) TestBinlogCoordinateModeInvalidIsRejected() { + this.config.BinlogCoordinateMode = "banana" + err := this.config.ValidateConfig() + this.Require().EqualError(err, `invalid BinlogCoordinateMode "banana", must be "file_position" or "gtid"`) +} + func TestConfig(t *testing.T) { testhelpers.SetupTest() suite.Run(t, new(ConfigTestSuite)) diff --git a/utils.go b/utils.go index 12ee253ee..57caba89f 100644 --- a/utils.go +++ b/utils.go @@ -258,6 +258,54 @@ func CheckDbIsAReplica(db *sql.DB) (bool, error) { return isReadOnly, err } +// ReadExecutedGTIDSet reads @@GLOBAL.GTID_EXECUTED from the given database and +// returns it as a canonical GTID set string. The returned string may be empty +// if the server has executed no GTID-tagged transactions yet; that is not an +// error. +func ReadExecutedGTIDSet(db *sql.DB) (string, error) { + var gtidSet string + row := db.QueryRow("SELECT @@GLOBAL.GTID_EXECUTED") + if err := row.Scan(>idSet); err != nil { + return "", fmt.Errorf("reading @@GLOBAL.GTID_EXECUTED: %w", err) + } + return gtidSet, nil +} + +// ReadCurrentGTIDCoordinate reads the current executed GTID set and returns it +// as a GTID BinlogCoordinate. The raw string is validated by parsing it. +func ReadCurrentGTIDCoordinate(db *sql.DB) (BinlogCoordinate, error) { + gtidSet, err := ReadExecutedGTIDSet(db) + if err != nil { + return BinlogCoordinate{}, err + } + + // Validate that the string parses as a MySQL GTID set. An empty string is + // a valid (empty) set. + if _, err := mysql.ParseMysqlGTIDSet(gtidSet); err != nil { + return BinlogCoordinate{}, fmt.Errorf("parsing executed GTID set %q: %w", gtidSet, err) + } + + return NewGTIDCoordinate(gtidSet), nil +} + +// CheckServerGTIDModeEnabled verifies that the server has GTID mode enabled +// (@@GLOBAL.gtid_mode = ON). It returns an error describing the current value +// otherwise. This is intended to be called during startup validation when +// BinlogCoordinateMode is "gtid". +func CheckServerGTIDModeEnabled(db *sql.DB) error { + var gtidMode string + row := db.QueryRow("SELECT @@GLOBAL.gtid_mode") + if err := row.Scan(>idMode); err != nil { + return fmt.Errorf("reading @@GLOBAL.gtid_mode: %w", err) + } + + if !strings.EqualFold(gtidMode, "ON") { + return fmt.Errorf("gtid_mode must be ON for GTID binlog coordinate mode, but was %q", gtidMode) + } + + return nil +} + func ConvertTableColumnsToStrings(columns []schema.TableColumn) []string { out := make([]string, 0, len(columns)) for _, column := range columns { From 697395253302a2c5aeb914d0d08523fd4f91bd17 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Wed, 22 Jul 2026 02:08:54 +0200 Subject: [PATCH 2/3] Unify coordinate reachability into HasReached Replace the separate Compare (file/position) and Contains (GTID) methods with a single HasReached(target) that answers "has this coordinate reached or passed the target?" and hides the representation-specific mechanics: position comparison for file/position, set containment for GTID. Callers no longer branch on coordinate type to ask the finish-line question. --- binlog_coordinate.go | 71 ++++++++++++------------------- test/go/binlog_coordinate_test.go | 56 +++++++++++------------- 2 files changed, 52 insertions(+), 75 deletions(-) diff --git a/binlog_coordinate.go b/binlog_coordinate.go index 84bf486b1..480bfffa2 100644 --- a/binlog_coordinate.go +++ b/binlog_coordinate.go @@ -135,61 +135,44 @@ func (c BinlogCoordinate) IsZero() bool { } } -// Compare orders two coordinates of the same type. +// HasReached reports whether this coordinate has reached or passed target, +// i.e. whether a stream positioned at c has already covered everything up to +// target. This is the single "have we crossed the finish line?" question used +// for stop and catchup conditions; it hides the representation-specific +// mechanics from callers. // -// It returns -1, 0 or 1 following the semantics of mysql.Position.Compare for -// file/position coordinates. Comparing coordinates of differing types is a -// programmer error and panics, because there is no meaningful total ordering -// across representations. Callers that may receive mixed types should branch on -// the coordinate type first. +// - For file/position coordinates it is a position comparison (c >= target). +// - For GTID coordinates it is set containment (c's set contains target's +// set), because GTID sets do not form a total order. // -// GTID coordinates do not form a total order (a set can contain another, be -// contained by it, both, or neither). Compare therefore does not support GTID -// coordinates; use Contains for GTID reachability checks instead. This mirrors -// the reality that "which GTID set is further ahead" is not well defined. -func (c BinlogCoordinate) Compare(other BinlogCoordinate) int { - if c.resolvedType() != other.resolvedType() { - panic(fmt.Sprintf( +// Both coordinates must be the same type; a mismatch returns an error rather +// than guessing across representations. +func (c BinlogCoordinate) HasReached(target BinlogCoordinate) (bool, error) { + if c.resolvedType() != target.resolvedType() { + return false, fmt.Errorf( "cannot compare binlog coordinates of different types: %q vs %q", - c.resolvedType(), other.resolvedType(), - )) + c.resolvedType(), target.resolvedType(), + ) } switch c.resolvedType() { case BinlogCoordinateFilePosition: - return c.FilePosition.Compare(other.FilePosition) + return c.FilePosition.Compare(target.FilePosition) >= 0, nil + case BinlogCoordinateGTID: + mine, err := c.ParsedGTIDSet() + if err != nil { + return false, fmt.Errorf("parsing GTID set %q: %w", c.GTIDSet, err) + } + theirs, err := target.ParsedGTIDSet() + if err != nil { + return false, fmt.Errorf("parsing GTID set %q: %w", target.GTIDSet, err) + } + return mine.Contain(theirs), nil default: - panic(fmt.Sprintf("Compare not supported for binlog coordinate type %q; use Contains for GTID", c.resolvedType())) + return false, fmt.Errorf("HasReached not supported for binlog coordinate type %q", c.resolvedType()) } } -// Contains reports whether this GTID coordinate's set fully contains the other -// GTID coordinate's set. This is the correct "have we reached/passed" check for -// GTID based stop and catchup conditions. -// -// Both coordinates must be GTID coordinates. Calling Contains on file/position -// coordinates returns an error, since containment is not the file/position -// reachability model (Compare is). -func (c BinlogCoordinate) Contains(other BinlogCoordinate) (bool, error) { - if c.resolvedType() != BinlogCoordinateGTID || other.resolvedType() != BinlogCoordinateGTID { - return false, fmt.Errorf( - "Contains is only defined for GTID coordinates, got %q and %q", - c.resolvedType(), other.resolvedType(), - ) - } - - mine, err := c.ParsedGTIDSet() - if err != nil { - return false, fmt.Errorf("parsing GTID set %q: %w", c.GTIDSet, err) - } - theirs, err := other.ParsedGTIDSet() - if err != nil { - return false, fmt.Errorf("parsing GTID set %q: %w", other.GTIDSet, err) - } - - return mine.Contain(theirs), nil -} - // String returns a human readable representation for logs and status output. func (c BinlogCoordinate) String() string { switch c.resolvedType() { diff --git a/test/go/binlog_coordinate_test.go b/test/go/binlog_coordinate_test.go index 599954fc0..ed5be080d 100644 --- a/test/go/binlog_coordinate_test.go +++ b/test/go/binlog_coordinate_test.go @@ -28,15 +28,23 @@ func TestBinlogCoordinate_ZeroValueIsFilePosition(t *testing.T) { assert.Equal(t, mysql.Position{}, coord.Position()) } -func TestBinlogCoordinate_Compare(t *testing.T) { +func TestBinlogCoordinate_HasReachedFilePosition(t *testing.T) { a := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 10}) b := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 20}) - c := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000002", Pos: 5}) - assert.Equal(t, -1, a.Compare(b)) - assert.Equal(t, 1, b.Compare(a)) - assert.Equal(t, 0, a.Compare(a)) - assert.Equal(t, -1, b.Compare(c)) + // b is ahead of a: b has reached a, a has not reached b. + reached, err := b.HasReached(a) + require.NoError(t, err) + assert.True(t, reached) + + reached, err = a.HasReached(b) + require.NoError(t, err) + assert.False(t, reached) + + // A coordinate has reached itself (>=). + reached, err = a.HasReached(a) + require.NoError(t, err) + assert.True(t, reached) } func TestBinlogCoordinate_JSONRoundTrip(t *testing.T) { @@ -102,24 +110,28 @@ func TestBinlogCoordinate_GTIDEmptyIsZero(t *testing.T) { assert.True(t, coord.IsZero()) } -func TestBinlogCoordinate_GTIDContains(t *testing.T) { +func TestBinlogCoordinate_HasReachedGTID(t *testing.T) { larger := ghostferry.NewGTIDCoordinate(testGTIDSetB) smaller := ghostferry.NewGTIDCoordinate(testGTIDSetA) - contains, err := larger.Contains(smaller) + // The larger set contains the smaller, so it has reached it. + reached, err := larger.HasReached(smaller) require.NoError(t, err) - assert.True(t, contains) + assert.True(t, reached) - contains, err = smaller.Contains(larger) + reached, err = smaller.HasReached(larger) require.NoError(t, err) - assert.False(t, contains) + assert.False(t, reached) } -func TestBinlogCoordinate_ContainsRejectsFilePosition(t *testing.T) { +func TestBinlogCoordinate_HasReachedRejectsMixedTypes(t *testing.T) { gtid := ghostferry.NewGTIDCoordinate(testGTIDSetA) filePos := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4}) - _, err := gtid.Contains(filePos) + _, err := gtid.HasReached(filePos) + assert.Error(t, err) + + _, err = filePos.HasReached(gtid) assert.Error(t, err) } @@ -142,21 +154,3 @@ func TestBinlogCoordinate_GTIDJSONRoundTrip(t *testing.T) { assert.True(t, decoded.IsGTID()) assert.Equal(t, testGTIDSetA, decoded.GTIDSet) } - -func TestBinlogCoordinate_CompareRejectsMixedTypes(t *testing.T) { - gtid := ghostferry.NewGTIDCoordinate(testGTIDSetA) - filePos := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4}) - - assert.Panics(t, func() { - gtid.Compare(filePos) - }) -} - -func TestBinlogCoordinate_CompareUnsupportedForGTID(t *testing.T) { - a := ghostferry.NewGTIDCoordinate(testGTIDSetA) - b := ghostferry.NewGTIDCoordinate(testGTIDSetB) - - assert.Panics(t, func() { - a.Compare(b) - }) -} From 03385a943387d87c1a44f3b91d18586074e3785e Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Wed, 22 Jul 2026 20:42:06 +0200 Subject: [PATCH 3/3] Cache parsed GTID set to avoid re-parsing on hot paths Add NewGTIDCoordinateFromSet, which clones and caches the parsed mysql.GTIDSet, and make HasReached/ParsedGTIDSet reuse the cache. This removes the per-event serialize+reparse in the stop-condition loop while keeping the value type JSON-stable (the cache is unexported and not serialized). ParsedGTIDSet still returns a mutable clone. --- binlog_coordinate.go | 48 ++++++++++++++++++++++++++++--- test/go/binlog_coordinate_test.go | 32 +++++++++++++++++++++ 2 files changed, 76 insertions(+), 4 deletions(-) diff --git a/binlog_coordinate.go b/binlog_coordinate.go index 480bfffa2..0d968ec41 100644 --- a/binlog_coordinate.go +++ b/binlog_coordinate.go @@ -55,6 +55,13 @@ type BinlogCoordinate struct { // GTIDs" which is distinct from a nil/unset coordinate; callers that need // that distinction should check Type and IsZero together. GTIDSet string + + // parsedGTIDSet is an optional cache of the parsed GTIDSet. It is populated + // by NewGTIDCoordinateFromSet (and lazily by parsedSet) so that hot paths + // such as the stop-condition check do not re-parse the canonical string on + // every binlog event. It is never mutated in place; accessors that hand a + // set to callers clone it. It is intentionally excluded from JSON. + parsedGTIDSet mysql.GTIDSet } // NewFilePositionCoordinate wraps a mysql.Position in a BinlogCoordinate. @@ -75,6 +82,23 @@ func NewGTIDCoordinate(gtidSet string) BinlogCoordinate { } } +// NewGTIDCoordinateFromSet builds a GTID coordinate from an already-parsed +// mysql.GTIDSet. The set is cloned so the coordinate does not alias (or later +// mutate) the caller's set, and the clone is cached to avoid re-parsing on hot +// paths. Prefer this over NewGTIDCoordinate when a parsed set is already in +// hand (e.g. from go-mysql event tracking). +func NewGTIDCoordinateFromSet(set mysql.GTIDSet) BinlogCoordinate { + if set == nil { + return BinlogCoordinate{Type: BinlogCoordinateGTID} + } + cloned := set.Clone() + return BinlogCoordinate{ + Type: BinlogCoordinateGTID, + GTIDSet: cloned.String(), + parsedGTIDSet: cloned, + } +} + // resolvedType returns the effective type, treating the empty string as // file/position for backwards compatibility with older serialized state and // zero-valued coordinates. @@ -109,10 +133,24 @@ func (c BinlogCoordinate) Position() mysql.Position { // // It is valid to call this only for GTID coordinates. A fresh mysql.GTIDSet is // returned on each call so callers may mutate it freely without affecting the -// coordinate. +// coordinate (the internal cache, if any, is cloned before returning). func (c BinlogCoordinate) ParsedGTIDSet() (mysql.GTIDSet, error) { + set, err := c.parsedSet() + if err != nil { + return nil, err + } + return set.Clone(), nil +} + +// parsedSet returns the parsed GTID set, using the cache when present. The +// returned set MUST NOT be mutated by callers; use ParsedGTIDSet for a +// mutable clone. This exists so hot paths (HasReached) avoid re-parsing. +func (c BinlogCoordinate) parsedSet() (mysql.GTIDSet, error) { if c.resolvedType() != BinlogCoordinateGTID { - return nil, fmt.Errorf("ParsedGTIDSet called on non-GTID coordinate of type %q", c.resolvedType()) + return nil, fmt.Errorf("parsedSet called on non-GTID coordinate of type %q", c.resolvedType()) + } + if c.parsedGTIDSet != nil { + return c.parsedGTIDSet, nil } return mysql.ParseMysqlGTIDSet(c.GTIDSet) } @@ -159,11 +197,13 @@ func (c BinlogCoordinate) HasReached(target BinlogCoordinate) (bool, error) { case BinlogCoordinateFilePosition: return c.FilePosition.Compare(target.FilePosition) >= 0, nil case BinlogCoordinateGTID: - mine, err := c.ParsedGTIDSet() + // Use the cached parsed sets (read-only) to avoid re-parsing on hot + // paths such as the per-event stop check. Contain does not mutate. + mine, err := c.parsedSet() if err != nil { return false, fmt.Errorf("parsing GTID set %q: %w", c.GTIDSet, err) } - theirs, err := target.ParsedGTIDSet() + theirs, err := target.parsedSet() if err != nil { return false, fmt.Errorf("parsing GTID set %q: %w", target.GTIDSet, err) } diff --git a/test/go/binlog_coordinate_test.go b/test/go/binlog_coordinate_test.go index ed5be080d..e732de228 100644 --- a/test/go/binlog_coordinate_test.go +++ b/test/go/binlog_coordinate_test.go @@ -110,6 +110,38 @@ func TestBinlogCoordinate_GTIDEmptyIsZero(t *testing.T) { assert.True(t, coord.IsZero()) } +func TestBinlogCoordinate_NewGTIDCoordinateFromSet(t *testing.T) { + set, err := mysql.ParseMysqlGTIDSet(testGTIDSetA) + require.NoError(t, err) + + coord := ghostferry.NewGTIDCoordinateFromSet(set) + assert.True(t, coord.IsGTID()) + assert.Equal(t, testGTIDSetA, coord.GTIDSet) + + // Mutating the caller's set must not affect the coordinate (it clones). + other, err := mysql.ParseMysqlGTIDSet(testGTIDSetB) + require.NoError(t, err) + require.NoError(t, set.(*mysql.MysqlGTIDSet).Add(*other.(*mysql.MysqlGTIDSet))) + assert.Equal(t, testGTIDSetA, coord.GTIDSet, "coordinate must not alias the caller's set") + + // The parsed-set cache is not serialized; JSON only carries the string. + data, err := json.Marshal(coord) + require.NoError(t, err) + assert.NotContains(t, string(data), "parsedGTIDSet") + + var decoded ghostferry.BinlogCoordinate + require.NoError(t, json.Unmarshal(data, &decoded)) + reached, err := decoded.HasReached(ghostferry.NewGTIDCoordinate(testGTIDSetA)) + require.NoError(t, err) + assert.True(t, reached) +} + +func TestBinlogCoordinate_NewGTIDCoordinateFromNilSet(t *testing.T) { + coord := ghostferry.NewGTIDCoordinateFromSet(nil) + assert.True(t, coord.IsGTID()) + assert.True(t, coord.IsZero()) +} + func TestBinlogCoordinate_HasReachedGTID(t *testing.T) { larger := ghostferry.NewGTIDCoordinate(testGTIDSetB) smaller := ghostferry.NewGTIDCoordinate(testGTIDSetA)