Skip to content
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

go/vt/wrangler: reduce VReplicationExec calls when getting copy state #14375

Merged
merged 20 commits into from
Dec 28, 2023
Merged

go/vt/wrangler: reduce VReplicationExec calls when getting copy state #14375

Show file tree
Hide file tree
Changes from 16 commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
cfc8b01
go/vt/wrangler: reduce VReplicationExec calls when getting copy state
maxenglander Oct 26, 2023
4e1e17c
Update go/vt/wrangler/vexec.go
maxenglander Oct 26, 2023
88c1157
update tests to work with last commit
maxenglander Oct 26, 2023
7bd38c4
start working on making the same thing happen in workflow server
maxenglander Oct 27, 2023
d100a77
update tests
maxenglander Oct 27, 2023
e48bc6d
rm notes.txt :face-palm:
maxenglander Oct 27, 2023
b0bfaef
dont fetch copy state if no ids
maxenglander Oct 29, 2023
5ce0461
merge <- main
maxenglander Nov 9, 2023
34ee205
merge <- main
maxenglander Dec 4, 2023
ce4acf9
merge <- main
maxenglander Dec 5, 2023
2701fca
Update go/vt/vtctl/workflow/server.go
maxenglander Dec 19, 2023
ba11b6a
Update go/vt/wrangler/vexec.go
maxenglander Dec 19, 2023
88c5345
merge <- main
maxenglander Dec 19, 2023
bd914fc
fix bad make(map) call
maxenglander Dec 19, 2023
937e6ff
cr: update tests to have multiple streams
maxenglander Dec 19, 2023
e62ac3c
trigger ci
maxenglander Dec 19, 2023
fadcb44
merge <- main; cr: allocate once, use vterrors
maxenglander Dec 27, 2023
0b35e4e
cr: use errgroup
maxenglander Dec 27, 2023
d9c9c34
Merge branch 'main' into maxeng-vexec-getcopystate
maxenglander Dec 28, 2023
32a71f4
fix slice allocation err
maxenglander Dec 28, 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
3,632 changes: 1,821 additions & 1,811 deletions go/vt/proto/vtctldata/vtctldata.pb.go
View file
Open in desktop

Large diffs are not rendered by default.

32 changes: 30 additions & 2 deletions go/vt/proto/vtctldata/vtctldata_vtproto.pb.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

