Online DDL: --singleton-table DDL strategy flag (#17169)

Signed-off-by: Shlomi Noach <[email protected]>

go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go

@@ -1685,7 +1685,7 @@ DROP TABLE IF EXISTS stress_test
 	t.Run("terminate throttled migration", func(t *testing.T) {
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, openEndedUUID, schema.OnlineDDLStatusRunning)
 		onlineddl.CheckCancelMigration(t, &vtParams, shards, openEndedUUID, true)
-		status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, openEndedUUID, 20*time.Second, schema.OnlineDDLStatusFailed, schema.OnlineDDLStatusCancelled)
+		status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, openEndedUUID, normalWaitTime, schema.OnlineDDLStatusFailed, schema.OnlineDDLStatusCancelled)
 		fmt.Printf("# Migration status (for debug purposes): <%s>\n", status)
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, openEndedUUID, schema.OnlineDDLStatusCancelled)
 	})
@@ -1718,6 +1718,21 @@ DROP TABLE IF EXISTS stress_test
 		checkTable(t, tableName, true)
 	})
 
+	// singleton-table
+	t.Run("fail singleton-table same table single submission", func(t *testing.T) {
+		_ = testOnlineDDLStatement(t, createParams(multiAlterTableThrottlingStatement, "vitess --singleton-table", "vtctl", "", "hint_col", "singleton-table migration rejected", false))
+		// The first of those migrations will make it, the other two will be rejected
+		onlineddl.CheckCancelAllMigrations(t, &vtParams, 1)
+	})
+	t.Run("fail singleton-table same table multi submission", func(t *testing.T) {
+		uuid := testOnlineDDLStatement(t, createParams(alterTableThrottlingStatement, "vitess --singleton-table --postpone-completion", "vtctl", "", "hint_col", "", false))
+		uuids = append(uuids, uuid)
+		onlineddl.WaitForMigrationStatus(t, &vtParams, shards, uuid, normalWaitTime, schema.OnlineDDLStatusRunning)
+
+		_ = testOnlineDDLStatement(t, createParams(alterTableThrottlingStatement, "vitess --singleton-table --postpone-completion", "vtctl", "", "hint_col", "singleton-table migration rejected", false))
+		onlineddl.CheckCancelAllMigrations(t, &vtParams, 1)
+	})
+
 	var throttledUUIDs []string
 	// singleton-context
 	t.Run("postponed migrations, singleton-context", func(t *testing.T) {
@@ -1774,29 +1789,29 @@ DROP TABLE IF EXISTS stress_test
 		uuids = append(uuids, uuid)
 		_ = testOnlineDDLStatement(t, createParams(dropNonexistentTableStatement, "vitess --singleton", "vtgate", "", "hint_col", "rejected", true))
 		onlineddl.CheckCompleteAllMigrations(t, &vtParams, len(shards))
-		status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, uuid, 20*time.Second, schema.OnlineDDLStatusComplete, schema.OnlineDDLStatusFailed)
+		status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, uuid, normalWaitTime, schema.OnlineDDLStatusComplete, schema.OnlineDDLStatusFailed)
 		fmt.Printf("# Migration status (for debug purposes): <%s>\n", status)
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
 	})
 	t.Run("fail concurrent singleton-context with revert", func(t *testing.T) {
 		revertUUID := testRevertMigration(t, createRevertParams(uuids[len(uuids)-1], "vitess --allow-concurrent --postpone-completion --singleton-context", "vtctl", "rev:ctx", "", false))
-		onlineddl.WaitForMigrationStatus(t, &vtParams, shards, revertUUID, 20*time.Second, schema.OnlineDDLStatusRunning)
+		onlineddl.WaitForMigrationStatus(t, &vtParams, shards, revertUUID, normalWaitTime, schema.OnlineDDLStatusRunning)
 		// revert is running
 		_ = testOnlineDDLStatement(t, createParams(dropNonexistentTableStatement, "vitess --allow-concurrent --singleton-context", "vtctl", "migrate:ctx", "", "rejected", true))
 		onlineddl.CheckCancelMigration(t, &vtParams, shards, revertUUID, true)
-		status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, revertUUID, 20*time.Second, schema.OnlineDDLStatusFailed, schema.OnlineDDLStatusCancelled)
+		status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, revertUUID, normalWaitTime, schema.OnlineDDLStatusFailed, schema.OnlineDDLStatusCancelled)
 		fmt.Printf("# Migration status (for debug purposes): <%s>\n", status)
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, revertUUID, schema.OnlineDDLStatusCancelled)
 	})
 	t.Run("success concurrent singleton-context with no-context revert", func(t *testing.T) {
 		revertUUID := testRevertMigration(t, createRevertParams(uuids[len(uuids)-1], "vitess --allow-concurrent --postpone-completion", "vtctl", "rev:ctx", "", false))
-		onlineddl.WaitForMigrationStatus(t, &vtParams, shards, revertUUID, 20*time.Second, schema.OnlineDDLStatusRunning)
+		onlineddl.WaitForMigrationStatus(t, &vtParams, shards, revertUUID, normalWaitTime, schema.OnlineDDLStatusRunning)
 		// revert is running but has no --singleton-context. Our next migration should be able to run.
 		uuid := testOnlineDDLStatement(t, createParams(dropNonexistentTableStatement, "vitess --allow-concurrent --singleton-context", "vtctl", "migrate:ctx", "", "", false))
 		uuids = append(uuids, uuid)
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
 		onlineddl.CheckCancelMigration(t, &vtParams, shards, revertUUID, true)
-		status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, revertUUID, 20*time.Second, schema.OnlineDDLStatusFailed, schema.OnlineDDLStatusCancelled)
+		status := onlineddl.WaitForMigrationStatus(t, &vtParams, shards, revertUUID, normalWaitTime, schema.OnlineDDLStatusFailed, schema.OnlineDDLStatusCancelled)
 		fmt.Printf("# Migration status (for debug purposes): <%s>\n", status)
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, revertUUID, schema.OnlineDDLStatusCancelled)
 	})

