diff --git a/binlog_coordinate.go b/binlog_coordinate.go new file mode 100644 index 00000000..00728064 --- /dev/null +++ b/binlog_coordinate.go @@ -0,0 +1,178 @@ +package ghostferry + +import ( + "encoding/json" + "fmt" + + "github.com/go-mysql-org/go-mysql/mysql" +) + +// BinlogCoordinateType identifies how a binlog position is expressed. +// +// This is the seam that lets Ghostferry evolve from file/position based +// replication coordinates towards GTID based coordinates without forcing every +// consumer to know which representation is in use. In this first iteration only +// the file/position representation is implemented; the GTID representation is +// reserved so that state serialized today remains forward compatible. +type BinlogCoordinateType string + +const ( + // BinlogCoordinateFilePosition is the classic (file, position) coordinate. + BinlogCoordinateFilePosition BinlogCoordinateType = "file_position" + + // BinlogCoordinateGTID is reserved for the future GTID based coordinate. + // It is intentionally defined now so that the serialization format and the + // strategy dispatch have a stable name to target. + BinlogCoordinateGTID BinlogCoordinateType = "gtid" +) + +// 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. +// +// 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. +type BinlogCoordinate struct { + // Type selects the active representation. An empty Type is treated as + // BinlogCoordinateFilePosition for backwards compatibility. + Type BinlogCoordinateType + + // FilePosition holds the coordinate when Type is BinlogCoordinateFilePosition. + FilePosition mysql.Position +} + +// NewFilePositionCoordinate wraps a mysql.Position in a BinlogCoordinate. +func NewFilePositionCoordinate(pos mysql.Position) BinlogCoordinate { + return BinlogCoordinate{ + Type: BinlogCoordinateFilePosition, + FilePosition: pos, + } +} + +// resolvedType returns the effective type, treating the empty string as +// file/position for backwards compatibility with older serialized state and +// zero-valued coordinates. +func (c BinlogCoordinate) resolvedType() BinlogCoordinateType { + if c.Type == "" { + return BinlogCoordinateFilePosition + } + return c.Type +} + +// IsFilePosition reports whether this coordinate is a file/position coordinate. +func (c BinlogCoordinate) IsFilePosition() bool { + return c.resolvedType() == BinlogCoordinateFilePosition +} + +// Position returns the underlying mysql.Position. +// +// It is valid to call this only for file/position coordinates. This accessor +// exists to keep the migration incremental: consumers that still require a +// concrete mysql.Position can obtain it here while the surrounding plumbing is +// converted to BinlogCoordinate. +func (c BinlogCoordinate) Position() mysql.Position { + return c.FilePosition +} + +// IsZero reports whether the coordinate carries no meaningful position. This +// mirrors the previous use of an empty mysql.Position as the "unset" sentinel. +func (c BinlogCoordinate) IsZero() bool { + switch c.resolvedType() { + case BinlogCoordinateFilePosition: + return c.FilePosition == (mysql.Position{}) + default: + return false + } +} + +// Compare orders two coordinates of the same type. +// +// 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. +func (c BinlogCoordinate) Compare(other BinlogCoordinate) int { + if c.resolvedType() != other.resolvedType() { + panic(fmt.Sprintf( + "cannot compare binlog coordinates of different types: %q vs %q", + c.resolvedType(), other.resolvedType(), + )) + } + + switch c.resolvedType() { + case BinlogCoordinateFilePosition: + return c.FilePosition.Compare(other.FilePosition) + default: + panic(fmt.Sprintf("comparison not implemented for binlog coordinate type %q", c.resolvedType())) + } +} + +// String returns a human readable representation for logs and status output. +func (c BinlogCoordinate) String() string { + switch c.resolvedType() { + case BinlogCoordinateFilePosition: + return fmt.Sprintf("%s:%d", c.FilePosition.Name, c.FilePosition.Pos) + default: + return fmt.Sprintf("", c.resolvedType()) + } +} + +// serializedBinlogCoordinate is the on-disk / on-wire shape of a +// BinlogCoordinate. It is deliberately explicit and self-describing so that a +// future GTID variant can be added as additional fields without breaking +// existing readers. +type serializedBinlogCoordinate struct { + Type BinlogCoordinateType `json:"Type"` + FilePosition *mysql.Position `json:"FilePosition,omitempty"` +} + +// MarshalJSON implements json.Marshaler. +func (c BinlogCoordinate) MarshalJSON() ([]byte, error) { + out := serializedBinlogCoordinate{Type: c.resolvedType()} + + switch c.resolvedType() { + case BinlogCoordinateFilePosition: + pos := c.FilePosition + out.FilePosition = &pos + default: + return nil, fmt.Errorf("cannot marshal binlog coordinate of type %q", c.resolvedType()) + } + + return json.Marshal(out) +} + +// UnmarshalJSON implements json.Unmarshaler. +// +// It accepts both the new self-describing form and, for defensive +// compatibility, a bare mysql.Position object of the form {"Name":...,"Pos":...}. +func (c *BinlogCoordinate) UnmarshalJSON(data []byte) error { + var typed serializedBinlogCoordinate + if err := json.Unmarshal(data, &typed); err == nil && typed.Type != "" { + c.Type = typed.Type + switch typed.Type { + case BinlogCoordinateFilePosition: + if typed.FilePosition != nil { + c.FilePosition = *typed.FilePosition + } else { + c.FilePosition = mysql.Position{} + } + return nil + default: + return fmt.Errorf("cannot unmarshal binlog coordinate of type %q", typed.Type) + } + } + + // Fall back to a bare mysql.Position for older/simpler payloads. + var pos mysql.Position + if err := json.Unmarshal(data, &pos); err != nil { + return err + } + c.Type = BinlogCoordinateFilePosition + c.FilePosition = pos + return nil +} diff --git a/binlog_streamer.go b/binlog_streamer.go index e74d070a..d762370d 100644 --- a/binlog_streamer.go +++ b/binlog_streamer.go @@ -340,6 +340,12 @@ func (s *BinlogStreamer) GetLastStreamedBinlogPosition() mysql.Position { return s.lastStreamedBinlogPosition } +// GetLastStreamedBinlogCoordinate is the coordinate-typed counterpart of +// GetLastStreamedBinlogPosition. +func (s *BinlogStreamer) GetLastStreamedBinlogCoordinate() BinlogCoordinate { + return NewFilePositionCoordinate(s.lastStreamedBinlogPosition) +} + func (s *BinlogStreamer) IsAlmostCaughtUp() bool { return time.Now().Sub(s.lastProcessedEventTime) < caughtUpThreshold } diff --git a/dml_events.go b/dml_events.go index 7b8a00f1..d0a2df9c 100644 --- a/dml_events.go +++ b/dml_events.go @@ -79,6 +79,11 @@ type DMLEvent interface { PaginationKey() (string, error) BinlogPosition() mysql.Position ResumableBinlogPosition() mysql.Position + // BinlogCoordinate and ResumableBinlogCoordinate are the coordinate-typed + // counterparts of BinlogPosition and ResumableBinlogPosition. They are the + // forward-looking API used while Ghostferry migrates off raw mysql.Position. + BinlogCoordinate() BinlogCoordinate + ResumableBinlogCoordinate() BinlogCoordinate Annotation() (string, error) Timestamp() time.Time } @@ -112,6 +117,14 @@ func (e *DMLEventBase) ResumableBinlogPosition() mysql.Position { return e.resumablePos } +func (e *DMLEventBase) BinlogCoordinate() BinlogCoordinate { + return NewFilePositionCoordinate(e.pos) +} + +func (e *DMLEventBase) ResumableBinlogCoordinate() BinlogCoordinate { + return NewFilePositionCoordinate(e.resumablePos) +} + // Annotation will return the first prefixed comment on the SQL string, // or an error if the query attribute of the DMLEvent is not set func (e *DMLEventBase) Annotation() (string, error) { @@ -504,10 +517,10 @@ func Int64Value(value interface{}) (int64, bool) { // // This is specifically mentioned in the the below link: // -// When BINARY values are stored, they are right-padded with the pad value -// to the specified length. The pad value is 0x00 (the zero byte). Values -// are right-padded with 0x00 for inserts, and no trailing bytes are removed -// for retrievals. +// When BINARY values are stored, they are right-padded with the pad value +// to the specified length. The pad value is 0x00 (the zero byte). Values +// are right-padded with 0x00 for inserts, and no trailing bytes are removed +// for retrievals. // // ref: https://dev.mysql.com/doc/refman/5.7/en/binary-varbinary.html func appendEscapedString(buffer []byte, value string, rightPadToLengthWithZeroBytes int) []byte { diff --git a/state_tracker.go b/state_tracker.go index 9f97221d..6f974cb8 100644 --- a/state_tracker.go +++ b/state_tracker.go @@ -49,7 +49,7 @@ func (s *SerializableState) MarshalJSON() ([]byte, error) { LastSuccessfulPaginationKeys map[string]json.RawMessage *Alias }{ - Alias: (*Alias)(s), + Alias: (*Alias)(s), LastSuccessfulPaginationKeys: make(map[string]json.RawMessage), } @@ -106,6 +106,13 @@ func (s *SerializableState) MinSourceBinlogPosition() mysql.Position { } } +// MinSourceBinlogCoordinate returns the safe source resume coordinate. It is +// the coordinate-typed counterpart of MinSourceBinlogPosition and currently +// delegates to it, wrapping the result as a file/position coordinate. +func (s *SerializableState) MinSourceBinlogCoordinate() BinlogCoordinate { + return NewFilePositionCoordinate(s.MinSourceBinlogPosition()) +} + // For tracking the speed of the copy type PaginationKeyPositionLog struct { Position float64 @@ -193,6 +200,46 @@ func (s *StateTracker) UpdateLastResumableBinlogPositionForTargetVerifier(pos my s.lastStoredBinlogPositionForTargetVerifier = pos } +// Coordinate-based accessors and mutators. +// +// These are the forward-looking API. They currently delegate to the +// file/position storage above so behavior is unchanged, but they let callers be +// written in terms of BinlogCoordinate ahead of introducing a GTID coordinate +// mode. + +func (s *StateTracker) UpdateLastResumableSourceBinlogCoordinate(coord BinlogCoordinate) { + s.UpdateLastResumableSourceBinlogPosition(coord.Position()) +} + +func (s *StateTracker) UpdateLastResumableSourceBinlogCoordinateForInlineVerifier(coord BinlogCoordinate) { + s.UpdateLastResumableSourceBinlogPositionForInlineVerifier(coord.Position()) +} + +func (s *StateTracker) UpdateLastResumableBinlogCoordinateForTargetVerifier(coord BinlogCoordinate) { + s.UpdateLastResumableBinlogPositionForTargetVerifier(coord.Position()) +} + +func (s *StateTracker) LastResumableSourceBinlogCoordinate() BinlogCoordinate { + s.BinlogRWMutex.RLock() + defer s.BinlogRWMutex.RUnlock() + + return NewFilePositionCoordinate(s.lastWrittenBinlogPosition) +} + +func (s *StateTracker) LastResumableSourceBinlogCoordinateForInlineVerifier() BinlogCoordinate { + s.BinlogRWMutex.RLock() + defer s.BinlogRWMutex.RUnlock() + + return NewFilePositionCoordinate(s.lastStoredBinlogPositionForInlineVerifier) +} + +func (s *StateTracker) LastResumableBinlogCoordinateForTargetVerifier() BinlogCoordinate { + s.BinlogRWMutex.RLock() + defer s.BinlogRWMutex.RUnlock() + + return NewFilePositionCoordinate(s.lastStoredBinlogPositionForTargetVerifier) +} + func (s *StateTracker) UpdateLastSuccessfulPaginationKey(table string, paginationKey PaginationKey, rowStats RowStats) { s.CopyRWMutex.Lock() defer s.CopyRWMutex.Unlock() diff --git a/test/go/binlog_coordinate_test.go b/test/go/binlog_coordinate_test.go new file mode 100644 index 00000000..808e50fa --- /dev/null +++ b/test/go/binlog_coordinate_test.go @@ -0,0 +1,78 @@ +package test + +import ( + "encoding/json" + "testing" + + "github.com/Shopify/ghostferry" + "github.com/go-mysql-org/go-mysql/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestBinlogCoordinate_FilePositionBasics(t *testing.T) { + pos := mysql.Position{Name: "mysql-bin.000123", Pos: 456} + coord := ghostferry.NewFilePositionCoordinate(pos) + + assert.True(t, coord.IsFilePosition()) + assert.False(t, coord.IsZero()) + assert.Equal(t, pos, coord.Position()) + assert.Equal(t, "mysql-bin.000123:456", coord.String()) +} + +func TestBinlogCoordinate_ZeroValueIsFilePosition(t *testing.T) { + var coord ghostferry.BinlogCoordinate + + assert.True(t, coord.IsFilePosition()) + assert.True(t, coord.IsZero()) + assert.Equal(t, mysql.Position{}, coord.Position()) +} + +func TestBinlogCoordinate_Compare(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)) +} + +func TestBinlogCoordinate_JSONRoundTrip(t *testing.T) { + coord := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000777", Pos: 999}) + + data, err := json.Marshal(coord) + require.NoError(t, err) + + var decoded ghostferry.BinlogCoordinate + require.NoError(t, json.Unmarshal(data, &decoded)) + + assert.True(t, decoded.IsFilePosition()) + assert.Equal(t, coord.Position(), decoded.Position()) +} + +// TestBinlogCoordinate_UnmarshalBareMysqlPosition guards the backwards +// compatible decoding path: a bare {"Name":...,"Pos":...} object (the shape an +// older mysql.Position produced) must decode into a file/position coordinate. +func TestBinlogCoordinate_UnmarshalBareMysqlPosition(t *testing.T) { + raw := []byte(`{"Name":"mysql-bin.000042","Pos":314}`) + + var decoded ghostferry.BinlogCoordinate + require.NoError(t, json.Unmarshal(raw, &decoded)) + + assert.True(t, decoded.IsFilePosition()) + assert.Equal(t, "mysql-bin.000042", decoded.Position().Name) + assert.Equal(t, uint32(314), decoded.Position().Pos) +} + +func TestBinlogCoordinate_UnmarshalTypedForm(t *testing.T) { + raw := []byte(`{"Type":"file_position","FilePosition":{"Name":"mysql-bin.000050","Pos":700}}`) + + var decoded ghostferry.BinlogCoordinate + require.NoError(t, json.Unmarshal(raw, &decoded)) + + assert.True(t, decoded.IsFilePosition()) + assert.Equal(t, "mysql-bin.000050", decoded.Position().Name) + assert.Equal(t, uint32(700), decoded.Position().Pos) +} diff --git a/test/go/binlog_writer_test.go b/test/go/binlog_writer_test.go index 83b7ba08..b2908eaa 100644 --- a/test/go/binlog_writer_test.go +++ b/test/go/binlog_writer_test.go @@ -24,8 +24,14 @@ func (e *stubDMLEvent) NewValues() ghostferry.RowData { return nil } func (e *stubDMLEvent) PaginationKey() (string, error) { return "", nil } func (e *stubDMLEvent) BinlogPosition() mysql.Position { return mysql.Position{} } func (e *stubDMLEvent) ResumableBinlogPosition() mysql.Position { return mysql.Position{} } -func (e *stubDMLEvent) Annotation() (string, error) { return "", nil } -func (e *stubDMLEvent) Timestamp() time.Time { return time.Time{} } +func (e *stubDMLEvent) BinlogCoordinate() ghostferry.BinlogCoordinate { + return ghostferry.NewFilePositionCoordinate(mysql.Position{}) +} +func (e *stubDMLEvent) ResumableBinlogCoordinate() ghostferry.BinlogCoordinate { + return ghostferry.NewFilePositionCoordinate(mysql.Position{}) +} +func (e *stubDMLEvent) Annotation() (string, error) { return "", nil } +func (e *stubDMLEvent) Timestamp() time.Time { return time.Time{} } // TestBinlogWriterBufferBinlogEventsBeforeRun verifies that BufferBinlogEvents // does not block when called before Run() has started in its own goroutine.