Skip to content

Commit a60297b

Browse files
mattlordmattlord
authored
VReplication: Enforce consistent order for table copies and diffs (#15152)
Signed-off-by: Matt Lord <mattalord@gmail.com>
1 parent 1f728ea commit a60297b

File tree

13 files changed

+413
-36
lines changed

13 files changed

+413
-36
lines changed

‎go/protoutil/binlogsource.go

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
/*
2+
Copyright 2024 The Vitess Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package protoutil
18+
19+
import (
20+
"slices"
21+
"sort"
22+
"strings"
23+
24+
binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
25+
)
26+
27+
// SortBinlogSourceTables sorts the table related contents of the
28+
// BinlogSource struct lexicographically by table name in order to
29+
// produce consistent results.
30+
func SortBinlogSourceTables(bls *binlogdatapb.BinlogSource) {
31+
if bls == nil {
32+
return
33+
}
34+
35+
// Sort the tables by name to ensure a consistent order.
36+
slices.Sort(bls.Tables)
37+
38+
if bls.Filter == nil || len(bls.Filter.Rules) == 0 {
39+
return
40+
}
41+
sort.Slice(bls.Filter.Rules, func(i, j int) bool {
42+
// Exclude filters should logically be processed first.
43+
if bls.Filter.Rules[i].Filter == "exclude" && bls.Filter.Rules[j].Filter != "exclude" {
44+
return true
45+
}
46+
if bls.Filter.Rules[j].Filter == "exclude" && bls.Filter.Rules[i].Filter != "exclude" {
47+
return false
48+
}
49+
50+
// Remove preceding slash from the match string.
51+
// That is used when the filter is a regular expression.
52+
fi, _ := strings.CutPrefix(bls.Filter.Rules[i].Match, "/")
53+
fj, _ := strings.CutPrefix(bls.Filter.Rules[j].Match, "/")
54+
if fi != fj {
55+
return fi < fj
56+
}
57+
58+
return bls.Filter.Rules[i].Filter < bls.Filter.Rules[j].Filter
59+
})
60+
}

‎go/protoutil/binlogsource_test.go

Lines changed: 209 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
1+
/*
2+
Copyright 2024 The Vitess Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package protoutil
18+
19+
import (
20+
"testing"
21+
22+
"github.com/stretchr/testify/require"
23+
"google.golang.org/protobuf/proto"
24+
25+
binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
26+
)
27+
28+
func TestSortBinlogSourceTables(t *testing.T) {
29+
tests := []struct {
30+
name string
31+
inSource *binlogdatapb.BinlogSource
32+
outSource *binlogdatapb.BinlogSource
33+
}{
34+
{
35+
name: "Basic",
36+
inSource: &binlogdatapb.BinlogSource{
37+
Tables: []string{"wuts1", "atable", "1table", "ztable2", "table3"},
38+
Filter: &binlogdatapb.Filter{
39+
Rules: []*binlogdatapb.Rule{
40+
{
41+
Match: "ztable2",
42+
},
43+
{
44+
Match: "table3",
45+
},
46+
{
47+
Match: "/wuts",
48+
},
49+
{
50+
Match: "1table",
51+
Filter: "a",
52+
},
53+
{
54+
Match: "1table",
55+
},
56+
{
57+
Match: "atable",
58+
},
59+
},
60+
},
61+
},
62+
outSource: &binlogdatapb.BinlogSource{
63+
Tables: []string{"1table", "atable", "table3", "wuts1", "ztable2"},
64+
Filter: &binlogdatapb.Filter{
65+
Rules: []*binlogdatapb.Rule{
66+
{
67+
Match: "1table",
68+
},
69+
{
70+
Match: "1table",
71+
Filter: "a",
72+
},
73+
{
74+
Match: "atable",
75+
},
76+
{
77+
Match: "table3",
78+
},
79+
{
80+
Match: "/wuts",
81+
},
82+
{
83+
Match: "ztable2",
84+
},
85+
},
86+
},
87+
},
88+
},
89+
{
90+
name: "With excludes",
91+
inSource: &binlogdatapb.BinlogSource{
92+
Filter: &binlogdatapb.Filter{
93+
Rules: []*binlogdatapb.Rule{
94+
{
95+
Match: "./*",
96+
},
97+
{
98+
Match: "no4",
99+
Filter: "exclude",
100+
},
101+
{
102+
Match: "no2",
103+
Filter: "exclude",
104+
},
105+
{
106+
Match: "ztable2",
107+
},
108+
{
109+
Match: "atable2",
110+
},
111+
},
112+
},
113+
},
114+
outSource: &binlogdatapb.BinlogSource{
115+
Filter: &binlogdatapb.Filter{
116+
Rules: []*binlogdatapb.Rule{
117+
{
118+
Match: "no2",
119+
Filter: "exclude",
120+
},
121+
{
122+
Match: "no4",
123+
Filter: "exclude",
124+
},
125+
{
126+
Match: "./*",
127+
},
128+
{
129+
Match: "atable2",
130+
},
131+
{
132+
Match: "ztable2",
133+
},
134+
},
135+
},
136+
},
137+
},
138+
{
139+
name: "With excludes",
140+
inSource: &binlogdatapb.BinlogSource{
141+
Filter: &binlogdatapb.Filter{
142+
Rules: []*binlogdatapb.Rule{
143+
{
144+
Match: "no4",
145+
Filter: "exclude",
146+
},
147+
{
148+
Match: "no2",
149+
Filter: "exclude",
150+
},
151+
{
152+
Match: "./*",
153+
},
154+
},
155+
},
156+
},
157+
outSource: &binlogdatapb.BinlogSource{
158+
Filter: &binlogdatapb.Filter{
159+
Rules: []*binlogdatapb.Rule{
160+
{
161+
Match: "no2",
162+
Filter: "exclude",
163+
},
164+
{
165+
Match: "no4",
166+
Filter: "exclude",
167+
},
168+
{
169+
Match: "./*",
170+
},
171+
},
172+
},
173+
},
174+
},
175+
{
176+
name: "Nil",
177+
inSource: nil,
178+
outSource: nil,
179+
},
180+
{
181+
name: "No filter",
182+
inSource: &binlogdatapb.BinlogSource{
183+
Tables: []string{"wuts1", "atable", "1table", "ztable2", "table3"},
184+
Filter: nil,
185+
},
186+
outSource: &binlogdatapb.BinlogSource{
187+
Tables: []string{"1table", "atable", "table3", "wuts1", "ztable2"},
188+
Filter: nil,
189+
},
190+
},
191+
{
192+
name: "No filter rules",
193+
inSource: &binlogdatapb.BinlogSource{
194+
Tables: []string{"wuts1", "atable", "1table", "ztable2", "table3"},
195+
Filter: &binlogdatapb.Filter{},
196+
},
197+
outSource: &binlogdatapb.BinlogSource{
198+
Tables: []string{"1table", "atable", "table3", "wuts1", "ztable2"},
199+
Filter: &binlogdatapb.Filter{},
200+
},
201+
},
202+
}
203+
for _, tt := range tests {
204+
t.Run(tt.name, func(t *testing.T) {
205+
SortBinlogSourceTables(tt.inSource)
206+
require.True(t, proto.Equal(tt.inSource, tt.outSource), "got: %s, want: %s", tt.inSource.String(), tt.outSource.String())
207+
})
208+
}
209+
}

‎go/test/endtoend/vreplication/vdiff_helper_test.go

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -90,13 +90,14 @@ func doVtctlclientVDiff(t *testing.T, keyspace, workflow, cells string, want *ex
9090

9191
func waitForVDiff2ToComplete(t *testing.T, useVtctlclient bool, ksWorkflow, cells, uuid string, completedAtMin time.Time) *vdiffInfo {
9292
var info *vdiffInfo
93+
var jsonStr string
9394
first := true
9495
previousProgress := vdiff2.ProgressReport{}
9596
ch := make(chan bool)
9697
go func() {
9798
for {
9899
time.Sleep(vdiffStatusCheckInterval)
99-
_, jsonStr := performVDiff2Action(t, useVtctlclient, ksWorkflow, cells, "show", uuid, false)
100+
_, jsonStr = performVDiff2Action(t, useVtctlclient, ksWorkflow, cells, "show", uuid, false)
100101
info = getVDiffInfo(jsonStr)
101102
require.NotNil(t, info)
102103
if info.State == "completed" {
@@ -142,7 +143,7 @@ func waitForVDiff2ToComplete(t *testing.T, useVtctlclient bool, ksWorkflow, cell
142143
case <-ch:
143144
return info
144145
case <-time.After(vdiffTimeout):
145-
log.Errorf("VDiff never completed for UUID %s", uuid)
146+
log.Errorf("VDiff never completed for UUID %s. Latest output: %s", uuid, jsonStr)
146147
require.FailNow(t, fmt.Sprintf("VDiff never completed for UUID %s", uuid))
147148
return nil
148149
}

‎go/vt/binlog/binlogplayer/binlog_player.go

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,14 +34,13 @@ import (
3434
"time"
3535

3636
"github.com/spf13/pflag"
37-
38-
"vitess.io/vitess/go/mysql/replication"
39-
"vitess.io/vitess/go/mysql/sqlerror"
40-
4137
"google.golang.org/protobuf/proto"
4238

4339
"vitess.io/vitess/go/history"
4440
"vitess.io/vitess/go/mysql"
41+
"vitess.io/vitess/go/mysql/replication"
42+
"vitess.io/vitess/go/mysql/sqlerror"
43+
"vitess.io/vitess/go/protoutil"
4544
"vitess.io/vitess/go/sqltypes"
4645
"vitess.io/vitess/go/stats"
4746
"vitess.io/vitess/go/vt/log"
@@ -606,6 +605,7 @@ func ReadVRSettings(dbClient DBClient, uid int32) (VRSettings, error) {
606605
// the _vt.vreplication table.
607606
func CreateVReplication(workflow string, source *binlogdatapb.BinlogSource, position string, maxTPS, maxReplicationLag, timeUpdated int64, dbName string,
608607
workflowType binlogdatapb.VReplicationWorkflowType, workflowSubType binlogdatapb.VReplicationWorkflowSubType, deferSecondaryKeys bool) string {
608+
protoutil.SortBinlogSourceTables(source)
609609
return fmt.Sprintf("insert into _vt.vreplication "+
610610
"(workflow, source, pos, max_tps, max_replication_lag, time_updated, transaction_timestamp, state, db_name, workflow_type, workflow_sub_type, defer_secondary_keys) "+
611611
"values (%v, %v, %v, %v, %v, %v, 0, '%v', %v, %d, %d, %v)",
@@ -616,6 +616,7 @@ func CreateVReplication(workflow string, source *binlogdatapb.BinlogSource, posi
616616
// CreateVReplicationState returns a statement to create a stopped vreplication.
617617
func CreateVReplicationState(workflow string, source *binlogdatapb.BinlogSource, position string, state binlogdatapb.VReplicationWorkflowState, dbName string,
618618
workflowType binlogdatapb.VReplicationWorkflowType, workflowSubType binlogdatapb.VReplicationWorkflowSubType) string {
619+
protoutil.SortBinlogSourceTables(source)
619620
return fmt.Sprintf("insert into _vt.vreplication "+
620621
"(workflow, source, pos, max_tps, max_replication_lag, time_updated, transaction_timestamp, state, db_name, workflow_type, workflow_sub_type) "+
621622
"values (%v, %v, %v, %v, %v, %v, 0, '%v', %v, %d, %d)",

‎go/vt/vtctl/workflow/resharder.go

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"context"
2222
"errors"
2323
"fmt"
24+
"slices"
2425
"sync"
2526
"time"
2627

@@ -281,8 +282,8 @@ func (rs *resharder) createStreams(ctx context.Context) error {
281282

282283
ig := vreplication.NewInsertGenerator(binlogdatapb.VReplicationWorkflowState_Stopped, targetPrimary.DbName())
283284

284-
// copy excludeRules to prevent data race.
285-
copyExcludeRules := append([]*binlogdatapb.Rule(nil), excludeRules...)
285+
// Clone excludeRules to prevent data races.
286+
copyExcludeRules := slices.Clone(excludeRules)
286287
for _, source := range rs.sourceShards {
287288
if !key.KeyRangeIntersect(target.KeyRange, source.KeyRange) {
288289
continue

‎go/vt/vttablet/tabletmanager/rpc_vreplication.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
"google.golang.org/protobuf/encoding/prototext"
2424

2525
"vitess.io/vitess/go/constants/sidecar"
26+
"vitess.io/vitess/go/protoutil"
2627
"vitess.io/vitess/go/sqltypes"
2728
"vitess.io/vitess/go/textutil"
2829
"vitess.io/vitess/go/vt/discovery"
@@ -57,6 +58,7 @@ func (tm *TabletManager) CreateVReplicationWorkflow(ctx context.Context, req *ta
5758
}
5859
res := &sqltypes.Result{}
5960
for _, bls := range req.BinlogSource {
61+
protoutil.SortBinlogSourceTables(bls)
6062
source, err := prototext.Marshal(bls)
6163
if err != nil {
6264
return nil, err

0 commit comments

Comments
 (0)