Skip to content

Commit

Permalink
v15 backport: Onlineddl: formalize "immediate operations", respect --…
Browse files Browse the repository at this point in the history 
…postpone-completion strategy flag (vitessio#13832)

Signed-off-by: Shlomi Noach <[email protected]>
Co-authored-by: Deepthi Sigireddi <[email protected]>
Co-authored-by: Andrew Mason <[email protected]>
  • Loading branch information
@shlomi-noach @deepthi @ajm188
3 people authored Aug 28, 2023
1 parent d6b4de8 commit a1e0045
Show file tree
Hide file tree
Showing 4 changed files with 202 additions and 68 deletions.
44 changes: 40 additions & 4 deletions go/test/endtoend/onlineddl/revert/onlineddl_revert_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -347,6 +347,7 @@ func TestSchemaChange(t *testing.T) {
// ALTER VIEW
t.Run("ALTER VIEW where view exists", func(t *testing.T) {
// The view exists
checkTable(t, viewName, true)
uuid := testOnlineDDLStatementForView(t, alterViewStatement, ddlStrategy, "vtgate", "success_alter")
uuids = append(uuids, uuid)
onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
Expand Down Expand Up @@ -627,7 +628,8 @@ func TestSchemaChange(t *testing.T) {
checkMigratedTable(t, tableName, alterHints[0])
testSelectTableMetrics(t)
})
t.Run("postponed revert", func(t *testing.T) {
testPostponedRevert := func(t *testing.T, expectStatuses ...schema.OnlineDDLStatus) {
require.NotEmpty(t, expectStatuses)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
var wg sync.WaitGroup
Expand All @@ -636,22 +638,56 @@ func TestSchemaChange(t *testing.T) {
defer wg.Done()
runMultipleConnections(ctx, t)
}()
uuid := testRevertMigration(t, uuids[len(uuids)-1], ddlStrategy+" -postpone-completion")
uuid := testRevertMigration(t, uuids[len(uuids)-1], ddlStrategy+" --postpone-completion")
uuids = append(uuids, uuid)
// Should be still running!
onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusRunning)
onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, expectStatuses...)
// Issue a complete and wait for successful completion
onlineddl.CheckCompleteMigration(t, &vtParams, shards, uuid, true)
// This part may take a while, because we depend on vreplicatoin polling
// This part may take a while, because we depend on vreplication polling
status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, uuid, 60*time.Second, schema.OnlineDDLStatusComplete, schema.OnlineDDLStatusFailed)
fmt.Printf("# Migration status (for debug purposes): <%s>\n", status)
onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
cancel() // will cause runMultipleConnections() to terminate
wg.Wait()
}
t.Run("postponed revert", func(t *testing.T) {
testPostponedRevert(t, schema.OnlineDDLStatusRunning)
checkMigratedTable(t, tableName, alterHints[1])
testSelectTableMetrics(t)
})

t.Run("postponed revert view", func(t *testing.T) {
t.Run("CREATE VIEW again", func(t *testing.T) {
// The view does not exist
uuid := testOnlineDDLStatementForView(t, createViewStatement, ddlStrategy, "vtgate", "success_create")
uuids = append(uuids, uuid)
onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
checkTable(t, viewName, true)
testRevertedUUID(t, uuid, "")
})
t.Run("ALTER VIEW, postpone completion", func(t *testing.T) {
// Technically this test better fits in `onlineddl_scheduler_test.go`, but since we've already laid the grounds here, this is where it landed.
// The view exists
checkTable(t, viewName, true)
uuid := testOnlineDDLStatementForView(t, alterViewStatement, ddlStrategy+" --postpone-completion", "vtgate", "success_create")
uuids = append(uuids, uuid)

onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusQueued, schema.OnlineDDLStatusReady)
// Issue a complete and wait for successful completion
onlineddl.CheckCompleteMigration(t, &vtParams, shards, uuid, true)
// This part may take a while, because we depend on vreplication polling
status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, uuid, 60*time.Second, schema.OnlineDDLStatusComplete, schema.OnlineDDLStatusFailed)
fmt.Printf("# Migration status (for debug purposes): <%s>\n", status)
onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
checkTable(t, viewName, true)
testRevertedUUID(t, uuid, "")
})
// now verify that the revert for ALTER VIEW respects `--postpone-completion`
testPostponedRevert(t, schema.OnlineDDLStatusQueued, schema.OnlineDDLStatusReady)
checkTable(t, viewName, true)
})