go/vt/schema/ddl_strategy.go

@@ -38,6 +38,7 @@ const (
 	skipTopoFlag           = "skip-topo" // legacy. Kept for backwards compatibility, but unused
 	singletonFlag          = "singleton"
 	singletonContextFlag   = "singleton-context"
+	singletonTableFlag     = "singleton-table"
 	allowZeroInDateFlag    = "allow-zero-in-date"
 	postponeLaunchFlag     = "postpone-launch"
 	postponeCompletionFlag = "postpone-completion"
@@ -177,6 +178,11 @@ func (setting *DDLStrategySetting) IsSingletonContext() bool {
 	return setting.hasFlag(singletonContextFlag)
 }
 
+// IsSingletonTable checks if strategy options include --singleton-table
+func (setting *DDLStrategySetting) IsSingletonTable() bool {
+	return setting.hasFlag(singletonTableFlag)
+}
+
 // IsAllowZeroInDateFlag checks if strategy options include --allow-zero-in-date
 func (setting *DDLStrategySetting) IsAllowZeroInDateFlag() bool {
 	return setting.hasFlag(allowZeroInDateFlag)
@@ -322,6 +328,7 @@ func (setting *DDLStrategySetting) RuntimeOptions() []string {
 		case isFlag(opt, skipTopoFlag): // deprecated flag, parsed for backwards compatibility
 		case isFlag(opt, singletonFlag):
 		case isFlag(opt, singletonContextFlag):
+		case isFlag(opt, singletonTableFlag):
 		case isFlag(opt, allowZeroInDateFlag):
 		case isFlag(opt, postponeLaunchFlag):
 		case isFlag(opt, postponeCompletionFlag):

go/vt/schema/ddl_strategy_test.go

@@ -189,6 +189,8 @@ func TestParseDDLStrategy(t *testing.T) {
 		options              string
 		isDeclarative        bool
 		isSingleton          bool
+		isSingletonContext   bool
+		isSingletonTable     bool
 		isPostponeLaunch     bool
 		isPostponeCompletion bool
 		isInOrderCompletion  bool
@@ -258,6 +260,20 @@ func TestParseDDLStrategy(t *testing.T) {
 			runtimeOptions:   "",
 			isSingleton:      true,
 		},
+		{
+			strategyVariable:   "vitess --singleton-context",
+			strategy:           DDLStrategyVitess,
+			options:            "--singleton-context",
+			runtimeOptions:     "",
+			isSingletonContext: true,
+		},
+		{
+			strategyVariable: "vitess --singleton-table",
+			strategy:         DDLStrategyVitess,
+			options:          "--singleton-table",
+			runtimeOptions:   "",
+			isSingletonTable: true,
+		},
 		{
 			strategyVariable: "online -postpone-launch",
 			strategy:         DDLStrategyOnline,
@@ -387,6 +403,8 @@ func TestParseDDLStrategy(t *testing.T) {
 			assert.Equal(t, ts.options, setting.Options)
 			assert.Equal(t, ts.isDeclarative, setting.IsDeclarative())
 			assert.Equal(t, ts.isSingleton, setting.IsSingleton())
+			assert.Equal(t, ts.isSingletonContext, setting.IsSingletonContext())
+			assert.Equal(t, ts.isSingletonTable, setting.IsSingletonTable())
 			assert.Equal(t, ts.isPostponeCompletion, setting.IsPostponeCompletion())
 			assert.Equal(t, ts.isPostponeLaunch, setting.IsPostponeLaunch())
 			assert.Equal(t, ts.isAllowConcurrent, setting.IsAllowConcurrent())

go/vt/vttablet/onlineddl/executor.go

@@ -4901,7 +4901,7 @@ func (e *Executor) submitCallbackIfNonConflicting(
 ) (
 	result *sqltypes.Result, err error,
 ) {
-	if !onlineDDL.StrategySetting().IsSingleton() && !onlineDDL.StrategySetting().IsSingletonContext() {
+	if !onlineDDL.StrategySetting().IsSingleton() && !onlineDDL.StrategySetting().IsSingletonContext() && !onlineDDL.StrategySetting().IsSingletonTable() {
 		// not a singleton. No conflict
 		return callback()
 	}
@@ -4947,6 +4947,15 @@ func (e *Executor) submitCallbackIfNonConflicting(
 				}
 				// no conflict? continue looking for other pending migrations
 			}
+		case onlineDDL.StrategySetting().IsSingletonTable():
+			// We will reject this migration if there's any pending migration for the same table
+			for _, row := range rows {
+				pendingTableName := row["mysql_table"].ToString()
+				if onlineDDL.Table == pendingTableName {
+					pendingUUID := row["migration_uuid"].ToString()
+					return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "singleton-table migration rejected: found pending migration: %s for the same table: %s", pendingUUID, onlineDDL.Table)
+				}
+			}
 		}
 		return nil
 	}()