OnlineDDL: add in-order-completion count to show outputs

go/vt/sidecardb/schema/onlineddl/schema_migrations.sql

@@ -73,11 +73,12 @@ CREATE TABLE IF NOT EXISTS schema_migrations
     `is_immediate_operation`          tinyint unsigned NOT NULL DEFAULT '0',
     `reviewed_timestamp`              timestamp        NULL DEFAULT NULL,
     `ready_to_complete_timestamp`     timestamp        NULL DEFAULT NULL,
-    `shadow_analyzed_timestamp`        timestamp        NULL DEFAULT NULL,
+    `shadow_analyzed_timestamp`       timestamp        NULL DEFAULT NULL,
     `removed_foreign_key_names`       text             NOT NULL,
     `last_cutover_attempt_timestamp`  timestamp        NULL DEFAULT NULL,
     `force_cutover`                   tinyint unsigned NOT NULL DEFAULT '0',
     `cutover_threshold_seconds`       int unsigned     NOT NULL DEFAULT '0',
+    `dependent_migrations`            text             NOT NULL,
     PRIMARY KEY (`id`),
     UNIQUE KEY `uuid_idx` (`migration_uuid`),
     KEY `keyspace_shard_idx` (`keyspace`(64), `shard`(64)),

go/vt/vtctl/grpcvtctldserver/query.go