// INSTANT DDL
t.Run("INSTANT DDL: add column", func(t *testing.T) {
uuid := testOnlineDDLStatementForTable(t, "alter table stress_test add column i_instant int not null default 0", ddlStrategy+" --fast-over-revertible", "vtgate", "i_instant")
Expand Down
21 changes: 14 additions & 7 deletions go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,9 @@ var (
shards []cluster.Shard
vtParams mysql.ConnParams

normalWaitTime = 20 * time.Second
extendedWaitTime = 60 * time.Second
normalWaitTime = 20 * time.Second
extendedWaitTime = 60 * time.Second
ensureStateNotChangedTime = 5 * time.Second

hostname = "localhost"
keyspaceName = "ks"
Expand Down Expand Up @@ -79,6 +80,9 @@ var (
trivialAlterT2Statement = `
ALTER TABLE t2_test ENGINE=InnoDB;
`
instantAlterT1Statement = `
ALTER TABLE t1_test ADD COLUMN i0 INT NOT NULL DEFAULT 0;
`
dropT1Statement = `
DROP TABLE IF EXISTS t1_test
`
Expand Down Expand Up @@ -161,6 +165,9 @@ func TestSchemaChange(t *testing.T) {
shards = clusterInstance.Keyspaces[0].Shards
require.Equal(t, 1, len(shards))

mysqlVersion := onlineddl.GetMySQLVersion(t, clusterInstance.Keyspaces[0].Shards[0].PrimaryTablet())
require.NotEmpty(t, mysqlVersion)

var t1uuid string
var t2uuid string

Expand Down Expand Up @@ -313,7 +320,7 @@ func TestSchemaChange(t *testing.T) {
onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t1uuid, normalWaitTime, schema.OnlineDDLStatusRunning)
// now that t1 is running, let's unblock t2. We expect it to remain queued.
onlineddl.CheckCompleteMigration(t, &vtParams, shards, t2uuid, true)
time.Sleep(5 * time.Second)
time.Sleep(ensureStateNotChangedTime)
// t1 should be still running!
onlineddl.CheckMigrationStatus(t, &vtParams, shards, t1uuid, schema.OnlineDDLStatusRunning)
// non-concurrent -- should be queued!
Expand Down Expand Up @@ -345,7 +352,7 @@ func TestSchemaChange(t *testing.T) {
t.Run("expect both running", func(t *testing.T) {
onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t1uuid, normalWaitTime, schema.OnlineDDLStatusRunning)
onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t2uuid, normalWaitTime, schema.OnlineDDLStatusRunning)
time.Sleep(5 * time.Second)
time.Sleep(ensureStateNotChangedTime)
// both should be still running!
onlineddl.CheckMigrationStatus(t, &vtParams, shards, t1uuid, schema.OnlineDDLStatusRunning)
onlineddl.CheckMigrationStatus(t, &vtParams, shards, t2uuid, schema.OnlineDDLStatusRunning)
Expand Down Expand Up @@ -384,7 +391,7 @@ func TestSchemaChange(t *testing.T) {
// since all migrations are throttled, t1 migration is not ready_to_complete, hence
// t2 should not be running
onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t2uuid, normalWaitTime, schema.OnlineDDLStatusQueued, schema.OnlineDDLStatusReady)
time.Sleep(5 * time.Second)
time.Sleep(ensureStateNotChangedTime)
// both should be still running!
onlineddl.CheckMigrationStatus(t, &vtParams, shards, t1uuid, schema.OnlineDDLStatusRunning)
onlineddl.CheckMigrationStatus(t, &vtParams, shards, t2uuid, schema.OnlineDDLStatusQueued, schema.OnlineDDLStatusReady)
Expand All @@ -393,7 +400,7 @@ func TestSchemaChange(t *testing.T) {
onlineddl.UnthrottleAllMigrations(t, &vtParams)
// t1 should now be ready_to_complete, hence t2 should start running
onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t2uuid, extendedWaitTime, schema.OnlineDDLStatusRunning)
time.Sleep(5 * time.Second)
time.Sleep(ensureStateNotChangedTime)
// both should be still running!
onlineddl.CheckMigrationStatus(t, &vtParams, shards, t1uuid, schema.OnlineDDLStatusRunning)
onlineddl.CheckMigrationStatus(t, &vtParams, shards, t2uuid, schema.OnlineDDLStatusRunning)
Expand Down Expand Up @@ -566,7 +573,7 @@ func TestSchemaChange(t *testing.T) {
})
drop1uuid := testOnlineDDLStatement(t, dropT1Statement, ddlStrategy+" -allow-concurrent", "vtgate", "", "", true) // skip wait
t.Run("t1drop blocked", func(t *testing.T) {
time.Sleep(5 * time.Second)
time.Sleep(ensureStateNotChangedTime)
// drop1 migration should block. It can run concurrently to t1, but conflicts on table name
onlineddl.CheckMigrationStatus(t, &vtParams, shards, drop1uuid, schema.OnlineDDLStatusReady)
})
Expand Down
Loading

0 comments on commit a1e0045

Please sign in to comment.