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.

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

VDiff/OnlineDDL: add support for running VDiffs for OnlineDDL migrations #15546

Merged
merged 4 commits into from
Apr 8, 2024
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
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
Next Next commit
Allow VDiff for OnlineDDLs. Add e2e test
Signed-off-by: Rohit Nayak <rohit@planetscale.com>
  • Loading branch information
rohit-nayak-ps committed Apr 8, 2024
commit ced11cc0385a39b2905decac3fa52a9295fc4c53
11 changes: 11 additions & 0 deletions go/test/endtoend/cluster/vtctldclient_process.go
Original file line number Diff line number Diff line change
Expand Up @@ -298,3 +298,14 @@ func (vtctldclient *VtctldClientProcess) OnlineDDLShowRecent(Keyspace string) (r
"recent",
)
}

// OnlineDDLShow responds with recent schema migration list
func (vtctldclient *VtctldClientProcess) OnlineDDLShow(keyspace, workflow string) (result string, err error) {
return vtctldclient.ExecuteCommandWithOutput(
"OnlineDDL",
"show",
"--json",
keyspace,
workflow,
)
}
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved
4 changes: 2 additions & 2 deletions go/test/endtoend/vreplication/helper_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -358,7 +358,7 @@ func waitForWorkflowState(t *testing.T, vc *VitessCluster, ksWorkflow string, wa
log.Infof("Waiting for workflow %q to fully reach %q state", ksWorkflow, wantState)
for {
output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
require.NoError(t, err)
require.NoError(t, err, output)
done = true
state := ""
result := gjson.Get(output, "ShardStatuses")
Expand Down Expand Up @@ -522,7 +522,7 @@ func validateDryRunResults(t *testing.T, output string, want []string) {
w = strings.TrimSpace(w[1:])
result := strings.HasPrefix(g, w)
match = result
//t.Logf("Partial match |%v|%v|%v\n", w, g, match)
// t.Logf("Partial match |%v|%v|%v\n", w, g, match)
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved
} else {
match = g == w
}
Expand Down
12 changes: 9 additions & 3 deletions go/test/endtoend/vreplication/vdiff_helper_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ func doVtctlclientVDiff(t *testing.T, keyspace, workflow, cells string, want *ex
} else {
require.Equal(t, "completed", info.State, "vdiff results: %+v", info)
require.False(t, info.HasMismatch, "vdiff results: %+v", info)
require.NotZero(t, info.RowsCompared)
}
if strings.Contains(t.Name(), "AcrossDBVersions") {
log.Errorf("VDiff resume cannot be guaranteed between major MySQL versions due to implied collation differences, skipping resume test...")
Expand Down Expand Up @@ -150,9 +151,10 @@ func waitForVDiff2ToComplete(t *testing.T, useVtctlclient bool, ksWorkflow, cell
}

type expectedVDiff2Result struct {
state string
shards []string
hasMismatch bool
state string
shards []string
hasMismatch bool
minimumRowsCompared int64
}

func doVtctldclientVDiff(t *testing.T, keyspace, workflow, cells string, want *expectedVDiff2Result, extraFlags ...string) {
Expand All @@ -175,6 +177,10 @@ func doVtctldclientVDiff(t *testing.T, keyspace, workflow, cells string, want *e
} else {
require.Equal(t, "completed", info.State, "vdiff results: %+v", info)
require.False(t, info.HasMismatch, "vdiff results: %+v", info)
if want.minimumRowsCompared > 0 {
require.Greater(t, info.RowsCompared, want.minimumRowsCompared,
"not enough rows compared: want at least %d, got %d", want.minimumRowsCompared, info.RowsCompared)
}
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved
}
if strings.Contains(t.Name(), "AcrossDBVersions") {
log.Errorf("VDiff resume cannot be guaranteed between major MySQL versions due to implied collation differences, skipping resume test...")
Expand Down
146 changes: 146 additions & 0 deletions go/test/endtoend/vreplication/vdiff_online_ddl_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
package vreplication

import (
"context"
"fmt"
"strconv"
"strings"
"testing"
"time"

"github.com/stretchr/testify/require"
"google.golang.org/protobuf/encoding/protojson"

"vitess.io/vitess/go/mysql"
"vitess.io/vitess/go/vt/log"
binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
"vitess.io/vitess/go/vt/proto/vtctldata"
)

// TestOnlineDDLVDiff is to run a vdiff on a table that is part of an OnlineDDL workflow.
func TestOnlineDDLVDiff(t *testing.T) {
setSidecarDBName("_vt")
defaultRdonly = 0
defaultReplicas = 0
defer func() {
defaultRdonly = 1
defaultReplicas = 1
}()
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved

vc = setupMinimalCluster(t)
defer vc.TearDown()
keyspace := "product"
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

createQuery := "create table temp (id int, name varchar(100), blb blob, primary key (id))"
dropQuery := "drop table temp"
alterQuery := "alter table temp add column extra1 int not null default 0"
insertTemplate := "insert into temp (id, name, blb) values (%d, 'name%d', 'blb%d')"
updateTemplate := "update temp set name = 'name_%d' where id = %d"
execOnlineDDL(t, "direct", keyspace, createQuery)
defer execOnlineDDL(t, "direct", keyspace, dropQuery)

var done = make(chan bool)
go populate(ctx, done, insertTemplate, updateTemplate)

var output string
waitForAdditionalRows(t, keyspace, "temp", 100)
output = execOnlineDDL(t, "vitess --postpone-completion", keyspace, alterQuery)
uuid := strings.TrimSpace(output)
waitForAdditionalRows(t, keyspace, "temp", 200)
want := &expectedVDiff2Result{
state: "completed",
minimumRowsCompared: 200,
hasMismatch: false,
shards: []string{"0"},
}
doVtctldclientVDiff(t, keyspace, uuid, "zone1", want)

cancel()
<-done
}

func execOnlineDDL(t *testing.T, strategy, keyspace, query string) string {
output, err := vc.VtctldClient.ExecuteCommandWithOutput("ApplySchema", "--ddl-strategy", strategy, "--sql", query, keyspace)
require.NoError(t, err, output)
uuid := strings.TrimSpace(output)
if strategy != "direct" {
err = waitForCondition("online ddl to start", func() bool {
var response vtctldata.GetSchemaMigrationsResponse
output, err := vc.VtctldClient.OnlineDDLShow(keyspace, uuid)
require.NoError(t, err, output)
err = protojson.Unmarshal([]byte(output), &response)
if err != nil {
log.Errorf("error unmarshalling response: %v", err)
return false
}
if len(response.Migrations) > 0 && response.Migrations[0].Status == vtctldata.SchemaMigration_RUNNING {
return true
}
return false
}, 30*time.Second)
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved
require.NoError(t, err)
uuid := strings.TrimSpace(output)
waitForWorkflowState(t, vc, fmt.Sprintf("%s.%s", keyspace, uuid), binlogdatapb.VReplicationWorkflowState_Running.String())
}
return output
}

func waitForAdditionalRows(t *testing.T, keyspace, table string, count int) {
vtgateConn, cancel := getVTGateConn()
defer cancel()

numRowsStart := getNumRows(t, vtgateConn, keyspace, table)
shortCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved
defer cancel()
for {
switch {
case shortCtx.Err() != nil:
t.Fatalf("Timed out waiting for additional rows")
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved
default:
numRows := getNumRows(t, vtgateConn, keyspace, table)
if numRows >= numRowsStart+count {
return
}
time.Sleep(10 * time.Millisecond)
}
}
}

func getNumRows(t *testing.T, vtgateConn *mysql.Conn, keyspace, table string) int {
qr := execVtgateQuery(t, vtgateConn, keyspace, fmt.Sprintf("SELECT COUNT(*) FROM %s", table))
require.NotNil(t, qr)
numRows, err := strconv.Atoi(qr.Rows[0][0].ToString())
require.NoError(t, err)
return numRows
}

func populate(ctx context.Context, done chan bool, insertTemplate, updateTemplate string) {
defer close(done)
vtgateConn, closeConn := getVTGateConn()
defer closeConn()
id := 1
for {
select {
case <-ctx.Done():
log.Infof("load cancelled")
return
default:
query := fmt.Sprintf(insertTemplate, id, id, id)
_, err := vtgateConn.ExecuteFetch(query, 1, false)
if err != nil {
log.Errorf("error in insert: %v", err)
panic(err)
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved
}
query = fmt.Sprintf(updateTemplate, id, id)
_, err = vtgateConn.ExecuteFetch(query, 1, false)
if err != nil {
log.Errorf("error in update: %v", err)
panic(err)
}
id++
time.Sleep(10 * time.Millisecond)
}
}
}
4 changes: 0 additions & 4 deletions go/vt/vttablet/tabletmanager/vdiff/workflow_differ.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@ import (
"vitess.io/vitess/go/vt/binlog/binlogplayer"
"vitess.io/vitess/go/vt/key"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/schema"
"vitess.io/vitess/go/vt/sqlparser"
"vitess.io/vitess/go/vt/vtctl/schematools"
"vitess.io/vitess/go/vt/vterrors"
Expand Down Expand Up @@ -343,9 +342,6 @@ func (wd *workflowDiffer) buildPlan(dbClient binlogplayer.DBClient, filter *binl
if len(specifiedTables) != 0 && !stringListContains(specifiedTables, table.Name) {
continue
}
if schema.IsInternalOperationTableName(table.Name) {
rohit-nayak-ps marked this conversation as resolved.
Show resolved Hide resolved
continue
}
rule, err := vreplication.MatchTable(table.Name, filter)
if err != nil {
return err
Expand Down
9 changes: 9 additions & 0 deletions test/config.json
Original file line number Diff line number Diff line change
Expand Up @@ -1076,6 +1076,15 @@
"RetryMax": 0,
"Tags": []
},
"vreplication_onlineddl_vdiff": {
"File": "unused.go",
"Args": ["vitess.io/vitess/go/test/endtoend/vreplication", "-run", "TestOnlineDDLVDiff"],
"Command": [],
"Manual": false,
"Shard": "vreplication_cellalias",
"RetryMax": 2,
"Tags": []
},
"vreplication_vschema_load": {
"File": "unused.go",
"Args": ["vitess.io/vitess/go/test/endtoend/vreplication", "-run", "TestVSchemaChangesUnderLoad"],
Expand Down