Skip to content

Commit 56e27cc

Browse files
pfcoperezpfcoperez
authored
[DBI-829] Extended GTID e2e and UTs (#4493)
- Expanded unit tests for GTID decoding, including case description refactor. - Added e2e roundtrip gtid test. - Re-just test matrix in flow.yaml (left mariadb-pos in to test the failure of the newly added tests) Follows: #4478 Closes: https://linear.app/clickhouse/issue/DBI-829
1 parent d1b3142 commit 56e27cc

4 files changed

Lines changed: 243 additions & 63 deletions

File tree

‎.env.example‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ CI_MARIADB_PORT=3308
1818
CI_MYSQL_ROOT_PASSWORD=cipass
1919
CI_MYSQL_GTID_VERSION=mysql-gtid
2020
CI_MYSQL_POS_VERSION=mysql-pos
21-
CI_MARIADB_VERSION=maria
21+
CI_MARIADB_VERSION=maria-gtid
2222

2323
CI_MYSQL_VERSION=mysql-gtid
2424
CI_MYSQL_PORT=3306 # Default is gtid

‎.github/workflows/flow.yml‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,8 +36,7 @@ jobs:
3636
db-version: [
3737
{pg: 16, mysql: 'mysql-gtid', mongo: '6.0', ch: 'lts'},
3838
{pg: 17, mysql: 'mysql-pos', mongo: '7.0', ch: 'stable'},
39-
{pg: 18, mysql: 'maria-pos', mongo: '8.0', ch: 'latest'},
40-
{pg: 18, mysql: 'maria-gtid', mongo: '8.0', ch: 'stable'},
39+
{pg: 18, mysql: 'maria-gtid', mongo: '8.0', ch: 'latest'},
4140
]
4241
# Per-version container settings consumed by the "MySQL" step, keyed by the matrix
4342
# db-version.mysql value. Grouped under "mysql" since these are MySQL/MariaDB families;

‎flow/connectors/mysql/gtid_test.go‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
package connmysql
2+
3+
import (
4+
"fmt"
5+
"testing"
6+
7+
"github.com/go-mysql-org/go-mysql/mysql"
8+
"github.com/stretchr/testify/require"
9+
10+
"github.com/PeerDB-io/peerdb/flow/generated/protos"
11+
)
12+
13+
func TestMySQLGetMasterGTIDSet(t *testing.T) {
14+
t.Parallel()
15+
ctx := t.Context()
16+
connector := newTestConnector(t, ctx)
17+
18+
gtidEnabled := connector.config.ReplicationMechanism == protos.MySqlReplicationMechanism_MYSQL_GTID
19+
20+
if !gtidEnabled {
21+
_, err := connector.GetMasterGTIDSet(ctx)
22+
require.ErrorContains(t, err, "gtid mode is not enabled")
23+
return
24+
}
25+
26+
// commit a transaction
27+
dbName := testDBName("gtidcheck")
28+
createTestDB(t, ctx, connector, dbName)
29+
table := fmt.Sprintf("`%s`.gtidcheck", dbName)
30+
_, err := connector.Execute(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY)", table))
31+
require.NoError(t, err)
32+
_, err = connector.Execute(ctx, fmt.Sprintf("INSERT INTO %s (id) VALUES (1)", table))
33+
require.NoError(t, err)
34+
35+
// confirm that the transaction left a non-empty GTID set
36+
gset, err := connector.GetMasterGTIDSet(ctx)
37+
require.NoError(t, err)
38+
require.NotNil(t, gset)
39+
require.False(t, gset.IsEmpty(), "a committed transaction should leave a non-empty GTID set")
40+
41+
// confirm the tested flavor generates the expected concrete GTID set type
42+
if connector.config.Flavor == protos.MySqlFlavor_MYSQL_MARIA {
43+
require.IsType(t, &mysql.MariadbGTIDSet{}, gset)
44+
} else {
45+
require.IsType(t, &mysql.MysqlGTIDSet{}, gset)
46+
}
47+
48+
// confirm that the GTID set round-trips through the flavor-aware parser
49+
reparsed, err := mysql.ParseGTIDSet(connector.Flavor(), gset.String())
50+
require.NoError(t, err)
51+
require.True(t, reparsed.Equal(gset), "re-parsed GTID set %q should equal the original %q", reparsed.String(), gset.String())
52+
}

‎flow/connectors/mysql/replication_mechanism_test.go‎

Lines changed: 189 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -9,68 +9,197 @@ import (
99
"github.com/PeerDB-io/peerdb/flow/generated/protos"
1010
)
1111

12+
// parseOffsetTestCase describes the validation case, separating gtid checks from fpos checks
13+
// ussing `offsetExpecation` interface.
14+
type parseOffsetTestCase struct {
15+
name string
16+
flavor string
17+
offsetText string
18+
19+
// If non-empty, parsing is expected to fail and the error must contain this substring.
20+
maybeErrContains string
21+
22+
expect offsetExpectation
23+
}
24+
25+
// offsetExpectation owns the mechanism-specific success assertions> gtid vs fpos.
26+
type offsetExpectation interface {
27+
// wantMechanism is the mechanism a successful parse must resolve to.
28+
wantMechanism() protos.MySqlReplicationMechanism
29+
// verifyParse asserts the parsed payload.
30+
verifyParse(t *testing.T, parsed parsedReplicationOffset)
31+
}
32+
33+
type gtidExpectation struct {
34+
gtidString string
35+
gtidEmpty bool
36+
}
37+
38+
func (gtidExpectation) wantMechanism() protos.MySqlReplicationMechanism {
39+
return protos.MySqlReplicationMechanism_MYSQL_GTID
40+
}
41+
42+
func (e gtidExpectation) verifyParse(t *testing.T, parsed parsedReplicationOffset) {
43+
t.Helper()
44+
require.Equal(t, e.wantMechanism().String(), parsed.mechanism)
45+
require.Equal(t, mysql.Position{}, parsed.pos)
46+
require.NotNil(t, parsed.gset)
47+
require.Equal(t, e.gtidString, parsed.gset.String())
48+
if e.gtidEmpty {
49+
require.True(t, parsed.gset.IsEmpty())
50+
}
51+
}
52+
53+
type fposExpectation struct {
54+
pos mysql.Position
55+
}
56+
57+
func (fposExpectation) wantMechanism() protos.MySqlReplicationMechanism {
58+
return protos.MySqlReplicationMechanism_MYSQL_FILEPOS
59+
}
60+
61+
func (e fposExpectation) verifyParse(t *testing.T, parsed parsedReplicationOffset) {
62+
t.Helper()
63+
require.Equal(t, e.wantMechanism().String(), parsed.mechanism)
64+
require.Nil(t, parsed.gset)
65+
require.Equal(t, e.pos, parsed.pos)
66+
}
67+
68+
// parseOffsetTestCases drives both TestParseReplicationOffsetText and
69+
// TestReplicationMechanismInUseFromOffsetText.
70+
var parseOffsetTestCases = []parseOffsetTestCase{
71+
{
72+
name: "mysql filepos",
73+
flavor: mysql.MySQLFlavor,
74+
offsetText: "!f:mysql-bin.000001,4d2",
75+
expect: fposExpectation{pos: mysql.Position{Name: "mysql-bin.000001", Pos: 0x4d2}},
76+
},
77+
{
78+
name: "mysql filepos missing comma",
79+
flavor: mysql.MySQLFlavor,
80+
offsetText: "!f:mysql-bin.000001",
81+
maybeErrContains: "no comma in file/pos offset",
82+
},
83+
{
84+
name: "mysql filepos invalid offset",
85+
flavor: mysql.MySQLFlavor,
86+
offsetText: "!f:mysql-bin.000001,zzz",
87+
maybeErrContains: "invalid offset in file/pos offset",
88+
},
89+
{
90+
name: "mysql gtid single uuid",
91+
flavor: mysql.MySQLFlavor,
92+
offsetText: "3E11FA47-71CA-11E1-9E33-C80AA9429562:23",
93+
expect: gtidExpectation{gtidString: "3e11fa47-71ca-11e1-9e33-c80aa9429562:23"},
94+
},
95+
{
96+
name: "mysql empty",
97+
flavor: mysql.MySQLFlavor,
98+
offsetText: "",
99+
expect: gtidExpectation{gtidEmpty: true},
100+
},
101+
{
102+
name: "mysql invalid gtid",
103+
flavor: mysql.MySQLFlavor,
104+
offsetText: "not-a-valid-offset",
105+
maybeErrContains: `failed to parse mysql offset text "not-a-valid-offset" as GTID set`,
106+
},
107+
{
108+
name: "mariadb gtid single domain",
109+
flavor: mysql.MariaDBFlavor,
110+
offsetText: "0-1-23",
111+
expect: gtidExpectation{gtidString: "0-1-23"},
112+
},
113+
{
114+
name: "mariadb gtid non-default domain and multi-digit fields",
115+
flavor: mysql.MariaDBFlavor,
116+
offsetText: "5-100-9999999",
117+
expect: gtidExpectation{gtidString: "5-100-9999999"},
118+
},
119+
{
120+
name: "mariadb gtid max uint64 sequence",
121+
flavor: mysql.MariaDBFlavor,
122+
offsetText: "5-100-18446744073709551615",
123+
expect: gtidExpectation{gtidString: "5-100-18446744073709551615"},
124+
},
125+
{
126+
name: "mariadb gtid multi-domain already sorted",
127+
flavor: mysql.MariaDBFlavor,
128+
offsetText: "0-1-23,1-2-45",
129+
expect: gtidExpectation{gtidString: "0-1-23,1-2-45"},
130+
},
131+
{
132+
name: "mariadb gtid multi-domain canonicalizes order",
133+
flavor: mysql.MariaDBFlavor,
134+
offsetText: "1-2-45,0-1-23",
135+
expect: gtidExpectation{gtidString: "0-1-23,1-2-45"},
136+
},
137+
{
138+
// A MariaDB position is one GTID per domain: the same domain twice collapses
139+
// to the newest sequence.
140+
name: "mariadb gtid same domain collapses to newest sequence",
141+
flavor: mysql.MariaDBFlavor,
142+
offsetText: "0-1-23,0-1-50",
143+
expect: gtidExpectation{gtidString: "0-1-50"},
144+
},
145+
{
146+
// Failover within a domain: server_id changes, the newest entry wins.
147+
name: "mariadb gtid same domain failover keeps newest server",
148+
flavor: mysql.MariaDBFlavor,
149+
offsetText: "0-1-23,0-2-50",
150+
expect: gtidExpectation{gtidString: "0-2-50"},
151+
},
152+
{
153+
name: "mariadb empty",
154+
flavor: mysql.MariaDBFlavor,
155+
offsetText: "",
156+
expect: gtidExpectation{gtidEmpty: true},
157+
},
158+
{
159+
// MySQL uuid:interval grammar must not parse under the MariaDB flavor.
160+
name: "mariadb rejects mysql-style uuid gtid",
161+
flavor: mysql.MariaDBFlavor,
162+
offsetText: "3E11FA47-71CA-11E1-9E33-C80AA9429562:23",
163+
maybeErrContains: "must domain-server-sequence",
164+
},
165+
{
166+
name: "mariadb gtid too few parts",
167+
flavor: mysql.MariaDBFlavor,
168+
offsetText: "0-1",
169+
maybeErrContains: "must domain-server-sequence",
170+
},
171+
{
172+
name: "mariadb gtid non-numeric domain",
173+
flavor: mysql.MariaDBFlavor,
174+
offsetText: "a-1-23",
175+
maybeErrContains: "invalid MariaDB GTID Domain ID",
176+
},
177+
}
178+
12179
func TestParseReplicationOffsetText(t *testing.T) {
13-
t.Run("filepos", func(t *testing.T) {
14-
parsedOffset, err := parseReplicationOffsetText("mysql", "!f:mysql-bin.000001,4d2")
15-
require.NoError(t, err)
16-
require.Equal(t, protos.MySqlReplicationMechanism_MYSQL_FILEPOS.String(), parsedOffset.mechanism)
17-
require.Nil(t, parsedOffset.gset)
18-
require.Equal(t, mysql.Position{Name: "mysql-bin.000001", Pos: 0x4d2}, parsedOffset.pos)
19-
})
20-
21-
t.Run("filepos missing comma", func(t *testing.T) {
22-
_, err := parseReplicationOffsetText("mysql", "!f:mysql-bin.000001")
23-
require.ErrorContains(t, err, "no comma in file/pos offset")
24-
})
25-
26-
t.Run("filepos invalid offset", func(t *testing.T) {
27-
_, err := parseReplicationOffsetText("mysql", "!f:mysql-bin.000001,zzz")
28-
require.ErrorContains(t, err, "invalid offset in file/pos offset")
29-
})
30-
31-
t.Run("gtid", func(t *testing.T) {
32-
parsedOffset, err := parseReplicationOffsetText("mysql", "3E11FA47-71CA-11E1-9E33-C80AA9429562:23")
33-
require.NoError(t, err)
34-
require.Equal(t, protos.MySqlReplicationMechanism_MYSQL_GTID.String(), parsedOffset.mechanism)
35-
require.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:23", parsedOffset.gset.String())
36-
require.Equal(t, mysql.Position{}, parsedOffset.pos)
37-
})
38-
39-
t.Run("empty", func(t *testing.T) {
40-
parsedOffset, err := parseReplicationOffsetText("mysql", "")
41-
require.NoError(t, err)
42-
require.Equal(t, protos.MySqlReplicationMechanism_MYSQL_GTID.String(), parsedOffset.mechanism)
43-
require.True(t, parsedOffset.gset.IsEmpty())
44-
require.Equal(t, mysql.Position{}, parsedOffset.pos)
45-
})
46-
47-
t.Run("invalid gtid", func(t *testing.T) {
48-
_, err := parseReplicationOffsetText("mysql", "not-a-valid-offset")
49-
require.ErrorContains(t, err, `failed to parse mysql offset text "not-a-valid-offset" as GTID set`)
50-
})
180+
for _, tc := range parseOffsetTestCases {
181+
t.Run(tc.name, func(t *testing.T) {
182+
parsed, err := parseReplicationOffsetText(tc.flavor, tc.offsetText)
183+
if tc.maybeErrContains != "" {
184+
require.ErrorContains(t, err, tc.maybeErrContains)
185+
return
186+
}
187+
require.NoError(t, err)
188+
tc.expect.verifyParse(t, parsed)
189+
})
190+
}
51191
}
52192

53193
func TestReplicationMechanismInUseFromOffsetText(t *testing.T) {
54-
t.Run("filepos", func(t *testing.T) {
55-
mechanism, err := replicationMechanismInUseFromOffsetText("mysql", "!f:mysql-bin.000001,4d2")
56-
require.NoError(t, err)
57-
require.Equal(t, protos.MySqlReplicationMechanism_MYSQL_FILEPOS.String(), mechanism)
58-
})
59-
60-
t.Run("gtid", func(t *testing.T) {
61-
mechanism, err := replicationMechanismInUseFromOffsetText("mysql", "3E11FA47-71CA-11E1-9E33-C80AA9429562:23")
62-
require.NoError(t, err)
63-
require.Equal(t, protos.MySqlReplicationMechanism_MYSQL_GTID.String(), mechanism)
64-
})
65-
66-
t.Run("empty", func(t *testing.T) {
67-
mechanism, err := replicationMechanismInUseFromOffsetText("mysql", "")
68-
require.NoError(t, err)
69-
require.Equal(t, protos.MySqlReplicationMechanism_MYSQL_GTID.String(), mechanism)
70-
})
71-
72-
t.Run("invalid", func(t *testing.T) {
73-
_, err := replicationMechanismInUseFromOffsetText("mysql", "not-a-valid-offset")
74-
require.ErrorContains(t, err, `failed to parse mysql offset text "not-a-valid-offset" as GTID set`)
75-
})
194+
for _, tc := range parseOffsetTestCases {
195+
t.Run(tc.name, func(t *testing.T) {
196+
mechanism, err := replicationMechanismInUseFromOffsetText(tc.flavor, tc.offsetText)
197+
if tc.maybeErrContains != "" {
198+
require.ErrorContains(t, err, tc.maybeErrContains)
199+
return
200+
}
201+
require.NoError(t, err)
202+
require.Equal(t, tc.expect.wantMechanism().String(), mechanism)
203+
})
204+
}
76205
}

0 commit comments

Comments
 (0)