79 changes: 47 additions & 32 deletions go/vt/vtctl/vtctl_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,8 @@ func TestMoveTables(t *testing.T) {
shard := "0"
sourceKs := "sourceks"
targetKs := "targetks"
table := "customer"
table1 := "customer"
table2 := "customer_order"
wf := "testwf"
ksWf := fmt.Sprintf("%s.%s", targetKs, wf)
minTableSize := 16384 // a single 16KiB InnoDB page
Expand All @@ -159,16 +160,22 @@ func TestMoveTables(t *testing.T) {
defer env.close()
source := env.addTablet(100, sourceKs, shard, &topodatapb.KeyRange{}, topodatapb.TabletType_PRIMARY)
target := env.addTablet(200, targetKs, shard, &topodatapb.KeyRange{}, topodatapb.TabletType_PRIMARY)
sourceCol := fmt.Sprintf(`keyspace:"%s" shard:"%s" filter:{rules:{match:"%s" filter:"select * from %s"}}`,
sourceKs, shard, table, table)
sourceCol := fmt.Sprintf(`keyspace:"%s" shard:"%s" filter:{rules:{match:"%s" filter:"select * from %s"} rules:{match:"%s" filter:"select * from %s"}}`,
sourceKs, shard, table1, table1, table2, table2)
bls := &binlogdatapb.BinlogSource{
Keyspace: sourceKs,
Shard: shard,
Filter: &binlogdatapb.Filter{
Rules: []*binlogdatapb.Rule{{
Match: table,
Filter: fmt.Sprintf("select * from %s", table),
}},
Rules: []*binlogdatapb.Rule{
{
Match: table1,
Filter: fmt.Sprintf("select * from %s", table1),
},
{
Match: table2,
Filter: fmt.Sprintf("select * from %s", table2),
},
},
},
}
now := time.Now().UTC().Unix()
Expand Down Expand Up @@ -200,12 +207,13 @@ func TestMoveTables(t *testing.T) {
expectResults: func() {
env.tmc.setVRResults(
target.tablet,
fmt.Sprintf("select table_name, lastpk from _vt.copy_state where vrepl_id = %d and id in (select max(id) from _vt.copy_state where vrepl_id = %d group by vrepl_id, table_name)",
fmt.Sprintf("select vrepl_id, table_name, lastpk from _vt.copy_state where vrepl_id in (%d) and id in (select max(id) from _vt.copy_state where vrepl_id in (%d) group by vrepl_id, table_name)",
vrID, vrID),
sqltypes.MakeTestResult(sqltypes.MakeTestFields(
"table_name|lastpk",
"varchar|varbinary"),
fmt.Sprintf("%s|", table),
"vrepl_id|table_name|lastpk",
"int64|varchar|varbinary"),
fmt.Sprintf("%d|%s|", vrID, table1),
fmt.Sprintf("%d|%s|", vrID, table2),
),
)
env.tmc.setDBAResults(
Expand All @@ -215,7 +223,8 @@ func TestMoveTables(t *testing.T) {
sqltypes.MakeTestResult(sqltypes.MakeTestFields(
"table_name",
"varchar"),
table,
table1,
table2,
),
)
env.tmc.setVRResults(
Expand All @@ -231,26 +240,28 @@ func TestMoveTables(t *testing.T) {
)
env.tmc.setDBAResults(
target.tablet,
fmt.Sprintf("select table_name, table_rows, data_length from information_schema.tables where table_schema = 'vt_%s' and table_name in ('%s')",
targetKs, table),
fmt.Sprintf("select table_name, table_rows, data_length from information_schema.tables where table_schema = 'vt_%s' and table_name in ('%s','%s')",
targetKs, table1, table2),
sqltypes.MakeTestResult(sqltypes.MakeTestFields(
"table_name|table_rows|data_length",
"varchar|int64|int64"),
fmt.Sprintf("%s|0|%d", table, minTableSize),
fmt.Sprintf("%s|0|%d", table1, minTableSize),
fmt.Sprintf("%s|0|%d", table2, minTableSize),
),
)
env.tmc.setDBAResults(
source.tablet,
fmt.Sprintf("select table_name, table_rows, data_length from information_schema.tables where table_schema = 'vt_%s' and table_name in ('%s')",
sourceKs, table),
fmt.Sprintf("select table_name, table_rows, data_length from information_schema.tables where table_schema = 'vt_%s' and table_name in ('%s','%s')",
sourceKs, table1, table2),
sqltypes.MakeTestResult(sqltypes.MakeTestFields(
"table_name|table_rows|data_length",
"varchar|int64|int64"),
fmt.Sprintf("%s|10|%d", table, minTableSize),
fmt.Sprintf("%s|10|%d", table1, minTableSize),
fmt.Sprintf("%s|10|%d", table2, minTableSize),
),
)
},
want: fmt.Sprintf("\nCopy Progress (approx):\n\n\ncustomer: rows copied 0/10 (0%%), size copied 16384/16384 (100%%)\n\n\n\nThe following vreplication streams exist for workflow %s:\n\nid=%d on %s/%s-0000000%d: Status: Copying. VStream has not started.\n\n\n",
want: fmt.Sprintf("\nCopy Progress (approx):\n\n\ncustomer: rows copied 0/10 (0%%), size copied 16384/16384 (100%%)\ncustomer_order: rows copied 0/10 (0%%), size copied 16384/16384 (100%%)\n\n\n\nThe following vreplication streams exist for workflow %s:\n\nid=%d on %s/%s-0000000%d: Status: Copying. VStream has not started.\n\n\n",
ksWf, vrID, shard, env.cell, target.tablet.Alias.Uid),
},
{
Expand All @@ -260,12 +271,13 @@ func TestMoveTables(t *testing.T) {
expectResults: func() {
env.tmc.setVRResults(
target.tablet,
fmt.Sprintf("select table_name, lastpk from _vt.copy_state where vrepl_id = %d and id in (select max(id) from _vt.copy_state where vrepl_id = %d group by vrepl_id, table_name)",
fmt.Sprintf("select vrepl_id, table_name, lastpk from _vt.copy_state where vrepl_id in (%d) and id in (select max(id) from _vt.copy_state where vrepl_id in (%d) group by vrepl_id, table_name)",
vrID, vrID),
sqltypes.MakeTestResult(sqltypes.MakeTestFields(
"table_name|lastpk",
"varchar|varbinary"),
fmt.Sprintf("%s|", table),
"vrepl_id|table_name|lastpk",
"int64|varchar|varbinary"),
fmt.Sprintf("%d|%s|", vrID, table1),
fmt.Sprintf("%d|%s|", vrID, table2),
),
)
env.tmc.setDBAResults(
Expand All @@ -275,7 +287,8 @@ func TestMoveTables(t *testing.T) {
sqltypes.MakeTestResult(sqltypes.MakeTestFields(
"table_name",
"varchar"),
table,
table1,
table2,
),
)
env.tmc.setVRResults(
Expand All @@ -291,26 +304,28 @@ func TestMoveTables(t *testing.T) {
)
env.tmc.setDBAResults(
target.tablet,
fmt.Sprintf("select table_name, table_rows, data_length from information_schema.tables where table_schema = 'vt_%s' and table_name in ('%s')",
targetKs, table),
fmt.Sprintf("select table_name, table_rows, data_length from information_schema.tables where table_schema = 'vt_%s' and table_name in ('%s','%s')",
targetKs, table1, table2),
sqltypes.MakeTestResult(sqltypes.MakeTestFields(
"table_name|table_rows|data_length",
"varchar|int64|int64"),
fmt.Sprintf("%s|5|%d", table, minTableSize),
fmt.Sprintf("%s|5|%d", table1, minTableSize),
fmt.Sprintf("%s|5|%d", table2, minTableSize),
),
)
env.tmc.setDBAResults(
source.tablet,
fmt.Sprintf("select table_name, table_rows, data_length from information_schema.tables where table_schema = 'vt_%s' and table_name in ('%s')",
sourceKs, table),
fmt.Sprintf("select table_name, table_rows, data_length from information_schema.tables where table_schema = 'vt_%s' and table_name in ('%s','%s')",
sourceKs, table1, table2),
sqltypes.MakeTestResult(sqltypes.MakeTestFields(
"table_name|table_rows|data_length",
"varchar|int64|int64"),
fmt.Sprintf("%s|10|%d", table, minTableSize),
fmt.Sprintf("%s|10|%d", table1, minTableSize),
fmt.Sprintf("%s|10|%d", table2, minTableSize),
),
)
},
want: fmt.Sprintf("\nCopy Progress (approx):\n\n\ncustomer: rows copied 5/10 (50%%), size copied 16384/16384 (100%%)\n\n\n\nThe following vreplication streams exist for workflow %s:\n\nid=%d on %s/%s-0000000%d: Status: Error: Duplicate entry '6' for key 'customer.PRIMARY' (errno 1062) (sqlstate 23000) during query: insert into customer(customer_id,email) values (6,'mlord@planetscale.com').\n\n\n",
want: fmt.Sprintf("\nCopy Progress (approx):\n\n\ncustomer: rows copied 5/10 (50%%), size copied 16384/16384 (100%%)\ncustomer_order: rows copied 5/10 (50%%), size copied 16384/16384 (100%%)\n\n\n\nThe following vreplication streams exist for workflow %s:\n\nid=%d on %s/%s-0000000%d: Status: Error: Duplicate entry '6' for key 'customer.PRIMARY' (errno 1062) (sqlstate 23000) during query: insert into customer(customer_id,email) values (6,'mlord@planetscale.com').\n\n\n",
ksWf, vrID, shard, env.cell, target.tablet.Alias.Uid),
},
{
Expand All @@ -320,7 +335,7 @@ func TestMoveTables(t *testing.T) {
expectResults: func() {
env.tmc.setVRResults(
target.tablet,
fmt.Sprintf("select table_name, lastpk from _vt.copy_state where vrepl_id = %d and id in (select max(id) from _vt.copy_state where vrepl_id = %d group by vrepl_id, table_name)",
fmt.Sprintf("select vrepl_id, table_name, lastpk from _vt.copy_state where vrepl_id in (%d) and id in (select max(id) from _vt.copy_state where vrepl_id in (%d) group by vrepl_id, table_name)",
vrID, vrID),
&sqltypes.Result{},
)
Expand Down
2 changes: 1 addition & 1 deletion go/vt/vtctl/workflow/materializer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ const mzSelectFrozenQuery = "select 1 from _vt.vreplication where db_name='vt_ta
const mzCheckJournal = "/select val from _vt.resharding_journal where id="
const mzGetWorkflowStatusQuery = "select id, workflow, source, pos, stop_pos, max_replication_lag, state, db_name, time_updated, transaction_timestamp, message, tags, workflow_type, workflow_sub_type, time_heartbeat, defer_secondary_keys, component_throttled, time_throttled, rows_copied from _vt.vreplication where workflow = 'workflow' and db_name = 'vt_targetks'"
const mzGetCopyState = "select distinct table_name from _vt.copy_state cs, _vt.vreplication vr where vr.id = cs.vrepl_id and vr.id = 1"
const mzGetLatestCopyState = "select table_name, lastpk from _vt.copy_state where vrepl_id = 1 and id in (select max(id) from _vt.copy_state where vrepl_id = 1 group by vrepl_id, table_name)"
const mzGetLatestCopyState = "select vrepl_id, table_name, lastpk from _vt.copy_state where vrepl_id in (1) and id in (select max(id) from _vt.copy_state where vrepl_id in (1) group by vrepl_id, table_name)"
const insertPrefix = `/insert into _vt.vreplication\(workflow, source, pos, max_tps, max_replication_lag, cell, tablet_types, time_updated, transaction_timestamp, state, db_name, workflow_type, workflow_sub_type, defer_secondary_keys\) values `
const eol = "$"

Expand Down
Loading
Loading