@@ -150,6 +150,7 @@ func rowToSchemaMigration(row sqltypes.RowNamedValues) (sm *vtctldatapb.SchemaMi
 	sm.VitessLivenessIndicator = row.AsInt64("vitess_liveness_indicator", 0)
 	sm.UserThrottleRatio = float32(row.AsFloat64("user_throttle_ratio", 0))
 	sm.SpecialPlan = row.AsString("special_plan", "")
+	sm.DependentMigrations = row.AsString("dependent_migrations", "")
 
 	sm.LastThrottledAt, err = valueToVTTime(row.AsString("last_throttled_timestamp", ""))
 	if err != nil {

go/vt/vtctl/grpcvtctldserver/server.go

@@ -1807,7 +1807,7 @@ func (s *VtctldServer) GetSchema(ctx context.Context, req *vtctldatapb.GetSchema
 }
 
 func (s *VtctldServer) GetSchemaMigrations(ctx context.Context, req *vtctldatapb.GetSchemaMigrationsRequest) (resp *vtctldatapb.GetSchemaMigrationsResponse, err error) {
-	span, ctx := trace.NewSpan(ctx, "VtctldServer.GetShard")
+	span, ctx := trace.NewSpan(ctx, "VtctldServer.GetSchemaMigrations")
 	defer span.Finish()
 
 	defer panicHandler(&err)

go/vt/vttablet/onlineddl/executor.go

@@ -3098,6 +3098,20 @@ func shouldCutOverAccordingToBackoff(
 	return false, false
 }
 
+// getDependentMigrations returns a slice of migrations that must cut-over in-order, before the provided
+// migration.
+func getDependentMigrations(onlineDDL *schema.OnlineDDL, pendingMigrationsUUIDs []string) []string {
+	dependentMigrations := make([]string, 0, len(pendingMigrationsUUIDs))
+	for _, pendingMigrationsUUID := range pendingMigrationsUUIDs {
+		if pendingMigrationsUUID == onlineDDL.UUID {
+			// found all dependencies if we found ourself
+			break
+		}
+		dependentMigrations = append(dependentMigrations, pendingMigrationsUUID)
+	}
+	return dependentMigrations
+}
+
 // reviewRunningMigrations iterates migrations in 'running' state. Normally there's only one running, which was
 // spawned by this tablet; but vreplication migrations could also resume from failure.
 func (e *Executor) reviewRunningMigrations(ctx context.Context) (countRunnning int, cancellable []*cancellableMigration, err error) {
@@ -3245,19 +3259,23 @@ func (e *Executor) reviewRunningMigrations(ctx context.Context) (countRunnning i
 				// In the case of a postponed migration, we will not complete it, but the user will
 				// understand whether "now is a good time" or "not there yet"
 				_ = e.updateMigrationReadyToComplete(ctx, uuid, isReady)
+				var postponeInOrderCompletion bool
+				if strategySetting.IsInOrderCompletion() {
+					if len(pendingMigrationsUUIDs) > 0 && pendingMigrationsUUIDs[0] != onlineDDL.UUID {
+						if err = e.updateDependentMigrations(ctx, onlineDDL.UUID, getDependentMigrations(onlineDDL, pendingMigrationsUUIDs)); err != nil {
+							return err
+						}
+						postponeInOrderCompletion = true
+					}
+				}
 				if !isReady {
+					// The migration is not ready yet.
 					return nil
 				}
-				if postponeCompletion {
+				if postponeCompletion || postponeInOrderCompletion {
 					// override. Even if migration is ready, we do not complete it.
 					return nil
 				}
-				if strategySetting.IsInOrderCompletion() {
-					if len(pendingMigrationsUUIDs) > 0 && pendingMigrationsUUIDs[0] != onlineDDL.UUID {
-						// wait for earlier pending migrations to complete
-						return nil
-					}
-				}
 				shouldCutOver, shouldForceCutOver := shouldCutOverAccordingToBackoff(
 					shouldForceCutOver, forceCutOverAfter, sinceReadyToComplete, sinceLastCutoverAttempt, cutoverAttempts,
 				)
@@ -3773,8 +3791,17 @@ func (e *Executor) updateMigrationStatusFailedOrCancelled(ctx context.Context, u
 }
 
 func (e *Executor) updateMigrationStatus(ctx context.Context, uuid string, status schema.OnlineDDLStatus) error {
+	var sqlUpdate string
+	switch status {
+	case schema.OnlineDDLStatusComplete, schema.OnlineDDLStatusFailed, schema.OnlineDDLStatusCancelled:
+		// sqlUpdateMigrationStatusFinal represents a final update to migration status.
+		sqlUpdate = sqlUpdateMigrationStatusFinal
+	default:
+		sqlUpdate = sqlUpdateMigrationStatus
+	}
+
 	log.Infof("updateMigrationStatus: transitioning migration: %s into status: %s", uuid, string(status))
-	query, err := sqlparser.ParseAndBind(sqlUpdateMigrationStatus,
+	query, err := sqlparser.ParseAndBind(sqlUpdate,
 		sqltypes.StringBindVariable(string(status)),
 		sqltypes.StringBindVariable(uuid),
 	)
@@ -4063,6 +4090,25 @@ func (e *Executor) updateMigrationUserThrottleRatio(ctx context.Context, uuid st
 	return err
 }
 
+func (e *Executor) updateDependentMigrations(
+	ctx context.Context,
+	uuid string,
+	dependentMigrations []string,
+) error {
+	if len(dependentMigrations) == 0 {
+		return nil
+	}
+	query, err := sqlparser.ParseAndBind(sqlUpdateDependentMigrations,
+		sqltypes.StringBindVariable(strings.Join(dependentMigrations, ",")),
+		sqltypes.StringBindVariable(uuid),
+	)
+	if err != nil {
+		return err
+	}
+	_, err = e.execQuery(ctx, query)
+	return err
+}
+
 // retryMigrationWhere retries a migration based on a given WHERE clause
 func (e *Executor) retryMigrationWhere(ctx context.Context, whereExpr string) (result *sqltypes.Result, err error) {
 	e.migrationMutex.Lock()
@@ -4090,8 +4136,23 @@ func (e *Executor) RetryMigration(ctx context.Context, uuid string) (result *sql
 	e.migrationMutex.Lock()
 	defer e.migrationMutex.Unlock()
 
+	onlineDDL, _, err := e.readMigration(ctx, uuid)
+	if err != nil {
+		return nil, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, err.Error())
+	}
+
+	var dependentMigrations []string
+	if onlineDDL.StrategySetting().IsInOrderCompletion() {
+		pendingMigrationsUUIDs, err := e.readPendingMigrationsUUIDs(ctx)
+		if err != nil {
+			return nil, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, err.Error())
+		}
+		dependentMigrations = getDependentMigrations(onlineDDL, pendingMigrationsUUIDs)
+	}
+
 	query, err := sqlparser.ParseAndBind(sqlRetryMigration,
 		sqltypes.StringBindVariable(e.TabletAliasString()),
+		sqltypes.StringBindVariable(strings.Join(dependentMigrations, ",")),
 		sqltypes.StringBindVariable(uuid),
 	)
 	if err != nil {
@@ -4577,6 +4638,15 @@ func (e *Executor) SubmitMigration(
 	}
 	log.Infof("SubmitMigration: request to submit migration %s; action=%s, table=%s", onlineDDL.UUID, actionStr, onlineDDL.Table)
 
+	var dependentMigrations []string
+	if onlineDDL.StrategySetting().IsInOrderCompletion() {
+		pendingMigrationsUUIDs, err := e.readPendingMigrationsUUIDs(ctx)
+		if err != nil {
+			return nil, err
+		}
+		dependentMigrations = getDependentMigrations(onlineDDL, pendingMigrationsUUIDs)
+	}
+
 	revertedUUID, _ := onlineDDL.GetRevertUUID(e.env.Environment().Parser()) // Empty value if the migration is not actually a REVERT. Safe to ignore error.
 	retainArtifactsSeconds := int64((retainOnlineDDLTables).Seconds())
 	if retainArtifacts, _ := onlineDDL.StrategySetting().RetainArtifactsDuration(); retainArtifacts != 0 {
@@ -4612,6 +4682,7 @@ func (e *Executor) SubmitMigration(
 		sqltypes.BoolBindVariable(allowConcurrentMigration),
 		sqltypes.StringBindVariable(revertedUUID),
 		sqltypes.BoolBindVariable(onlineDDL.IsView(e.env.Environment().Parser())),
+		sqltypes.StringBindVariable(strings.Join(dependentMigrations, ",")),
 	)
 	if err != nil {
 		return nil, err

go/vt/vttablet/onlineddl/schema.go

@@ -37,9 +37,10 @@ const (
 		postpone_completion,
 		allow_concurrent,
 		reverted_uuid,
-		is_view
+		is_view,
+		dependent_migrations
 	) VALUES (
-		%a, %a, %a, %a, %a, %a, %a, %a, %a, NOW(6), %a, %a, %a, %a, %a, %a, %a, %a, %a, %a
+		%a, %a, %a, %a, %a, %a, %a, %a, %a, NOW(6), %a, %a, %a, %a, %a, %a, %a, %a, %a, %a, %a
 	)`
 
 	sqlSelectQueuedMigrations = `SELECT
@@ -66,9 +67,16 @@ const (
 		WHERE
 			migration_uuid=%a
 	`
+	sqlUpdateMigrationStatusFinal = `UPDATE _vt.schema_migrations
+			SET migration_status=%a,
+			dependent_migrations=''
+		WHERE
+			migration_uuid=%a
+	`
 	sqlUpdateMigrationStatusFailedOrCancelled = `UPDATE _vt.schema_migrations
 			SET migration_status=IF(cancelled_timestamp IS NULL, 'failed', 'cancelled'),
-			completed_timestamp=NOW(6)
+			completed_timestamp=NOW(6),
+			dependent_migrations=''
 		WHERE
 			migration_uuid=%a
 	`
@@ -139,6 +147,11 @@ const (
 		WHERE
 			migration_uuid=%a
 	`
+	sqlUpdateDependentMigrations = `UPDATE _vt.schema_migrations
+			SET dependent_migrations=%a
+		WHERE
+			migration_uuid=%a
+	`
 	sqlUpdateArtifacts = `UPDATE _vt.schema_migrations
 			SET artifacts=concat(%a, ',', artifacts), cleanup_timestamp=NULL
 		WHERE
@@ -310,7 +323,8 @@ const (
 			completed_timestamp=NULL,
 			last_cutover_attempt_timestamp=NULL,
 			shadow_analyzed_timestamp=NULL,
-			cleanup_timestamp=NULL
+			cleanup_timestamp=NULL,
+			dependent_migrations=%a
 		WHERE
 			migration_status IN ('failed', 'cancelled')
 			AND migration_uuid=%a

proto/vtctldata.proto

@@ -175,6 +175,7 @@ message SchemaMigration {
   vttime.Time reviewed_at = 52;
   vttime.Time ready_to_complete_at = 53;
   string removed_foreign_key_names = 54;
+  string dependent_migrations = 55;
 
   enum Strategy {
     option allow_alias = true;