Repository: vitessio/vitess
Pull request #11194 (2022-09-09 11:59:34)

Replication Manager Improvements
<!--
  Thank you for your contribution to the Vitess project.
  How to contribute: https://vitess.io/docs/contributing/
  Please first make sure there is an open Issue to discuss the feature/fix suggested in this PR.
  If this is a new feature, please mark the Issue as "RFC".
 -->

<!-- if this PR is Work in Progress please create it as a Draft Pull Request -->

## Description

<!-- A few sentences describing the overall goals of the pull request's commits. -->
<!-- If this is a bug fix and you think the fix should be backported, please write so. -->

This PR improves the replication manager. The proposed improvements are two-fold - 
1. The first is that the replication manager should be able to catch the case where the replication is broken because the primary's hostname and port changed due to kubernetes pod reschedule or some other reason. 
2. The second improvement is the addition of a flag that allows the replication manager to be turned off, if so desired.

Tests of both these changes have been added. An end to end test for the first one and a unit test for the other.

## Related Issue(s)
#9809 
<!-- List related issues and pull requests. If this PR fixes an issue, please add it using Fixes #????  -->

## Checklist

-   [x] "Backport me!" label has been added if this change should be backported
-   [x] Tests were added or are not required
-   [x] Documentation was added or is not required

## Deployment Notes

<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->



Patch:
diff --git a/.github/workflows/code_freeze.yml b/.github/workflows/code_freeze.yml
new file mode 100644
index 0000000000..381793fa84
--- /dev/null
+++ b/.github/workflows/code_freeze.yml
@@ -0,0 +1,12 @@
+name: Code Freeze
+on:
+  pull_request:
+
+jobs:
+  build:
+    name: Code Freeze
+    runs-on: ubuntu-latest
+    steps:
+    - name: Fail if Code Freeze is enabled
+      run: |
+        exit 0
diff --git a/go/cmd/query_analyzer/query_analyzer.go b/go/cmd/query_analyzer/query_analyzer.go
index d081c08f90..f8ce05a275 100644
--- a/go/cmd/query_analyzer/query_analyzer.go
+++ b/go/cmd/query_analyzer/query_analyzer.go
@@ -29,6 +29,7 @@ import (
 	"vitess.io/vitess/go/exit"
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/logutil"
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/sqlparser"
 
 	// Include deprecation warnings for soon-to-be-unsupported flag invocations.
@@ -65,6 +66,7 @@ func main() {
 	fs := pflag.NewFlagSet("query_analyzer", pflag.ExitOnError)
 	log.RegisterFlags(fs)
 	logutil.RegisterFlags(fs)
+	servenv.RegisterMySQLServerFlags(fs)
 	_flag.Parse(fs)
 	for _, filename := range _flag.Args() {
 		fmt.Printf("processing: %s\n", filename)
diff --git a/go/cmd/rulesctl/main.go b/go/cmd/rulesctl/main.go
index 577ef29ade..a5276615a6 100644
--- a/go/cmd/rulesctl/main.go
+++ b/go/cmd/rulesctl/main.go
@@ -6,12 +6,14 @@ import (
 	"vitess.io/vitess/go/cmd/rulesctl/cmd"
 	vtlog "vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/logutil"
+	"vitess.io/vitess/go/vt/servenv"
 )
 
 func main() {
 	rootCmd := cmd.Main()
 	vtlog.RegisterFlags(rootCmd.PersistentFlags())
 	logutil.RegisterFlags(rootCmd.PersistentFlags())
+	servenv.RegisterMySQLServerFlags(rootCmd.PersistentFlags())
 	if err := rootCmd.Execute(); err != nil {
 		log.Printf("%v", err)
 	}
diff --git a/go/cmd/vtbench/vtbench.go b/go/cmd/vtbench/vtbench.go
index a931f1410f..deab7f4e6b 100644
--- a/go/cmd/vtbench/vtbench.go
+++ b/go/cmd/vtbench/vtbench.go
@@ -30,6 +30,7 @@ import (
 	"vitess.io/vitess/go/vt/grpccommon"
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/logutil"
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vtbench"
 
 	// Import and register the gRPC vtgateconn client
@@ -111,6 +112,7 @@ func main() {
 	grpccommon.RegisterFlags(fs)
 	log.RegisterFlags(fs)
 	logutil.RegisterFlags(fs)
+	servenv.RegisterMySQLServerFlags(fs)
 	_flag.Parse(fs)
 
 	clientProto := vtbench.MySQL
diff --git a/go/cmd/vtclient/vtclient.go b/go/cmd/vtclient/vtclient.go
index 97c8eeccd7..b570f5b920 100644
--- a/go/cmd/vtclient/vtclient.go
+++ b/go/cmd/vtclient/vtclient.go
@@ -37,6 +37,7 @@ import (
 	"vitess.io/vitess/go/vt/grpccommon"
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/logutil"
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vitessdriver"
 	"vitess.io/vitess/go/vt/vterrors"
@@ -153,6 +154,7 @@ func run() (*results, error) {
 	grpccommon.RegisterFlags(fs)
 	log.RegisterFlags(fs)
 	logutil.RegisterFlags(fs)
+	servenv.RegisterMySQLServerFlags(fs)
 	_flag.Parse(fs)
 	args := _flag.Args()
 
diff --git a/go/cmd/vtctldclient/main.go b/go/cmd/vtctldclient/main.go
index 57b7279182..84920db379 100644
--- a/go/cmd/vtctldclient/main.go
+++ b/go/cmd/vtctldclient/main.go
@@ -24,6 +24,7 @@ import (
 	"vitess.io/vitess/go/exit"
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/logutil"
+	"vitess.io/vitess/go/vt/servenv"
 )
 
 func main() {
@@ -33,6 +34,7 @@ func main() {
 	command.Root.PersistentFlags().AddGoFlagSet(flag.CommandLine)
 	log.RegisterFlags(command.Root.PersistentFlags())
 	logutil.RegisterFlags(command.Root.PersistentFlags())
+	servenv.RegisterMySQLServerFlags(command.Root.PersistentFlags())
 
 	// hack to get rid of an "ERROR: logging before flag.Parse"
 	args := os.Args[:]
diff --git a/go/flags/endtoend/vtgr.txt b/go/flags/endtoend/vtgr.txt
index 2409440005..8ce42ff2b4 100644
--- a/go/flags/endtoend/vtgr.txt
+++ b/go/flags/endtoend/vtgr.txt
@@ -28,7 +28,6 @@ Usage of vtgr:
       --log_rotate_max_size uint                   size in bytes at which logs are rotated (glog.MaxSize) (default 1887436800)
       --logtostderr                                log to standard error instead of files
       --mysql_server_flush_delay duration          Delay after which buffered response will be flushed to the client. (default 100ms)
-      --mysql_server_version string                MySQL server version to advertise.
       --ping_tablet_timeout duration               time to wait when we ping a tablet (default 2s)
       --pprof strings                              enable profiling
       --purge_logs_interval duration               how often try to remove old logs (default 1h0m0s)
diff --git a/go/flags/endtoend/vttablet.txt b/go/flags/endtoend/vttablet.txt
index 0d0fd23a90..0c68b5b3ff 100644
--- a/go/flags/endtoend/vttablet.txt
+++ b/go/flags/endtoend/vttablet.txt
@@ -180,6 +180,7 @@ Usage of vttablet:
       --dba_idle_timeout duration                                        Idle timeout for dba connections (default 1m0s)
       --dba_pool_size int                                                Size of the connection pool for dba connections (default 20)
       --degraded_threshold duration                                      replication lag after which a replica is considered degraded (default 30s)
+      --disable-replication-manager                                      Disable replication manager to prevent replication repairs.
       --disable_active_reparents                                         if set, do not allow active reparents. Use this to protect a cluster using external reparents.
       --emit_stats                                                       If set, emit stats to push-based monitoring and stats backends
       --enable-consolidator                                              Synonym to -enable_consolidator (default true)
diff --git a/go/mysql/collations/local.go b/go/mysql/collations/local.go
index 17fd6f20bc..988de15697 100644
--- a/go/mysql/collations/local.go
+++ b/go/mysql/collations/local.go
@@ -35,13 +35,13 @@ func Local() *Environment {
 		if !flag.Parsed() {
 			panic("collations.Local() called too early")
 		}
-		if *servenv.MySQLServerVersion == "" {
+		if mySQLVersion := servenv.MySQLServerVersion(); mySQLVersion == "" {
 			// The default server version used by vtgate is 5.7.9
 			// NOTE: this should be changed along with the effective default
 			//       for the vtgate mysql_server_version flag.
 			defaultEnv = fetchCacheEnvironment(collverMySQL57)
 		} else {
-			defaultEnv = NewEnvironment(*servenv.MySQLServerVersion)
+			defaultEnv = NewEnvironment(mySQLVersion)
 		}
 	})
 	return defaultEnv
diff --git a/go/vt/discovery/tablet_picker.go b/go/vt/discovery/tablet_picker.go
index ec46e4a6d9..225b3c50b7 100644
--- a/go/vt/discovery/tablet_picker.go
+++ b/go/vt/discovery/tablet_picker.go
@@ -102,9 +102,9 @@ func NewTabletPicker(ts *topo.Server, cells []string, keyspace, shard, tabletTyp
 	}, nil
 }
 
-// PickForStreaming picks an available tablet
+// PickForStreaming picks an available tablet.
 // All tablets that belong to tp.cells are evaluated and one is
-// chosen at random
+// chosen at random.
 func (tp *TabletPicker) PickForStreaming(ctx context.Context) (*topodatapb.Tablet, error) {
 	rand.Seed(time.Now().UnixNano())
 	// keep trying at intervals (tabletPickerRetryDelay) until a tablet is found
diff --git a/go/vt/orchestrator/app/cli.go b/go/vt/orchestrator/app/cli.go
index 3f7ed5872a..07acbb326e 100644
--- a/go/vt/orchestrator/app/cli.go
+++ b/go/vt/orchestrator/app/cli.go
@@ -504,15 +504,7 @@ func Cli(command string, strict bool, instance string, destination string, owner
 		}
 	case registerCliCommand("reattach-replica-primary-host", "Replication, general", `Undo a detach-replica-primary-host operation`):
 		{
-			instanceKey, _ = inst.FigureInstanceKey(instanceKey, thisInstanceKey)
-			if instanceKey == nil {
-				log.Fatal("Cannot deduce instance:", instance)
-			}
-			_, err := inst.ReattachReplicaPrimaryHost(instanceKey)
-			if err != nil {
-				log.Fatal(err)
-			}
-			fmt.Println(instanceKey.DisplayString())
+			fmt.Println("Command deprecated")
 		}
 	case registerCliCommand("primary-pos-wait", "Replication, general", `Wait until replica reaches given replication coordinates (--binlog=file:pos)`):
 		{
@@ -671,7 +663,7 @@ func Cli(command string, strict bool, instance string, destination string, owner
 				log.Fatal(err)
 			}
 			for _, clusterPoolInstance := range clusterPoolInstances {
-				fmt.Printf("%s\t%s\t%s\t%s:%d\n", clusterPoolInstance.ClusterName, clusterPoolInstance.ClusterAlias, clusterPoolInstance.Pool, clusterPoolInstance.Hostname, clusterPoolInstance.Port)
+				fmt.Printf("%s\t%s\t%s:%d\n", clusterPoolInstance.ClusterName, clusterPoolInstance.Pool, clusterPoolInstance.Hostname, clusterPoolInstance.Port)
 			}
 		}
 	case registerCliCommand("which-heuristic-cluster-pool-instances", "Pools", `List instances of a given cluster which are in either any pool or in a specific pool`):
@@ -731,7 +723,7 @@ func Cli(command string, strict bool, instance string, destination string, owner
 				log.Fatal(err)
 			}
 			for _, cluster := range clusters {
-				fmt.Printf("%s\t%s\n", cluster.ClusterName, cluster.ClusterAlias)
+				fmt.Printf("%s\n", cluster.ClusterName)
 			}
 		}
 	case registerCliCommand("all-clusters-primaries", "Information", `List of writeable primaries, one per cluster`):
@@ -804,7 +796,7 @@ func Cli(command string, strict bool, instance string, destination string, owner
 			if err != nil {
 				log.Fatal(err)
 			}
-			fmt.Println(clusterInfo.ClusterAlias)
+			fmt.Println(clusterInfo.ClusterName)
 		}
 	case registerCliCommand("which-cluster-domain", "Information", `Output the domain name of the cluster an instance belongs to, or error if unknown to orchestrator`):
 		{
diff --git a/go/vt/orchestrator/config/config.go b/go/vt/orchestrator/config/config.go
index 4833f60906..58a6280157 100644
--- a/go/vt/orchestrator/config/config.go
+++ b/go/vt/orchestrator/config/config.go
@@ -199,10 +199,10 @@ type Configuration struct {
 	RecoverPrimaryClusterFilters                []string          // Only do primary recovery on clusters matching these regexp patterns (of course the ".*" pattern matches everything)
 	RecoverIntermediatePrimaryClusterFilters    []string          // Only do IM recovery on clusters matching these regexp patterns (of course the ".*" pattern matches everything)
 	ProcessesShellCommand                       string            // Shell that executes command scripts
-	OnFailureDetectionProcesses                 []string          // Processes to execute when detecting a failover scenario (before making a decision whether to failover or not). May and should use some of these placeholders: {failureType}, {instanceType}, {isPrimary}, {isCoPrimary}, {failureDescription}, {command}, {failedHost}, {failureCluster}, {failureClusterAlias}, {failureClusterDomain}, {failedPort}, {successorHost}, {successorPort}, {successorAlias}, {countReplicas}, {replicaHosts}, {isDowntimed}, {autoPrimaryRecovery}, {autoIntermediatePrimaryRecovery}
-	PreFailoverProcesses                        []string          // Processes to execute before doing a failover (aborting operation should any once of them exits with non-zero code; order of execution undefined). May and should use some of these placeholders: {failureType}, {instanceType}, {isPrimary}, {isCoPrimary}, {failureDescription}, {command}, {failedHost}, {failureCluster}, {failureClusterAlias}, {failureClusterDomain}, {failedPort}, {countReplicas}, {replicaHosts}, {isDowntimed}
-	PostFailoverProcesses                       []string          // Processes to execute after doing a failover (order of execution undefined). May and should use some of these placeholders: {failureType}, {instanceType}, {isPrimary}, {isCoPrimary}, {failureDescription}, {command}, {failedHost}, {failureCluster}, {failureClusterAlias}, {failureClusterDomain}, {failedPort}, {successorHost}, {successorPort}, {successorAlias}, {countReplicas}, {replicaHosts}, {isDowntimed}, {isSuccessful}, {lostReplicas}, {countLostReplicas}
-	PostUnsuccessfulFailoverProcesses           []string          // Processes to execute after a not-completely-successful failover (order of execution undefined). May and should use some of these placeholders: {failureType}, {instanceType}, {isPrimary}, {isCoPrimary}, {failureDescription}, {command}, {failedHost}, {failureCluster}, {failureClusterAlias}, {failureClusterDomain}, {failedPort}, {successorHost}, {successorPort}, {successorAlias}, {countReplicas}, {replicaHosts}, {isDowntimed}, {isSuccessful}, {lostReplicas}, {countLostReplicas}
+	OnFailureDetectionProcesses                 []string          // Processes to execute when detecting a failover scenario (before making a decision whether to failover or not). May and should use some of these placeholders: {failureType}, {instanceType}, {isPrimary}, {isCoPrimary}, {failureDescription}, {command}, {failedHost}, {failureCluster}, {failureClusterDomain}, {failedPort}, {successorHost}, {successorPort}, {successorAlias}, {countReplicas}, {replicaHosts}, {isDowntimed}, {autoPrimaryRecovery}, {autoIntermediatePrimaryRecovery}
+	PreFailoverProcesses                        []string          // Processes to execute before doing a failover (aborting operation should any once of them exits with non-zero code; order of execution undefined). May and should use some of these placeholders: {failureType}, {instanceType}, {isPrimary}, {isCoPrimary}, {failureDescription}, {command}, {failedHost}, {failureCluster}, {failureClusterDomain}, {failedPort}, {countReplicas}, {replicaHosts}, {isDowntimed}
+	PostFailoverProcesses                       []string          // Processes to execute after doing a failover (order of execution undefined). May and should use some of these placeholders: {failureType}, {instanceType}, {isPrimary}, {isCoPrimary}, {failureDescription}, {command}, {failedHost}, {failureCluster}, {failureClusterDomain}, {failedPort}, {successorHost}, {successorPort}, {successorAlias}, {countReplicas}, {replicaHosts}, {isDowntimed}, {isSuccessful}, {lostReplicas}, {countLostReplicas}
+	PostUnsuccessfulFailoverProcesses           []string          // Processes to execute after a not-completely-successful failover (order of execution undefined). May and should use some of these placeholders: {failureType}, {instanceType}, {isPrimary}, {isCoPrimary}, {failureDescription}, {command}, {failedHost}, {failureCluster}, {failureClusterDomain}, {failedPort}, {successorHost}, {successorPort}, {successorAlias}, {countReplicas}, {replicaHosts}, {isDowntimed}, {isSuccessful}, {lostReplicas}, {countLostReplicas}
 	PostPrimaryFailoverProcesses                []string          // Processes to execute after doing a primary failover (order of execution undefined). Uses same placeholders as PostFailoverProcesses
 	PostIntermediatePrimaryFailoverProcesses    []string          // Processes to execute after doing a primary failover (order of execution undefined). Uses same placeholders as PostFailoverProcesses
 	PostTakePrimaryProcesses                    []string          // Processes to execute after a successful Take-Primary event has taken place
diff --git a/go/vt/orchestrator/db/generate_base.go b/go/vt/orchestrator/db/generate_base.go
index b0a9848bde..007e273581 100644
--- a/go/vt/orchestrator/db/generate_base.go
+++ b/go/vt/orchestrator/db/generate_base.go
@@ -272,13 +272,6 @@ var generateSQLBase = []string{
 	`
 		CREATE INDEX resolved_timestamp_idx_hostname_resolve ON hostname_resolve (resolved_timestamp)
 	`,
-	`
-		CREATE TABLE IF NOT EXISTS cluster_alias (
-		  cluster_name varchar(128) CHARACTER SET ascii NOT NULL,
-		  alias varchar(128) NOT NULL,
-		  PRIMARY KEY (cluster_name)
-		) ENGINE=InnoDB DEFAULT CHARSET=ascii
-	`,
 	`
 		CREATE TABLE IF NOT EXISTS active_node (
 		  anchor tinyint unsigned NOT NULL,
@@ -425,7 +418,6 @@ var generateSQLBase = []string{
 			processcing_node_token varchar(128) NOT NULL,
 			analysis varchar(128) NOT NULL,
 			cluster_name varchar(128) NOT NULL,
-			cluster_alias varchar(128) NOT NULL,
 			count_affected_replicas int unsigned NOT NULL,
 			replica_hosts text NOT NULL,
 			PRIMARY KEY (detection_id)
@@ -736,13 +728,6 @@ var generateSQLBase = []string{
 			PRIMARY KEY (disable_recovery)
 		) ENGINE=InnoDB DEFAULT CHARSET=ascii
 	`,
-	`
-		CREATE TABLE IF NOT EXISTS cluster_alias_override (
-			cluster_name varchar(128) CHARACTER SET ascii NOT NULL,
-			alias varchar(128) NOT NULL,
-			PRIMARY KEY (cluster_name)
-		) ENGINE=InnoDB DEFAULT CHARSET=ascii
-	`,
 	`
 		CREATE TABLE IF NOT EXISTS topology_recovery_steps (
 			recovery_step_id bigint unsigned not null auto_increment,
diff --git a/go/vt/orchestrator/db/generate_patches.go b/go/vt/orchestrator/db/generate_patches.go
index aa7c77e657..b50d20b82b 100644
--- a/go/vt/orchestrator/db/generate_patches.go
+++ b/go/vt/orchestrator/db/generate_patches.go
@@ -92,11 +92,6 @@ var generateSQLPatches = []string{
 			database_instance_maintenance
 			ADD KEY active_timestamp_idx (maintenance_active, begin_timestamp)
 	`,
-	`
-		ALTER TABLE
-			cluster_alias
-			ADD UNIQUE KEY alias_uidx (alias)
-	`,
 	`
 		ALTER TABLE
 			database_instance
@@ -122,11 +117,6 @@ var generateSQLPatches = []string{
 			topology_recovery
 			ADD COLUMN cluster_name          varchar(128) CHARACTER SET ascii NOT NULL
 	`,
-	`
-		ALTER TABLE
-			topology_recovery
-			ADD COLUMN cluster_alias         varchar(128) CHARACTER SET ascii NOT NULL
-	`,
 	`
 		ALTER TABLE
 			topology_recovery
@@ -181,19 +171,6 @@ var generateSQLPatches = []string{
 			database_instance
 			ADD COLUMN server_uuid varchar(64) CHARACTER SET ascii NOT NULL AFTER server_id
 	`,
-	`
-		ALTER TABLE
-			database_instance
-			ADD COLUMN suggested_cluster_alias varchar(128) CHARACTER SET ascii NOT NULL AFTER cluster_name
-	`,
-	`
-		ALTER TABLE cluster_alias
-			ADD COLUMN last_registered TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
-	`,
-	`
-		ALTER TABLE cluster_alias
-			ADD KEY last_registered_idx (last_registered)
-	`,
 	`
 		ALTER TABLE
 			topology_recovery
@@ -454,9 +431,6 @@ var generateSQLPatches = []string{
 	`
 		CREATE INDEX end_timestamp_idx_database_instance_downtime ON database_instance_downtime(end_timestamp)
 	`,
-	`
-		CREATE INDEX suggested_cluster_alias_idx_database_instance ON database_instance(suggested_cluster_alias)
-	`,
 	`
 		ALTER TABLE
 			topology_failure_detection
diff --git a/go/vt/orchestrator/http/api.go b/go/vt/orchestrator/http/api.go
index 66f9e70474..b9b5174f64 100644
--- a/go/vt/orchestrator/http/api.go
+++ b/go/vt/orchestrator/http/api.go
@@ -620,26 +620,8 @@ func (httpAPI *API) DetachReplicaPrimaryHost(params martini.Params, r render.Ren
 	Respond(r, &APIResponse{Code: OK, Message: fmt.Sprintf("Replica detached: %+v", instance.Key), Details: instance})
 }
 
-// ReattachReplicaPrimaryHost reverts a detachReplicaPrimaryHost command
-// by resetting the original primary hostname in CHANGE MASTER TO
-func (httpAPI *API) ReattachReplicaPrimaryHost(params martini.Params, r render.Render, req *http.Request, user auth.User) {
-	if !isAuthorizedForAction(req, user) {
-		Respond(r, &APIResponse{Code: ERROR, Message: "Unauthorized"})
-		return
-	}
-	instanceKey, err := httpAPI.getInstanceKey(params["host"], params["port"])
-
-	if err != nil {
-		Respond(r, &APIResponse{Code: ERROR, Message: err.Error()})
-		return
-	}
-	instance, err := inst.ReattachReplicaPrimaryHost(&instanceKey)
-	if err != nil {
-		Respond(r, &APIResponse{Code: ERROR, Message: err.Error()})
-		return
-	}
-
-	Respond(r, &APIResponse{Code: OK, Message: fmt.Sprintf("Replica reattached: %+v", instance.Key), Details: instance})
+func (httpAPI *API) NoOp(params martini.Params, r render.Render, req *http.Request, user auth.User) {
+	Respond(r, &APIResponse{Code: OK, Message: "API Deprecated"})
 }
 
 // EnableGTID attempts to enable GTID on a replica
@@ -1375,13 +1357,7 @@ func (httpAPI *API) Cluster(params martini.Params, r render.Render, req *http.Re
 
 // ClusterByAlias provides list of instances in given cluster
 func (httpAPI *API) ClusterByAlias(params martini.Params, r render.Render, req *http.Request) {
-	clusterName, err := inst.GetClusterByAlias(params["clusterAlias"])
-	if err != nil {
-		Respond(r, &APIResponse{Code: ERROR, Message: fmt.Sprintf("%+v", err)})
-		return
-	}
-
-	params["clusterName"] = clusterName
+	params["clusterName"] = params["clusterAlias"]
 	httpAPI.Cluster(params, r, req)
 }
 
@@ -1421,13 +1397,7 @@ func (httpAPI *API) ClusterInfo(params martini.Params, r render.Render, req *htt
 
 // Cluster provides list of instances in given cluster
 func (httpAPI *API) ClusterInfoByAlias(params martini.Params, r render.Render, req *http.Request) {
-	clusterName, err := inst.GetClusterByAlias(params["clusterAlias"])
-	if err != nil {
-		Respond(r, &APIResponse{Code: ERROR, Message: fmt.Sprintf("%+v", err)})
-		return
-	}
-
-	params["clusterName"] = clusterName
+	params["clusterName"] = params["clusterAlias"]
 	httpAPI.ClusterInfo(params, r, req)
 }
 
@@ -1448,29 +1418,6 @@ func (httpAPI *API) ClusterOSCReplicas(params martini.Params, r render.Render, r
 	r.JSON(http.StatusOK, instances)
 }
 
-// SetClusterAlias will change an alias for a given clustername
-func (httpAPI *API) SetClusterAliasManualOverride(params martini.Params, r render.Render, req *http.Request, user auth.User) {
-	if !isAuthorizedForAction(req, user) {
-		Respond(r, &APIResponse{Code: ERROR, Message: "Unauthorized"})
-		return
-	}
-	clusterName := params["clusterName"]
-	alias := req.URL.Query().Get("alias")
-
-	var err error
-	if orcraft.IsRaftEnabled() {
-		_, err = orcraft.PublishCommand("set-cluster-alias-manual-override", []string{clusterName, alias})
-	} else {
-		err = inst.SetClusterAliasManualOverride(clusterName, alias)
-	}
-
-	if err != nil {
-		Respond(r, &APIResponse{Code: ERROR, Message: fmt.Sprintf("%+v", err)})
-		return
-	}
-	Respond(r, &APIResponse{Code: OK, Message: fmt.Sprintf("Cluster %s now has alias '%s'", clusterName, alias)})
-}
-
 // Clusters provides list of known clusters
 func (httpAPI *API) Clusters(params martini.Params, r render.Render, req *http.Request) {
 	clusterNames, err := inst.ReadClusters()
@@ -1878,11 +1825,7 @@ func (httpAPI *API) GetHeuristicClusterPoolInstancesLag(params martini.Params, r
 		Respond(r, &APIResponse{Code: ERROR, Message: "Unauthorized"})
 		return
 	}
-	clusterName, err := inst.ReadClusterNameByAlias(params["clusterName"])
-	if err != nil {
-		Respond(r, &APIResponse{Code: ERROR, Message: fmt.Sprintf("%+v", err)})
-		return
-	}
+	clusterName := params["clusterName"]
 	pool := params["pool"]
 
 	lag, err := inst.GetHeuristicClusterPoolInstancesLag(clusterName, pool)
@@ -2199,12 +2142,7 @@ func (httpAPI *API) ReplicationAnalysis(params martini.Params, r render.Render,
 
 // ReplicationAnalysis retuens list of issues
 func (httpAPI *API) ReplicationAnalysisForCluster(params martini.Params, r render.Render, req *http.Request) {
-	var clusterName string
-	var err error
-	if clusterName, err = inst.DeduceClusterName(params["clusterName"]); err != nil {
-		Respond(r, &APIResponse{Code: ERROR, Message: fmt.Sprintf("Cannot get analysis: %+v", err)})
-		return
-	}
+	clusterName := params["clusterName"]
 	if clusterName == "" {
 		Respond(r, &APIResponse{Code: ERROR, Message: fmt.Sprintf("Cannot get cluster name: %+v", params["clusterName"])})
 		return
@@ -2415,7 +2353,7 @@ func (httpAPI *API) AuditFailureDetection(params martini.Params, r render.Render
 		if derr != nil || page < 0 {
 			page = 0
 		}
-		audits, err = logic.ReadRecentFailureDetections(params["clusterAlias"], page)
+		audits, err = logic.ReadRecentFailureDetections(params["clusterName"], page)
 	}
 
 	if err != nil {
@@ -2467,7 +2405,7 @@ func (httpAPI *API) AuditRecovery(params martini.Params, r render.Render, req *h
 		}
 		unacknowledgedOnly := (req.URL.Query().Get("unacknowledged") == "true")
 
-		audits, err = logic.ReadRecentRecoveries(params["clusterName"], params["clusterAlias"], unacknowledgedOnly, page)
+		audits, err = logic.ReadRecentRecoveries(params["clusterName"], unacknowledgedOnly, page)
 	}
 
 	if err != nil {
@@ -2530,7 +2468,7 @@ func (httpAPI *API) AcknowledgeClusterRecoveries(params martini.Params, r render
 	var clusterName string
 	var err error
 	if params["clusterAlias"] != "" {
-		clusterName, err = inst.GetClusterByAlias(params["clusterAlias"])
+		clusterName = params["clusterAlias"]
 	} else {
 		clusterName, err = figureClusterName(getClusterHint(params))
 	}
@@ -2813,9 +2751,9 @@ func (httpAPI *API) RegisterRequests(m *martini.ClassicMartini) {
 	httpAPI.registerAPIRequest(m, "stop-replica-nice/:host/:port", httpAPI.StopReplicationNicely)
 	httpAPI.registerAPIRequest(m, "reset-replica/:host/:port", httpAPI.ResetReplication)
 	httpAPI.registerAPIRequest(m, "detach-replica/:host/:port", httpAPI.DetachReplicaPrimaryHost)
-	httpAPI.registerAPIRequest(m, "reattach-replica/:host/:port", httpAPI.ReattachReplicaPrimaryHost)
+	httpAPI.registerAPIRequest(m, "reattach-replica/:host/:port", httpAPI.NoOp)
 	httpAPI.registerAPIRequest(m, "detach-replica-primary-host/:host/:port", httpAPI.DetachReplicaPrimaryHost)
-	httpAPI.registerAPIRequest(m, "reattach-replica-primary-host/:host/:port", httpAPI.ReattachReplicaPrimaryHost)
+	httpAPI.registerAPIRequest(m, "reattach-replica-primary-host/:host/:port", httpAPI.NoOp)
 	httpAPI.registerAPIRequest(m, "flush-binary-logs/:host/:port", httpAPI.FlushBinaryLogs)
 	httpAPI.registerAPIRequest(m, "purge-binary-logs/:host/:port/:logFile", httpAPI.PurgeBinaryLogs)
 	httpAPI.registerAPIRequest(m, "restart-replica-statements/:host/:port", httpAPI.RestartReplicationStatements)
@@ -2849,7 +2787,7 @@ func (httpAPI *API) RegisterRequests(m *martini.ClassicMartini) {
 	httpAPI.registerAPIRequest(m, "cluster-info/:clusterHint", httpAPI.ClusterInfo)
 	httpAPI.registerAPIRequest(m, "cluster-info/alias/:clusterAlias", httpAPI.ClusterInfoByAlias)
 	httpAPI.registerAPIRequest(m, "cluster-osc-replicas/:clusterHint", httpAPI.ClusterOSCReplicas)
-	httpAPI.registerAPIRequest(m, "set-cluster-alias/:clusterName", httpAPI.SetClusterAliasManualOverride)
+	httpAPI.registerAPIRequest(m, "set-cluster-alias/:clusterName", httpAPI.NoOp)
 	httpAPI.registerAPIRequest(m, "clusters", httpAPI.Clusters)
 	httpAPI.registerAPIRequest(m, "clusters-info", httpAPI.ClustersInfo)
 
diff --git a/go/vt/orchestrator/http/web.go b/go/vt/orchestrator/http/web.go
index 671bb1beaf..b784f46866 100644
--- a/go/vt/orchestrator/http/web.go
+++ b/go/vt/orchestrator/http/web.go
@@ -113,14 +113,7 @@ func (httpWeb *Web) Cluster(params martini.Params, r render.Render, req *http.Re
 }
 
 func (httpWeb *Web) ClusterByAlias(params martini.Params, r render.Render, req *http.Request, user auth.User) {
-	clusterName, err := inst.GetClusterByAlias(params["clusterAlias"])
-	// Willing to accept the case of multiple clusters; we just present one
-	if clusterName == "" && err != nil {
-		r.JSON(200, &APIResponse{Code: ERROR, Message: fmt.Sprintf("%+v", err)})
-		return
-	}
-
-	params["clusterName"] = clusterName
+	params["clusterName"] = params["clusterAlias"]
 	httpWeb.Cluster(params, r, req, user)
 }
 
diff --git a/go/vt/orchestrator/inst/analysis.go b/go/vt/orchestrator/inst/analysis.go
index ce0a29bd73..efb6a8a0fc 100644
--- a/go/vt/orchestrator/inst/analysis.go
+++ b/go/vt/orchestrator/inst/analysis.go
@@ -118,7 +118,6 @@ type ReplicationAnalysis struct {
 	AnalyzedInstancePrimaryKey                InstanceKey
 	TabletType                                topodatapb.TabletType
 	PrimaryTimeStamp                          time.Time
-	SuggestedClusterAlias                     string
 	ClusterDetails                            ClusterInfo
 	AnalyzedInstanceDataCenter                string
 	AnalyzedInstanceRegion                    string
diff --git a/go/vt/orchestrator/inst/analysis_dao.go b/go/vt/orchestrator/inst/analysis_dao.go
index 7ace24cf31..cebcb09ffc 100644
--- a/go/vt/orchestrator/inst/analysis_dao.go
+++ b/go/vt/orchestrator/inst/analysis_dao.go
@@ -88,7 +88,6 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 		MIN(primary_instance.cluster_name) AS cluster_name,
 		MIN(primary_instance.binary_log_file) AS binary_log_file,
 		MIN(primary_instance.binary_log_pos) AS binary_log_pos,
-		MIN(primary_instance.suggested_cluster_alias) AS suggested_cluster_alias,
 		MIN(primary_tablet.info) AS primary_tablet_info,
 		MIN(
 			IFNULL(
@@ -98,12 +97,6 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 				0
 			)
 		) AS is_stale_binlog_coordinates,
-		MIN(
-			IFNULL(
-				cluster_alias.alias,
-				primary_instance.cluster_name
-			)
-		) AS cluster_alias,
 		MIN(
 			IFNULL(
 				cluster_domain_name.domain_name,
@@ -129,13 +122,6 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 			)
 		) AS is_primary,
 		MIN(primary_instance.is_co_primary) AS is_co_primary,
-		MIN(
-			CONCAT(
-				primary_instance.hostname,
-				':',
-				primary_instance.port
-			) = primary_instance.cluster_name
-		) AS is_cluster_primary,
 		MIN(primary_instance.gtid_mode) AS gtid_mode,
 		COUNT(replica_instance.server_id) AS count_replicas,
 		IFNULL(
@@ -353,9 +339,6 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 			AND replica_instance.port = replica_downtime.port
 			AND replica_downtime.downtime_active = 1
 		)
-		LEFT JOIN cluster_alias ON (
-			cluster_alias.cluster_name = primary_instance.cluster_name
-		)
 		LEFT JOIN cluster_domain_name ON (
 			cluster_domain_name.cluster_name = primary_instance.cluster_name
 		)
@@ -415,9 +398,7 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 			Type:    BinaryLog,
 		}
 		isStaleBinlogCoordinates := m.GetBool("is_stale_binlog_coordinates")
-		a.SuggestedClusterAlias = m.GetString("suggested_cluster_alias")
 		a.ClusterDetails.ClusterName = m.GetString("cluster_name")
-		a.ClusterDetails.ClusterAlias = m.GetString("cluster_alias")
 		a.ClusterDetails.ClusterDomain = m.GetString("cluster_domain")
 		a.GTIDMode = m.GetString("gtid_mode")
 		a.LastCheckValid = m.GetBool("is_last_check_valid")
@@ -476,11 +457,11 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 				log.Infof(analysisMessage)
 			}
 		}
-		if clusters[a.SuggestedClusterAlias] == nil {
-			clusters[a.SuggestedClusterAlias] = &clusterAnalysis{}
+		if clusters[a.ClusterDetails.ClusterName] == nil {
+			clusters[a.ClusterDetails.ClusterName] = &clusterAnalysis{}
 			if a.TabletType == topodatapb.TabletType_PRIMARY {
 				a.IsClusterPrimary = true
-				clusters[a.SuggestedClusterAlias].primaryKey = &a.AnalyzedInstanceKey
+				clusters[a.ClusterDetails.ClusterName].primaryKey = &a.AnalyzedInstanceKey
 			}
 			durabilityPolicy := m.GetString("durability_policy")
 			if durabilityPolicy == "" {
@@ -492,10 +473,10 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 				log.Errorf("can't get the durability policy %v - %v. Skipping keyspace - %v.", durabilityPolicy, err, a.AnalyzedKeyspace)
 				return nil
 			}
-			clusters[a.SuggestedClusterAlias].durability = durability
+			clusters[a.ClusterDetails.ClusterName].durability = durability
 		}
 		// ca has clusterwide info
-		ca := clusters[a.SuggestedClusterAlias]
+		ca := clusters[a.ClusterDetails.ClusterName]
 		if ca.hasClusterwideAction {
 			// We can only take one cluster level action at a time.
 			return nil
diff --git a/go/vt/orchestrator/inst/cluster.go b/go/vt/orchestrator/inst/cluster.go
index 6e2abfaf7e..a20ce548e2 100644
--- a/go/vt/orchestrator/inst/cluster.go
+++ b/go/vt/orchestrator/inst/cluster.go
@@ -18,30 +18,13 @@ package inst
 
 import (
 	"regexp"
-	"strings"
 
 	"vitess.io/vitess/go/vt/orchestrator/config"
 )
 
-// mappedClusterNameToAlias attempts to match a cluster with an alias based on
-// configured ClusterNameToAlias map
-func mappedClusterNameToAlias(clusterName string) string {
-	for pattern, alias := range config.Config.ClusterNameToAlias {
-		if pattern == "" {
-			// sanity
-			continue
-		}
-		if matched, _ := regexp.MatchString(pattern, clusterName); matched {
-			return alias
-		}
-	}
-	return ""
-}
-
 // ClusterInfo makes for a cluster status/info summary
 type ClusterInfo struct {
 	ClusterName                             string
-	ClusterAlias                            string // Human friendly alias
 	ClusterDomain                           string // CNAME/VIP/A-record/whatever of the primary of this cluster
 	CountInstances                          uint
 	HeuristicLag                            int64
@@ -61,22 +44,7 @@ func (clusterInfo *ClusterInfo) filtersMatchCluster(filters []string) bool {
 		if filter == clusterInfo.ClusterName {
 			return true
 		}
-		if filter == clusterInfo.ClusterAlias {
-			return true
-		}
-		if strings.HasPrefix(filter, "alias=") {
-			// Match by exact cluster alias name
-			alias := strings.SplitN(filter, "=", 2)[1]
-			if alias == clusterInfo.ClusterAlias {
-				return true
-			}
-		} else if strings.HasPrefix(filter, "alias~=") {
-			// Match by cluster alias regex
-			aliasPattern := strings.SplitN(filter, "~=", 2)[1]
-			if matched, _ := regexp.MatchString(aliasPattern, clusterInfo.ClusterAlias); matched {
-				return true
-			}
-		} else if filter == "*" {
+		if filter == "*" {
 			return true
 		} else if matched, _ := regexp.MatchString(filter, clusterInfo.ClusterName); matched && filter != "" {
 			return true
@@ -84,14 +52,3 @@ func (clusterInfo *ClusterInfo) filtersMatchCluster(filters []string) bool {
 	}
 	return false
 }
-
-// ApplyClusterAlias updates the given clusterInfo's ClusterAlias property
-func (clusterInfo *ClusterInfo) ApplyClusterAlias() {
-	if clusterInfo.ClusterAlias != "" && clusterInfo.ClusterAlias != clusterInfo.ClusterName {
-		// Already has an alias; abort
-		return
-	}
-	if alias := mappedClusterNameToAlias(clusterInfo.ClusterName); alias != "" {
-		clusterInfo.ClusterAlias = alias
-	}
-}
diff --git a/go/vt/orchestrator/inst/cluster_alias.go b/go/vt/orchestrator/inst/cluster_alias.go
deleted file mode 100644
index ccb3a1bf0b..0000000000
--- a/go/vt/orchestrator/inst/cluster_alias.go
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
-   Copyright 2015 Shlomi Noach, courtesy Booking.com
-
-   Licensed under the Apache License, Version 2.0 (the "License");
-   you may not use this file except in compliance with the License.
-   You may obtain a copy of the License at
-
-       http://www.apache.org/licenses/LICENSE-2.0
-
-   Unless required by applicable law or agreed to in writing, software
-   distributed under the License is distributed on an "AS IS" BASIS,
-   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-   See the License for the specific language governing permissions and
-   limitations under the License.
-*/
-
-package inst
-
-// SetClusterAlias will write (and override) a single cluster name mapping
-func SetClusterAlias(clusterName string, alias string) error {
-	return writeClusterAlias(clusterName, alias)
-}
-
-// SetClusterAliasManualOverride will write (and override) a single cluster name mapping
-func SetClusterAliasManualOverride(clusterName string, alias string) error {
-	return writeClusterAliasManualOverride(clusterName, alias)
-}
-
-// GetClusterByAlias returns the cluster name associated with given alias.
-// The function returns with error when:
-// - No cluster is associated with the alias
-// - More than one cluster is associated with the alias
-func GetClusterByAlias(alias string) (string, error) {
-	return ReadClusterNameByAlias(alias)
-}
diff --git a/go/vt/orchestrator/inst/cluster_alias_dao.go b/go/vt/orchestrator/inst/cluster_alias_dao.go
deleted file mode 100644
index 552f0d5828..0000000000
--- a/go/vt/orchestrator/inst/cluster_alias_dao.go
+++ /dev/null
@@ -1,245 +0,0 @@
-/*
-   Copyright 2015 Shlomi Noach, courtesy Booking.com
-
-   Licensed under the Apache License, Version 2.0 (the "License");
-   you may not use this file except in compliance with the License.
-   You may obtain a copy of the License at
-
-       http://www.apache.org/licenses/LICENSE-2.0
-
-   Unless required by applicable law or agreed to in writing, software
-   distributed under the License is distributed on an "AS IS" BASIS,
-   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-   See the License for the specific language governing permissions and
-   limitations under the License.
-*/
-
-package inst
-
-import (
-	"fmt"
-
-	"vitess.io/vitess/go/vt/log"
-
-	"vitess.io/vitess/go/vt/orchestrator/db"
-	"vitess.io/vitess/go/vt/orchestrator/external/golib/sqlutils"
-)
-
-// ReadClusterNameByAlias
-func ReadClusterNameByAlias(alias string) (clusterName string, err error) {
-	query := `
-		select
-			cluster_name
-		from
-			cluster_alias
-		where
-			alias = ?
-			or cluster_name = ?
-		`
-	err = db.QueryOrchestrator(query, sqlutils.Args(alias, alias), func(m sqlutils.RowMap) error {
-		clusterName = m.GetString("cluster_name")
-		return nil
-	})
-	if err != nil {
-		return "", err
-	}
-	if clusterName == "" {
-		err = fmt.Errorf("No cluster found for alias %s", alias)
-	}
-	return clusterName, err
-}
-
-// DeduceClusterName attempts to resolve a cluster name given a name or alias.
-// If unsuccessful to match by alias, the function returns the same given string
-func DeduceClusterName(nameOrAlias string) (clusterName string, err error) {
-	if nameOrAlias == "" {
-		return "", fmt.Errorf("empty cluster name")
-	}
-	if name, err := ReadClusterNameByAlias(nameOrAlias); err == nil {
-		return name, nil
-	}
-	return nameOrAlias, nil
-}
-
-// ReadAliasByClusterName returns the cluster alias for the given cluster name,
-// or the cluster name itself if not explicit alias found
-func ReadAliasByClusterName(clusterName string) (alias string, err error) {
-	alias = clusterName // default return value
-	query := `
-		select
-			alias
-		from
-			cluster_alias
-		where
-			cluster_name = ?
-		`
-	err = db.QueryOrchestrator(query, sqlutils.Args(clusterName), func(m sqlutils.RowMap) error {
-		alias = m.GetString("alias")
-		return nil
-	})
-	return clusterName, err
-}
-
-// WriteClusterAlias will write (and override) a single cluster name mapping
-func writeClusterAlias(clusterName string, alias string) error {
-	writeFunc := func() error {
-		_, err := db.ExecOrchestrator(`
-			replace into
-					cluster_alias (cluster_name, alias, last_registered)
-				values
-					(?, ?, now())
-			`,
-			clusterName, alias)
-		if err != nil {
-			log.Error(err)
-		}
-		return err
-	}
-	return ExecDBWriteFunc(writeFunc)
-}
-
-// writeClusterAliasManualOverride will write (and override) a single cluster name mapping
-func writeClusterAliasManualOverride(clusterName string, alias string) error {
-	writeFunc := func() error {
-		_, err := db.ExecOrchestrator(`
-			replace into
-					cluster_alias_override (cluster_name, alias)
-				values
-					(?, ?)
-			`,
-			clusterName, alias)
-		if err != nil {
-			log.Error(err)
-		}
-		return err
-	}
-	return ExecDBWriteFunc(writeFunc)
-}
-
-// UpdateClusterAliases writes down the cluster_alias table based on information
-// gained from database_instance
-func UpdateClusterAliases() error {
-	writeFunc := func() error {
-		_, err := db.ExecOrchestrator(`
-			replace into
-					cluster_alias (alias, cluster_name, last_registered)
-				select
-				    suggested_cluster_alias,
-						cluster_name,
-						now()
-					from
-				    database_instance
-				    left join database_instance_downtime using (hostname, port)
-				  where
-				    suggested_cluster_alias!=''
-						/* exclude newly demoted, downtimed primaries */
-						and ifnull(
-								database_instance_downtime.downtime_active = 1
-								and database_instance_downtime.end_timestamp > now()
-								and database_instance_downtime.reason = ?
-							, 0) = 0
-					order by
-						ifnull(last_checked <= last_seen, 0) asc,
-						read_only desc,
-						num_replica_hosts asc
-			`, DowntimeLostInRecoveryMessage)
-		if err != nil {
-			log.Error(err)
-		}
-		return err
-	}
-	if err := ExecDBWriteFunc(writeFunc); err != nil {
-		return err
-	}
-	writeFunc = func() error {
-		// Handling the case where no cluster alias exists: we write a dummy alias in the form of the real cluster name.
-		_, err := db.ExecOrchestrator(`
-			replace into
-					cluster_alias (alias, cluster_name, last_registered)
-				select
-						cluster_name as alias, cluster_name, now()
-				  from
-				    database_instance
-				  group by
-				    cluster_name
-					having
-						sum(suggested_cluster_alias = '') = count(*)
-			`)
-		if err != nil {
-			log.Error(err)
-		}
-		return err
-	}
-	if err := ExecDBWriteFunc(writeFunc); err != nil {
-		return err
-	}
-	return nil
-}
-
-// ReplaceAliasClusterName replaces alis mapping of one cluster name onto a new cluster name.
-// Used in topology failover/recovery
-func ReplaceAliasClusterName(oldClusterName string, newClusterName string) (err error) {
-	{
-		writeFunc := func() error {
-			_, err := db.ExecOrchestrator(`
-			update cluster_alias
-				set cluster_name = ?
-				where cluster_name = ?
-			`,
-				newClusterName, oldClusterName)
-			if err != nil {
-				log.Error(err)
-			}
-			return err
-		}
-		err = ExecDBWriteFunc(writeFunc)
-	}
-	{
-		writeFunc := func() error {
-			_, err := db.ExecOrchestrator(`
-			update cluster_alias_override
-				set cluster_name = ?
-				where cluster_name = ?
-			`,
-				newClusterName, oldClusterName)
-			if err != nil {
-				log.Error(err)
-			}
-			return err
-		}
-		if ferr := ExecDBWriteFunc(writeFunc); ferr != nil {
-			err = ferr
-		}
-	}
-	return err
-}
-
-// ReadUnambiguousSuggestedClusterAliases reads potential primary hostname:port who have suggested cluster aliases,
-// where no one else shares said suggested cluster alias. Such hostname:port are likely true owners
-// of the alias.
-func ReadUnambiguousSuggestedClusterAliases() (result map[string]InstanceKey, err error) {
-	result = map[string]InstanceKey{}
-
-	query := `
-		select
-			suggested_cluster_alias,
-			min(hostname) as hostname,
-			min(port) as port
-		from
-			database_instance
-		where
-			suggested_cluster_alias != ''
-			and replication_depth=0
-		group by
-			suggested_cluster_alias
-		having
-			count(*) = 1
-		`
-	err = db.QueryOrchestrator(query, sqlutils.Args(), func(m sqlutils.RowMap) error {
-		key := InstanceKey{Hostname: m.GetString("hostname"), Port: m.GetInt("port")}
-		suggestedAlias := m.GetString("suggested_cluster_alias")
-		result[suggestedAlias] = key
-		return nil
-	})
-	return result, err
-}
diff --git a/go/vt/orchestrator/inst/downtime_dao.go b/go/vt/orchestrator/inst/downtime_dao.go
index 8f41da5631..fa39d050f2 100644
--- a/go/vt/orchestrator/inst/downtime_dao.go
+++ b/go/vt/orchestrator/inst/downtime_dao.go
@@ -142,10 +142,6 @@ func expireLostInRecoveryDowntime() error {
 	if len(instances) == 0 {
 		return nil
 	}
-	unambiguousAliases, err := ReadUnambiguousSuggestedClusterAliases()
-	if err != nil {
-		return err
-	}
 	for _, instance := range instances {
 		// We _may_ expire this downtime, but only after a minute
 		// This is a graceful period, during which other servers can claim ownership of the alias,
@@ -156,21 +152,8 @@ func expireLostInRecoveryDowntime() error {
 		if !instance.IsLastCheckValid {
 			continue
 		}
-		endDowntime := false
 		if instance.ReplicaRunning() {
 			// back, alive, replicating in some topology
-			endDowntime = true
-		} else if instance.ReplicationDepth == 0 {
-			// instance makes the appearance of a primary
-			if unambiguousKey, ok := unambiguousAliases[instance.SuggestedClusterAlias]; ok {
-				if unambiguousKey.Equals(&instance.Key) {
-					// This instance seems to be a primary, which is valid, and has a suggested alias,
-					// and is the _only_ one to have this suggested alias (i.e. no one took its place)
-					endDowntime = true
-				}
-			}
-		}
-		if endDowntime {
 			if _, err := EndDowntime(&instance.Key); err != nil {
 				return err
 			}
diff --git a/go/vt/orchestrator/inst/instance.go b/go/vt/orchestrator/inst/instance.go
index d889f29752..8fbf48eb7e 100644
--- a/go/vt/orchestrator/inst/instance.go
+++ b/go/vt/orchestrator/inst/instance.go
@@ -81,7 +81,6 @@ type Instance struct {
 	ReplicationLagSeconds              sql.NullInt64
 	Replicas                           InstanceKeyMap
 	ClusterName                        string
-	SuggestedClusterAlias              string
 	DataCenter                         string
 	Region                             string
 	PhysicalEnvironment                string
diff --git a/go/vt/orchestrator/inst/instance_dao.go b/go/vt/orchestrator/inst/instance_dao.go
index 47e03dee4c..de15f7ea45 100644
--- a/go/vt/orchestrator/inst/instance_dao.go
+++ b/go/vt/orchestrator/inst/instance_dao.go
@@ -297,6 +297,7 @@ func ReadTopologyInstanceBufferable(instanceKey *InstanceKey, bufferWrites bool,
 	if err != nil {
 		goto Cleanup
 	}
+	instance.ClusterName = fmt.Sprintf("%v:%v", tablet.Keyspace, tablet.Shard)
 
 	fullStatus, err = FullStatus(*instanceKey)
 	if err != nil {
@@ -634,9 +635,6 @@ func ReadTopologyInstanceBufferable(instanceKey *InstanceKey, bufferWrites bool,
 	err = RegisterCandidateInstance(NewCandidateDatabaseInstance(instanceKey, instance.PromotionRule).WithCurrentTime())
 	_ = logReadTopologyInstanceError(instanceKey, "RegisterCandidateInstance", err)
 
-	// TODO(sougou): delete cluster_alias_override metadata
-	instance.SuggestedClusterAlias = fmt.Sprintf("%v:%v", tablet.Keyspace, tablet.Shard)
-
 	if instance.ReplicationDepth == 0 && config.Config.DetectClusterDomainQuery != "" {
 		// Only need to do on primary tablets
 		domainName := ""
@@ -776,7 +774,6 @@ func ReadReplicationGroupPrimary(instance *Instance) (err error) {
 // instances.
 func ReadInstanceClusterAttributes(instance *Instance) (err error) {
 	var primaryOrGroupPrimaryInstanceKey InstanceKey
-	var primaryOrGroupPrimaryClusterName string
 	var primaryOrGroupPrimaryReplicationDepth uint
 	var ancestryUUID string
 	var primaryOrGroupPrimaryExecutedGtidSet string
@@ -785,8 +782,6 @@ func ReadInstanceClusterAttributes(instance *Instance) (err error) {
 	// Read the cluster_name of the _primary_ or _group_primary_ of our instance, derive it from there.
 	query := `
 			select
-					cluster_name,
-					suggested_cluster_alias,
 					replication_depth,
 					source_host,
 					source_port,
@@ -805,7 +800,6 @@ func ReadInstanceClusterAttributes(instance *Instance) (err error) {
 	}
 	args := sqlutils.Args(primaryOrGroupPrimaryInstanceKey.Hostname, primaryOrGroupPrimaryInstanceKey.Port)
 	err = db.QueryOrchestrator(query, args, func(m sqlutils.RowMap) error {
-		primaryOrGroupPrimaryClusterName = m.GetString("cluster_name")
 		primaryOrGroupPrimaryReplicationDepth = m.GetUint("replication_depth")
 		primaryOrGroupPrimaryInstanceKey.Hostname = m.GetString("source_host")
 		primaryOrGroupPrimaryInstanceKey.Port = m.GetInt("source_port")
@@ -820,34 +814,14 @@ func ReadInstanceClusterAttributes(instance *Instance) (err error) {
 	}
 
 	var replicationDepth uint
-	var clusterName string
 	if primaryOrGroupPrimaryDataFound {
 		replicationDepth = primaryOrGroupPrimaryReplicationDepth + 1
-		clusterName = primaryOrGroupPrimaryClusterName
-	}
-	clusterNameByInstanceKey := instance.Key.StringCode()
-	if clusterName == "" {
-		// Nothing from primary; we set it to be named after the instance itself
-		clusterName = clusterNameByInstanceKey
 	}
-
 	isCoPrimary := false
 	if primaryOrGroupPrimaryInstanceKey.Equals(&instance.Key) {
 		// co-primary calls for special case, in fear of the infinite loop
 		isCoPrimary = true
-		clusterNameByCoPrimaryKey := instance.SourceKey.StringCode()
-		if clusterName != clusterNameByInstanceKey && clusterName != clusterNameByCoPrimaryKey {
-			// Can be caused by a co-primary topology failover
-			log.Errorf("ReadInstanceClusterAttributes: in co-primary topology %s is not in (%s, %s). Forcing it to become one of them", clusterName, clusterNameByInstanceKey, clusterNameByCoPrimaryKey)
-			clusterName = math.TernaryString(instance.Key.SmallerThan(&instance.SourceKey), clusterNameByInstanceKey, clusterNameByCoPrimaryKey)
-		}
-		if clusterName == clusterNameByInstanceKey {
-			// circular replication. Avoid infinite ++ on replicationDepth
-			replicationDepth = 0
-			ancestryUUID = ""
-		} // While the other stays "1"
-	}
-	instance.ClusterName = clusterName
+	}
 	instance.ReplicationDepth = replicationDepth
 	instance.IsCoPrimary = isCoPrimary
 	instance.AncestryUUID = ancestryUUID
@@ -944,7 +918,6 @@ func readInstanceRow(m sqlutils.RowMap) *Instance {
 	instance.SQLDelay = m.GetUint("sql_delay")
 	replicasJSON := m.GetString("replica_hosts")
 	instance.ClusterName = m.GetString("cluster_name")
-	instance.SuggestedClusterAlias = m.GetString("suggested_cluster_alias")
 	instance.DataCenter = m.GetString("data_center")
 	instance.Region = m.GetString("region")
 	instance.PhysicalEnvironment = m.GetString("physical_environment")
@@ -1035,6 +1008,7 @@ func readInstancesByCondition(condition string, args []any, sort string) ([](*In
     	ifnull(database_instance_downtime.end_timestamp, '') as downtime_end_timestamp
 		from
 			database_instance
+			left join vitess_tablet using (hostname, port)
 			left join candidate_database_instance using (hostname, port)
 			left join hostname_unresolve using (hostname)
 			left join database_instance_downtime using (hostname, port)
@@ -1109,8 +1083,9 @@ func ReadClusterWriteablePrimary(clusterName string) ([](*Instance), error) {
 		cluster_name = ?
 		and read_only = 0
 		and (replication_depth = 0 or is_co_primary)
+		and tablet_type = ?
 	`
-	return readInstancesByCondition(condition, sqlutils.Args(clusterName), "replication_depth asc")
+	return readInstancesByCondition(condition, sqlutils.Args(clusterName, topodatapb.TabletType_PRIMARY), "replication_depth asc")
 }
 
 // ReadClusterPrimary returns the primary of this cluster.
@@ -1120,8 +1095,9 @@ func ReadClusterPrimary(clusterName string) ([](*Instance), error) {
 	condition := `
 		cluster_name = ?
 		and (replication_depth = 0 or is_co_primary)
+		and tablet_type = ?
 	`
-	return readInstancesByCondition(condition, sqlutils.Args(clusterName), "read_only asc, replication_depth asc")
+	return readInstancesByCondition(condition, sqlutils.Args(clusterName, topodatapb.TabletType_PRIMARY), "read_only asc, replication_depth asc")
 }
 
 // ReadWriteableClustersPrimaries returns writeable primaries of all clusters, but only one
@@ -1130,8 +1106,9 @@ func ReadWriteableClustersPrimaries() (instances [](*Instance), err error) {
 	condition := `
 		read_only = 0
 		and (replication_depth = 0 or is_co_primary)
+		and tablet_type = ?
 	`
-	allPrimaries, err := readInstancesByCondition(condition, sqlutils.Args(), "cluster_name asc, replication_depth asc")
+	allPrimaries, err := readInstancesByCondition(condition, sqlutils.Args(topodatapb.TabletType_PRIMARY), "cluster_name asc, replication_depth asc")
 	if err != nil {
 		return instances, err
 	}
@@ -1236,11 +1213,10 @@ func SearchInstances(searchString string) ([](*Instance), error) {
 			or instr(version, ?) > 0
 			or instr(version_comment, ?) > 0
 			or instr(concat(hostname, ':', port), ?) > 0
-			or instr(suggested_cluster_alias, ?) > 0
 			or concat(server_id, '') = ?
 			or concat(port, '') = ?
 		`
-	args := sqlutils.Args(searchString, searchString, searchString, searchString, searchString, searchString, searchString, searchString)
+	args := sqlutils.Args(searchString, searchString, searchString, searchString, searchString, searchString, searchString)
 	return readInstancesByCondition(condition, args, `replication_depth asc, num_replica_hosts desc, cluster_name, hostname, port`)
 }
 
@@ -1569,59 +1545,6 @@ func GetHeuristicClusterPoolInstancesLag(clusterName string, pool string) (int64
 	return GetInstancesMaxLag(instances)
 }
 
-// updateInstanceClusterName
-func updateInstanceClusterName(instance *Instance) error {
-	writeFunc := func() error {
-		_, err := db.ExecOrchestrator(`
-			update
-				database_instance
-			set
-				cluster_name=?
-			where
-				hostname=? and port=?
-        	`, instance.ClusterName, instance.Key.Hostname, instance.Key.Port,
-		)
-		if err != nil {
-			log.Error(err)
-			return err
-		}
-		_ = AuditOperation("update-cluster-name", &instance.Key, fmt.Sprintf("set to %s", instance.ClusterName))
-		return nil
-	}
-	return ExecDBWriteFunc(writeFunc)
-}
-
-// ReviewUnseenInstances reviews instances that have not been seen (suposedly dead) and updates some of their data
-func ReviewUnseenInstances() error {
-	instances, err := ReadUnseenInstances()
-	if err != nil {
-		log.Error(err)
-		return err
-	}
-	operations := 0
-	for _, instance := range instances {
-		instance := instance
-
-		primaryHostname, err := ResolveHostname(instance.SourceKey.Hostname)
-		if err != nil {
-			log.Error(err)
-			continue
-		}
-		instance.SourceKey.Hostname = primaryHostname
-		savedClusterName := instance.ClusterName
-
-		if err := ReadInstanceClusterAttributes(instance); err != nil {
-			log.Error(err)
-		} else if instance.ClusterName != savedClusterName {
-			_ = updateInstanceClusterName(instance)
-			operations++
-		}
-	}
-
-	_ = AuditOperation("review-unseen-instances", nil, fmt.Sprintf("Operations: %d", operations))
-	return err
-}
-
 // readUnseenPrimaryKeys will read list of primaries that have never been seen, and yet whose replicas
 // seem to be replicating.
 func readUnseenPrimaryKeys() ([]InstanceKey, error) {
@@ -1923,11 +1846,9 @@ func ReadClustersInfo(clusterName string) ([]ClusterInfo, error) {
 		select
 			cluster_name,
 			count(*) as count_instances,
-			ifnull(min(alias), cluster_name) as alias,
 			ifnull(min(domain_name), '') as domain_name
 		from
 			database_instance
-			left join cluster_alias using (cluster_name)
 			left join cluster_domain_name using (cluster_name)
 		%s
 		group by
@@ -1937,10 +1858,8 @@ func ReadClustersInfo(clusterName string) ([]ClusterInfo, error) {
 		clusterInfo := ClusterInfo{
 			ClusterName:    m.GetString("cluster_name"),
 			CountInstances: m.GetUint("count_instances"),
-			ClusterAlias:   m.GetString("alias"),
 			ClusterDomain:  m.GetString("domain_name"),
 		}
-		clusterInfo.ApplyClusterAlias()
 		clusterInfo.ReadRecoveryInfo()
 
 		clusters = append(clusters, clusterInfo)
@@ -2149,7 +2068,6 @@ func mkInsertOdkuForInstances(instances []*Instance, instanceWasActuallyFound bo
 		"num_replica_hosts",
 		"replica_hosts",
 		"cluster_name",
-		"suggested_cluster_alias",
 		"data_center",
 		"region",
 		"physical_environment",
@@ -2239,7 +2157,6 @@ func mkInsertOdkuForInstances(instances []*Instance, instanceWasActuallyFound bo
 		args = append(args, len(instance.Replicas))
 		args = append(args, instance.Replicas.ToJSONString())
 		args = append(args, instance.ClusterName)
-		args = append(args, instance.SuggestedClusterAlias)
 		args = append(args, instance.DataCenter)
 		args = append(args, instance.Region)
 		args = append(args, instance.PhysicalEnvironment)
@@ -2692,14 +2609,6 @@ func FigureClusterName(clusterHint string, instanceKey *InstanceKey, thisInstanc
 		if clusterInfo, err := ReadClusterInfo(clusterHint); err == nil && clusterInfo != nil {
 			return clusterInfo.ClusterName, nil
 		}
-		// Exact cluster alias match:
-		if clustersInfo, err := ReadClustersInfo(""); err == nil {
-			for _, clusterInfo := range clustersInfo {
-				if clusterInfo.ClusterAlias == clusterHint {
-					return clusterInfo.ClusterName, nil
-				}
-			}
-		}
 	}
 
 	clusterByInstanceKey := func(instanceKey *InstanceKey) (hasResult bool, clusterName string, err error) {
diff --git a/go/vt/orchestrator/inst/instance_topology.go b/go/vt/orchestrator/inst/instance_topology.go
index 417cd47896..ed3bf7daf6 100644
--- a/go/vt/orchestrator/inst/instance_topology.go
+++ b/go/vt/orchestrator/inst/instance_topology.go
@@ -1087,54 +1087,6 @@ Cleanup:
 	return instance, err
 }
 
-// ReattachReplicaPrimaryHost reattaches a replica back onto its primary by undoing a DetachReplicaPrimaryHost operation
-func ReattachReplicaPrimaryHost(instanceKey *InstanceKey) (*Instance, error) {
-	instance, err := ReadTopologyInstance(instanceKey)
-	if err != nil {
-		return instance, err
-	}
-	if !instance.IsReplica() {
-		return instance, fmt.Errorf("instance is not a replica: %+v", *instanceKey)
-	}
-	if !instance.SourceKey.IsDetached() {
-		return instance, fmt.Errorf("instance does not seem to be detached: %+v", *instanceKey)
-	}
-
-	reattachedPrimaryKey := instance.SourceKey.ReattachedKey()
-
-	log.Infof("Will reattach primary host on %+v. Reattached key is %+v", *instanceKey, *reattachedPrimaryKey)
-
-	if maintenanceToken, merr := BeginMaintenance(instanceKey, GetMaintenanceOwner(), "reattach-replica-primary-host"); merr != nil {
-		err = fmt.Errorf("Cannot begin maintenance on %+v: %v", *instanceKey, merr)
-		goto Cleanup
-	} else {
-		defer EndMaintenance(maintenanceToken)
-	}
-
-	instance, err = StopReplication(instanceKey)
-	if err != nil {
-		goto Cleanup
-	}
-
-	_, err = ChangePrimaryTo(instanceKey, reattachedPrimaryKey, &instance.ExecBinlogCoordinates, true, GTIDHintNeutral)
-	if err != nil {
-		goto Cleanup
-	}
-	// Just in case this instance used to be a primary:
-	_ = ReplaceAliasClusterName(instanceKey.StringCode(), reattachedPrimaryKey.StringCode())
-
-Cleanup:
-	instance, _ = StartReplication(instanceKey)
-	if err != nil {
-		log.Error(err)
-		return instance, err
-	}
-	// and we're done (pending deferred functions)
-	AuditOperation("repoint", instanceKey, fmt.Sprintf("replica %+v reattached to primary %+v", *instanceKey, *reattachedPrimaryKey))
-
-	return instance, err
-}
-
 // EnableGTID will attempt to enable GTID-mode (either Oracle or MariaDB)
 func EnableGTID(instanceKey *InstanceKey) (*Instance, error) {
 	instance, err := ReadTopologyInstance(instanceKey)
diff --git a/go/vt/orchestrator/inst/pool.go b/go/vt/orchestrator/inst/pool.go
index 834b8c042a..648cde192e 100644
--- a/go/vt/orchestrator/inst/pool.go
+++ b/go/vt/orchestrator/inst/pool.go
@@ -45,11 +45,10 @@ func NewPoolInstancesSubmission(pool string, instances string) *PoolInstancesSub
 
 // ClusterPoolInstance is an instance mapping a cluster, pool & instance
 type ClusterPoolInstance struct {
-	ClusterName  string
-	ClusterAlias string
-	Pool         string
-	Hostname     string
-	Port         int
+	ClusterName string
+	Pool        string
+	Hostname    string
+	Port        int
 }
 
 func ApplyPoolInstances(submission *PoolInstancesSubmission) error {
diff --git a/go/vt/orchestrator/inst/pool_dao.go b/go/vt/orchestrator/inst/pool_dao.go
index dc9b4fb297..db82d93d8e 100644
--- a/go/vt/orchestrator/inst/pool_dao.go
+++ b/go/vt/orchestrator/inst/pool_dao.go
@@ -70,21 +70,18 @@ func ReadClusterPoolInstances(clusterName string, pool string) (result [](*Clust
 	query := fmt.Sprintf(`
 		select
 			cluster_name,
-			ifnull(alias, cluster_name) as alias,
 			database_instance_pool.*
 		from
 			database_instance
 			join database_instance_pool using (hostname, port)
-			left join cluster_alias using (cluster_name)
 		%s
 		`, whereClause)
 	err = db.QueryOrchestrator(query, args, func(m sqlutils.RowMap) error {
 		clusterPoolInstance := ClusterPoolInstance{
-			ClusterName:  m.GetString("cluster_name"),
-			ClusterAlias: m.GetString("alias"),
-			Pool:         m.GetString("pool"),
-			Hostname:     m.GetString("hostname"),
-			Port:         m.GetInt("port"),
+			ClusterName: m.GetString("cluster_name"),
+			Pool:        m.GetString("pool"),
+			Hostname:    m.GetString("hostname"),
+			Port:        m.GetInt("port"),
 		}
 		result = append(result, &clusterPoolInstance)
 		return nil
diff --git a/go/vt/orchestrator/logic/command_applier.go b/go/vt/orchestrator/logic/command_applier.go
index 650a20a230..d0e6ffe474 100644
--- a/go/vt/orchestrator/logic/command_applier.go
+++ b/go/vt/orchestrator/logic/command_applier.go
@@ -77,7 +77,7 @@ func (applier *CommandApplier) ApplyCommand(op string, value []byte) any {
 	case "delete-instance-tag":
 		return applier.deleteInstanceTag(value)
 	case "set-cluster-alias-manual-override":
-		return applier.setClusterAliasManualOverride(value)
+		return applier.noop(value)
 	}
 	errMsg := fmt.Sprintf("Unknown command op: %s", op)
 	log.Errorf(errMsg)
@@ -268,13 +268,6 @@ func (applier *CommandApplier) deleteInstanceTag(value []byte) any {
 	return err
 }
 
-func (applier *CommandApplier) setClusterAliasManualOverride(value []byte) any {
-	var params [2]string
-	if err := json.Unmarshal(value, &params); err != nil {
-		log.Error(err)
-		return err
-	}
-	clusterName, alias := params[0], params[1]
-	err := inst.SetClusterAliasManualOverride(clusterName, alias)
-	return err
+func (applier *CommandApplier) noop(value []byte) any {
+	return nil
 }
diff --git a/go/vt/orchestrator/logic/orchestrator.go b/go/vt/orchestrator/logic/orchestrator.go
index 78c1f9b5ba..daa3c46da0 100644
--- a/go/vt/orchestrator/logic/orchestrator.go
+++ b/go/vt/orchestrator/logic/orchestrator.go
@@ -386,7 +386,6 @@ func ContinuousDiscovery() {
 				// But rather should invoke such routinely operations that need to be as (or roughly as) frequent
 				// as instance poll
 				if IsLeaderOrActive() {
-					go inst.UpdateClusterAliases()
 					go inst.ExpireDowntime()
 					go injectSeeds(&seedOnce)
 				}
@@ -395,7 +394,6 @@ func ContinuousDiscovery() {
 			// Various periodic internal maintenance tasks
 			go func() {
 				if IsLeaderOrActive() {
-					go inst.ReviewUnseenInstances()
 					go inst.InjectUnseenPrimaries()
 
 					go inst.ForgetLongUnseenInstances()
diff --git a/go/vt/orchestrator/logic/topology_recovery.go b/go/vt/orchestrator/logic/topology_recovery.go
index 21ddae2302..d243d7df73 100644
--- a/go/vt/orchestrator/logic/topology_recovery.go
+++ b/go/vt/orchestrator/logic/topology_recovery.go
@@ -273,7 +273,6 @@ func prepareCommand(command string, topologyRecovery *TopologyRecovery) (result
 	command = strings.Replace(command, "{failedHost}", analysisEntry.AnalyzedInstanceKey.Hostname, -1)
 	command = strings.Replace(command, "{failedPort}", fmt.Sprintf("%d", analysisEntry.AnalyzedInstanceKey.Port), -1)
 	command = strings.Replace(command, "{failureCluster}", analysisEntry.ClusterDetails.ClusterName, -1)
-	command = strings.Replace(command, "{failureClusterAlias}", analysisEntry.ClusterDetails.ClusterAlias, -1)
 	command = strings.Replace(command, "{failureClusterDomain}", analysisEntry.ClusterDetails.ClusterDomain, -1)
 	command = strings.Replace(command, "{countReplicas}", fmt.Sprintf("%d", analysisEntry.CountReplicas), -1)
 	command = strings.Replace(command, "{isDowntimed}", fmt.Sprint(analysisEntry.IsDowntimed), -1)
@@ -311,7 +310,6 @@ func applyEnvironmentVariables(topologyRecovery *TopologyRecovery) []string {
 	env = append(env, fmt.Sprintf("ORC_FAILED_HOST=%s", analysisEntry.AnalyzedInstanceKey.Hostname))
 	env = append(env, fmt.Sprintf("ORC_FAILED_PORT=%d", analysisEntry.AnalyzedInstanceKey.Port))
 	env = append(env, fmt.Sprintf("ORC_FAILURE_CLUSTER=%s", analysisEntry.ClusterDetails.ClusterName))
-	env = append(env, fmt.Sprintf("ORC_FAILURE_CLUSTER_ALIAS=%s", analysisEntry.ClusterDetails.ClusterAlias))
 	env = append(env, fmt.Sprintf("ORC_FAILURE_CLUSTER_DOMAIN=%s", analysisEntry.ClusterDetails.ClusterDomain))
 	env = append(env, fmt.Sprintf("ORC_COUNT_REPLICAS=%d", analysisEntry.CountReplicas))
 	env = append(env, fmt.Sprintf("ORC_IS_DOWNTIMED=%v", analysisEntry.IsDowntimed))
@@ -665,18 +663,6 @@ func postErsCompletion(topologyRecovery *TopologyRecovery, analysisEntry inst.Re
 			}
 			topologyRecovery.AddPostponedFunction(postponedFunction, fmt.Sprintf("RecoverDeadPrimary, detaching promoted primary host %+v", promotedReplica.Key))
 		}
-		func() error {
-			before := analysisEntry.AnalyzedInstanceKey.StringCode()
-			after := promotedReplica.Key.StringCode()
-			AuditTopologyRecovery(topologyRecovery, fmt.Sprintf("- RecoverDeadPrimary: updating cluster_alias: %v -> %v", before, after))
-			//~~~inst.ReplaceClusterName(before, after)
-			if alias := analysisEntry.ClusterDetails.ClusterAlias; alias != "" {
-				inst.SetClusterAlias(promotedReplica.Key.StringCode(), alias)
-			} else {
-				inst.ReplaceAliasClusterName(before, after)
-			}
-			return nil
-		}()
 
 		attributes.SetGeneralAttribute(analysisEntry.ClusterDetails.ClusterDomain, promotedReplica.Key.StringCode())
 
@@ -1304,20 +1290,6 @@ func postPrsCompletion(topologyRecovery *TopologyRecovery, analysisEntry inst.Re
 	if promotedReplica != nil {
 		// Success!
 		AuditTopologyRecovery(topologyRecovery, fmt.Sprintf("%+v: successfully promoted %+v", analysisEntry.Analysis, promotedReplica.Key))
-
-		func() error {
-			before := analysisEntry.AnalyzedInstanceKey.StringCode()
-			after := promotedReplica.Key.StringCode()
-			AuditTopologyRecovery(topologyRecovery, fmt.Sprintf("- %+v: updating cluster_alias: %v -> %v", analysisEntry.Analysis, before, after))
-			//~~~inst.ReplaceClusterName(before, after)
-			if alias := analysisEntry.ClusterDetails.ClusterAlias; alias != "" {
-				inst.SetClusterAlias(promotedReplica.Key.StringCode(), alias)
-			} else {
-				inst.ReplaceAliasClusterName(before, after)
-			}
-			return nil
-		}()
-
 		attributes.SetGeneralAttribute(analysisEntry.ClusterDetails.ClusterDomain, promotedReplica.Key.StringCode())
 	}
 }
@@ -1329,7 +1301,7 @@ func electNewPrimary(ctx context.Context, analysisEntry inst.ReplicationAnalysis
 		AuditTopologyRecovery(topologyRecovery, fmt.Sprintf("found an active or recent recovery on %+v. Will not issue another electNewPrimary.", analysisEntry.AnalyzedInstanceKey))
 		return false, nil, err
 	}
-	log.Infof("Analysis: %v, will elect a new primary: %v", analysisEntry.Analysis, analysisEntry.SuggestedClusterAlias)
+	log.Infof("Analysis: %v, will elect a new primary: %v", analysisEntry.Analysis, analysisEntry.ClusterDetails.ClusterName)
 
 	var promotedReplica *inst.Instance
 	// This has to be done in the end; whether successful or not, we should mark that the recovery is done.
diff --git a/go/vt/orchestrator/logic/topology_recovery_dao.go b/go/vt/orchestrator/logic/topology_recovery_dao.go
index 05a0ca5879..48789f42e2 100644
--- a/go/vt/orchestrator/logic/topology_recovery_dao.go
+++ b/go/vt/orchestrator/logic/topology_recovery_dao.go
@@ -39,7 +39,6 @@ func AttemptFailureDetectionRegistration(analysisEntry *inst.ReplicationAnalysis
 		util.ProcessToken.Hash,
 		string(analysisEntry.Analysis),
 		analysisEntry.ClusterDetails.ClusterName,
-		analysisEntry.ClusterDetails.ClusterAlias,
 		analysisEntry.CountReplicas,
 		analysisEntry.Replicas.ToCommaDelimitedList(),
 		analysisEntry.IsActionableRecovery,
@@ -61,7 +60,6 @@ func AttemptFailureDetectionRegistration(analysisEntry *inst.ReplicationAnalysis
 					processcing_node_token,
 					analysis,
 					cluster_name,
-					cluster_alias,
 					count_affected_replicas,
 					replica_hosts,
 					is_actionable,
@@ -78,7 +76,6 @@ func AttemptFailureDetectionRegistration(analysisEntry *inst.ReplicationAnalysis
 					?,
 					?,
 					?,
-					?,
 					%s
 				)
 			`, startActivePeriodHint)
@@ -149,7 +146,6 @@ func writeTopologyRecovery(topologyRecovery *TopologyRecovery) (*TopologyRecover
 					processcing_node_token,
 					analysis,
 					cluster_name,
-					cluster_alias,
 					count_affected_replicas,
 					replica_hosts,
 					last_detection_id
@@ -167,7 +163,6 @@ func writeTopologyRecovery(topologyRecovery *TopologyRecovery) (*TopologyRecover
 					?,
 					?,
 					?,
-					?,
 					(select ifnull(max(detection_id), 0) from topology_failure_detection where hostname=? and port=?)
 				)
 			`,
@@ -177,7 +172,6 @@ func writeTopologyRecovery(topologyRecovery *TopologyRecovery) (*TopologyRecover
 		process.ThisHostname, util.ProcessToken.Hash,
 		string(analysisEntry.Analysis),
 		analysisEntry.ClusterDetails.ClusterName,
-		analysisEntry.ClusterDetails.ClusterAlias,
 		analysisEntry.CountReplicas, analysisEntry.Replicas.ToCommaDelimitedList(),
 		analysisEntry.AnalyzedInstanceKey.Hostname, analysisEntry.AnalyzedInstanceKey.Port,
 	)
@@ -433,18 +427,6 @@ func AcknowledgeClusterRecoveries(clusterName string, owner string, comment stri
 		}
 		countAcknowledgedEntries = countAcknowledgedEntries + count
 	}
-	{
-		clusterInfo, _ := inst.ReadClusterInfo(clusterName)
-		whereClause := `cluster_alias = ? and cluster_alias != ''`
-		args := sqlutils.Args(clusterInfo.ClusterAlias)
-		_ = clearAcknowledgedFailureDetections(whereClause, args)
-		count, err := acknowledgeRecoveries(owner, comment, false, whereClause, args)
-		if err != nil {
-			return count, err
-		}
-		countAcknowledgedEntries = countAcknowledgedEntries + count
-
-	}
 	return countAcknowledgedEntries, nil
 }
 
@@ -535,7 +517,6 @@ func readRecoveries(whereCondition string, limit string, args []any) ([]*Topolog
       ifnull(successor_alias, '') as successor_alias,
       analysis,
       cluster_name,
-      cluster_alias,
       count_affected_replicas,
       replica_hosts,
       participating_instances,
@@ -569,7 +550,6 @@ func readRecoveries(whereCondition string, limit string, args []any) ([]*Topolog
 		topologyRecovery.AnalysisEntry.AnalyzedInstanceKey.Port = m.GetInt("port")
 		topologyRecovery.AnalysisEntry.Analysis = inst.AnalysisCode(m.GetString("analysis"))
 		topologyRecovery.AnalysisEntry.ClusterDetails.ClusterName = m.GetString("cluster_name")
-		topologyRecovery.AnalysisEntry.ClusterDetails.ClusterAlias = m.GetString("cluster_alias")
 		topologyRecovery.AnalysisEntry.CountReplicas = m.GetUint("count_affected_replicas")
 		_ = topologyRecovery.AnalysisEntry.ReadReplicaHostsFromString(m.GetString("replica_hosts"))
 
@@ -664,7 +644,7 @@ func ReadRecoveryByUID(recoveryUID string) ([]*TopologyRecovery, error) {
 }
 
 // ReadRecentRecoveries reads latest recovery entries from topology_recovery
-func ReadRecentRecoveries(clusterName string, clusterAlias string, unacknowledgedOnly bool, page int) ([]*TopologyRecovery, error) {
+func ReadRecentRecoveries(clusterName string, unacknowledgedOnly bool, page int) ([]*TopologyRecovery, error) {
 	whereConditions := []string{}
 	whereClause := ""
 	args := sqlutils.Args()
@@ -674,9 +654,6 @@ func ReadRecentRecoveries(clusterName string, clusterAlias string, unacknowledge
 	if clusterName != "" {
 		whereConditions = append(whereConditions, `cluster_name=?`)
 		args = append(args, clusterName)
-	} else if clusterAlias != "" {
-		whereConditions = append(whereConditions, `cluster_alias=?`)
-		args = append(args, clusterAlias)
 	}
 	if len(whereConditions) > 0 {
 		whereClause = fmt.Sprintf("where %s", strings.Join(whereConditions, " and "))
@@ -703,7 +680,6 @@ func readFailureDetections(whereCondition string, limit string, args []any) ([]*
       processcing_node_token,
       analysis,
       cluster_name,
-      cluster_alias,
       count_affected_replicas,
       replica_hosts,
       (select max(recovery_id) from topology_recovery where topology_recovery.last_detection_id = detection_id) as related_recovery_id
@@ -727,7 +703,6 @@ func readFailureDetections(whereCondition string, limit string, args []any) ([]*
 		failureDetection.AnalysisEntry.AnalyzedInstanceKey.Port = m.GetInt("port")
 		failureDetection.AnalysisEntry.Analysis = inst.AnalysisCode(m.GetString("analysis"))
 		failureDetection.AnalysisEntry.ClusterDetails.ClusterName = m.GetString("cluster_name")
-		failureDetection.AnalysisEntry.ClusterDetails.ClusterAlias = m.GetString("cluster_alias")
 		failureDetection.AnalysisEntry.CountReplicas = m.GetUint("count_affected_replicas")
 		_ = failureDetection.AnalysisEntry.ReadReplicaHostsFromString(m.GetString("replica_hosts"))
 		failureDetection.AnalysisEntry.StartActivePeriod = m.GetString("start_active_period")
@@ -747,12 +722,12 @@ func readFailureDetections(whereCondition string, limit string, args []any) ([]*
 }
 
 // ReadRecentFailureDetections
-func ReadRecentFailureDetections(clusterAlias string, page int) ([]*TopologyRecovery, error) {
+func ReadRecentFailureDetections(clusterName string, page int) ([]*TopologyRecovery, error) {
 	whereClause := ""
 	args := sqlutils.Args()
-	if clusterAlias != "" {
-		whereClause = `where cluster_alias = ?`
-		args = append(args, clusterAlias)
+	if clusterName != "" {
+		whereClause = `where cluster_name = ?`
+		args = append(args, clusterName)
 	}
 	limit := `
 		limit ?
diff --git a/go/vt/servenv/buildinfo.go b/go/vt/servenv/buildinfo.go
index f1c99c59af..c03c85009e 100644
--- a/go/vt/servenv/buildinfo.go
+++ b/go/vt/servenv/buildinfo.go
@@ -91,8 +91,8 @@ func (v *versionInfo) String() string {
 }
 
 func (v *versionInfo) MySQLVersion() string {
-	if *MySQLServerVersion != "" {
-		return *MySQLServerVersion
+	if mySQLServerVersion != "" {
+		return mySQLServerVersion
 	}
 	return "5.7.9-vitess-" + v.version
 }
diff --git a/go/vt/servenv/mysql.go b/go/vt/servenv/mysql.go
index eb2d79df53..e74d4d07ac 100644
--- a/go/vt/servenv/mysql.go
+++ b/go/vt/servenv/mysql.go
@@ -16,9 +16,49 @@ limitations under the License.
 
 package servenv
 
-import "flag"
+import (
+	"github.com/spf13/pflag"
+)
 
-// MySQLServerVersion is what Vitess will present as it's version during the connection handshake,
+// mySQLServerVersion is what Vitess will present as it's version during the connection handshake,
 // and as the value to the @@version system variable. If nothing is provided, Vitess will report itself as
 // a specific MySQL version with the vitess version appended to it
-var MySQLServerVersion = flag.String("mysql_server_version", "", "MySQL server version to advertise.")
+var mySQLServerVersion string
+
+// RegisterMySQLServerFlags installs the flags needed to specify or expose a
+// particular MySQL server version from Vitess.
+func RegisterMySQLServerFlags(fs *pflag.FlagSet) {
+	fs.StringVar(&mySQLServerVersion, "mysql_server_version", mySQLServerVersion, "MySQL server version to advertise.")
+}
+
+// MySQLServerVersion returns the value of the `--mysql_server_version` flag.
+func MySQLServerVersion() string {
+	return mySQLServerVersion
+}
+
+// SetMySQLServerVersionForTest sets the value of the `--mysql_server_version`
+// flag. It is intended for use in tests that require a specific MySQL server
+// version (for example, collations) that cannot specify that via the command
+// line.
+func SetMySQLServerVersionForTest(version string) {
+	mySQLServerVersion = version
+}
+
+func init() {
+	for _, cmd := range []string{
+		"mysqlctl",
+		"mysqlctld",
+		"vtbackup",
+		"vtcombo",
+		"vtctl",
+		"vtctld",
+		"vtctldclient",
+		"vtexplain",
+		"vtgate",
+		"vtgateclienttest",
+		"vttablet",
+		"vttestserver",
+	} {
+		OnParseFor(cmd, RegisterMySQLServerFlags)
+	}
+}
diff --git a/go/vt/sqlparser/parser.go b/go/vt/sqlparser/parser.go
index e96fe527ee..a1033ef7c5 100644
--- a/go/vt/sqlparser/parser.go
+++ b/go/vt/sqlparser/parser.go
@@ -17,13 +17,13 @@ limitations under the License.
 package sqlparser
 
 import (
-	"flag"
 	"fmt"
 	"io"
 	"strconv"
 	"strings"
 	"sync"
 
+	"vitess.io/vitess/go/internal/flag"
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/vterrors"
@@ -108,8 +108,8 @@ func Parse2(sql string) (Statement, BindVars, error) {
 func checkParserVersionFlag() {
 	if flag.Parsed() {
 		versionFlagSync.Do(func() {
-			if *servenv.MySQLServerVersion != "" {
-				convVersion, err := convertMySQLVersionToCommentVersion(*servenv.MySQLServerVersion)
+			if mySQLVersion := servenv.MySQLServerVersion(); mySQLVersion != "" {
+				convVersion, err := convertMySQLVersionToCommentVersion(mySQLVersion)
 				if err != nil {
 					log.Error(err)
 				} else {
diff --git a/go/vt/vtgate/planbuilder/physical/route.go b/go/vt/vtgate/planbuilder/physical/route.go
index 871324b042..ee39c9217e 100644
--- a/go/vt/vtgate/planbuilder/physical/route.go
+++ b/go/vt/vtgate/planbuilder/physical/route.go
@@ -498,6 +498,12 @@ func (r *Route) planInOp(ctx *plancontext.PlanningContext, cmp *sqlparser.Compar
 	switch left := cmp.Left.(type) {
 	case *sqlparser.ColName:
 		vdValue := cmp.Right
+
+		valTuple, isTuple := vdValue.(sqlparser.ValTuple)
+		if isTuple && len(valTuple) == 1 {
+			return r.planEqualOp(ctx, &sqlparser.ComparisonExpr{Left: left, Right: valTuple[0], Operator: sqlparser.EqualOp})
+		}
+
 		value := r.makeEvalEngineExpr(ctx, vdValue)
 		if value == nil {
 			return false
@@ -600,6 +606,8 @@ func (r *Route) planCompositeInOpRecursive(
 	return foundVindex
 }
 
+// Reset all vindex predicates on this route and re-build their options from
+// the list of seen routing predicates.
 func (r *Route) resetRoutingSelections(ctx *plancontext.PlanningContext) error {
 	switch r.RouteOpCode {
 	case engine.DBA, engine.Next, engine.Reference, engine.Unsharded:
diff --git a/go/vt/vtgate/planbuilder/physical/route_planning.go b/go/vt/vtgate/planbuilder/physical/route_planning.go
index aaa7847c95..5b52a89a00 100644
--- a/go/vt/vtgate/planbuilder/physical/route_planning.go
+++ b/go/vt/vtgate/planbuilder/physical/route_planning.go
@@ -562,6 +562,7 @@ func createRouteOperatorForJoin(aRoute, bRoute *Route, joinPredicates []sqlparse
 		Keyspace:            aRoute.Keyspace,
 		VindexPreds:         append(aRoute.VindexPreds, bRoute.VindexPreds...),
 		SysTableTableSchema: append(aRoute.SysTableTableSchema, bRoute.SysTableTableSchema...),
+		SeenPredicates:      append(aRoute.SeenPredicates, bRoute.SeenPredicates...),
 		SysTableTableName:   sysTableName,
 		Source: &ApplyJoin{
 			LHS:       aRoute.Source,
diff --git a/go/vt/vtgate/planbuilder/physical/subquery_planning.go b/go/vt/vtgate/planbuilder/physical/subquery_planning.go
index 4177274cc7..88d38bc4ba 100644
--- a/go/vt/vtgate/planbuilder/physical/subquery_planning.go
+++ b/go/vt/vtgate/planbuilder/physical/subquery_planning.go
@@ -107,6 +107,22 @@ func mergeSubQueryOp(ctx *plancontext.PlanningContext, outer *Route, inner *Rout
 		outer.SysTableTableName[k] = v
 	}
 
+	// When merging an inner query with its outer query, we can remove the
+	// inner query from the list of predicates that can influence routing of
+	// the outer query.
+	//
+	// Note that not all inner queries necessarily are part of the routing
+	// predicates list, so this might be a no-op.
+	for i, predicate := range outer.SeenPredicates {
+		if sqlparser.EqualsExpr(predicate, subq.ExtractedSubquery) {
+			outer.SeenPredicates = append(outer.SeenPredicates[:i], outer.SeenPredicates[i+1:]...)
+
+			// The `ExtractedSubquery` of an inner query is unique (due to the uniqueness of bind variable names)
+			// so we can stop after the first match.
+			break
+		}
+	}
+
 	err = outer.resetRoutingSelections(ctx)
 	if err != nil {
 		return nil, err
diff --git a/go/vt/vtgate/plugin_mysql_server.go b/go/vt/vtgate/plugin_mysql_server.go
index 1e6545aef8..bbb15ac4db 100644
--- a/go/vt/vtgate/plugin_mysql_server.go
+++ b/go/vt/vtgate/plugin_mysql_server.go
@@ -437,8 +437,8 @@ func initMySQLProtocol() {
 		if err != nil {
 			log.Exitf("mysql.NewListener failed: %v", err)
 		}
-		if *servenv.MySQLServerVersion != "" {
-			mysqlListener.ServerVersion = *servenv.MySQLServerVersion
+		if mySQLVersion := servenv.MySQLServerVersion(); mySQLVersion != "" {
+			mysqlListener.ServerVersion = mySQLVersion
 		}
 		if *mysqlSslCert != "" && *mysqlSslKey != "" {
 			tlsVersion, err := vttls.TLSVersionToNumber(*mysqlTLSMinVersion)
diff --git a/go/vt/vttablet/tabletmanager/replmanager.go b/go/vt/vttablet/tabletmanager/replmanager.go
index 3f949494be..8a131162f0 100644
--- a/go/vt/vttablet/tabletmanager/replmanager.go
+++ b/go/vt/vttablet/tabletmanager/replmanager.go
@@ -68,7 +68,7 @@ func newReplManager(ctx context.Context, tm *TabletManager, interval time.Durati
 // SetTabletType starts/stops the replication manager ticks based on the tablet type provided.
 // It stops the ticks if the tablet type is not a replica type, starts the ticks otherwise.
 func (rm *replManager) SetTabletType(tabletType topodatapb.TabletType) {
-	if *mysqlctl.DisableActiveReparents {
+	if *mysqlctl.DisableActiveReparents || disableReplicationManager {
 		return
 	}
 	if !topo.IsReplicaType(tabletType) {
@@ -109,11 +109,24 @@ func (rm *replManager) checkActionLocked() {
 			return
 		}
 	} else {
-		// If only one of the threads is stopped, it's probably
-		// intentional. So, we don't repair replication.
-		if status.SQLHealthy() || status.IOHealthy() {
+		// Cases to consider for replication repair -
+		// 1. If both the threads are healthy, then there is nothing to repair.
+		// 2. If the IO thread is healthy but SQL thread isn't, then we assume it is intentional, and we don't repair replication.
+		// 3. If SQL thread is healthy but IO thread is in Connecting state with an IO Error, then we know that no Vitess operation stopped the IO thread
+		//	  and it is having trouble connecting to the primary Maybe the primary host-port changed, or something else happened.
+		//	  This could be an ephemeral error, so we should try and fix it.
+		// 4. If SQL thread is healthy but IO thread is stopped, then we assume it is intentional, and we don't repair replication.
+		// 5. Both the threads are unhealthy, then we need to repair replication
+		if status.IOHealthy() {
+			// This covers cases 1 and 2.
 			return
 		}
+		if status.SQLHealthy() && (status.IOState != mysql.ReplicationStateConnecting || status.LastIOError == "") {
+			// This covers case 4.
+			return
+		}
+		// We now attempt to repair replication.
+		// This covers cases 3 and 5.
 	}
 
 	if !rm.failed {
@@ -135,7 +148,7 @@ func (rm *replManager) checkActionLocked() {
 // reset the replication manager state and deleting the marker-file.
 // it does not start or stop the ticks. Use setReplicationStopped instead to change that.
 func (rm *replManager) reset() {
-	if *mysqlctl.DisableActiveReparents {
+	if *mysqlctl.DisableActiveReparents || disableReplicationManager {
 		return
 	}
 
@@ -152,7 +165,7 @@ func (rm *replManager) reset() {
 // setReplicationStopped performs a best effort attempt of
 // remembering a decision to stop replication.
 func (rm *replManager) setReplicationStopped(stopped bool) {
-	if *mysqlctl.DisableActiveReparents {
+	if *mysqlctl.DisableActiveReparents || disableReplicationManager {
 		return
 	}
 
diff --git a/go/vt/vttablet/tabletmanager/rpc_replication.go b/go/vt/vttablet/tabletmanager/rpc_replication.go
index a33d22dfd2..49dd010f28 100644
--- a/go/vt/vttablet/tabletmanager/rpc_replication.go
+++ b/go/vt/vttablet/tabletmanager/rpc_replication.go
@@ -39,12 +39,14 @@ import (
 )
 
 var setSuperReadOnly bool
+var disableReplicationManager bool
 
 func registerReplicationFlags(fs *pflag.FlagSet) {
 	fs.Bool("enable_semi_sync", false, "")
 	fs.MarkDeprecated("enable_semi_sync", "--enable_semi_sync is deprecated; please set the correct durability policy on the keyspace instead.")
 
 	fs.BoolVar(&setSuperReadOnly, "use_super_read_only", setSuperReadOnly, "Set super_read_only flag when performing planned failover.")
+	fs.BoolVar(&disableReplicationManager, "disable-replication-manager", disableReplicationManager, "Disable replication manager to prevent replication repairs.")
 }
 
 func init() {
diff --git a/go/vt/vttablet/tabletmanager/vdiff/action.go b/go/vt/vttablet/tabletmanager/vdiff/action.go
index bf21ff42cf..2966f2be0b 100644
--- a/go/vt/vttablet/tabletmanager/vdiff/action.go
+++ b/go/vt/vttablet/tabletmanager/vdiff/action.go
@@ -115,23 +115,19 @@ func (vde *Engine) fixupOptions(options *tabletmanagerdatapb.VDiffOptions) (*tab
 	// Assign defaults to sourceCell and targetCell if not specified.
 	sourceCell := options.PickerOptions.SourceCell
 	targetCell := options.PickerOptions.TargetCell
-	if sourceCell == "" && targetCell == "" {
-		cells, err := vde.ts.GetCellInfoNames(vde.ctx)
+	var defaultCell string
+	var err error
+	if sourceCell == "" || targetCell == "" {
+		defaultCell, err = vde.getDefaultCell()
 		if err != nil {
 			return nil, err
 		}
-		if len(cells) == 0 {
-			// Unreachable
-			return nil, fmt.Errorf("there are no cells in the topo")
-		}
-		sourceCell = cells[0]
-		targetCell = sourceCell
 	}
 	if sourceCell == "" {
-		sourceCell = targetCell
+		sourceCell = defaultCell
 	}
 	if targetCell == "" {
-		targetCell = sourceCell
+		targetCell = defaultCell
 	}
 	options.PickerOptions.SourceCell = sourceCell
 	options.PickerOptions.TargetCell = targetCell
@@ -139,6 +135,18 @@ func (vde *Engine) fixupOptions(options *tabletmanagerdatapb.VDiffOptions) (*tab
 	return options, nil
 }
 
+func (vde *Engine) getDefaultCell() (string, error) {
+	cells, err := vde.ts.GetCellInfoNames(vde.ctx)
+	if err != nil {
+		return "", err
+	}
+	if len(cells) == 0 {
+		// Unreachable
+		return "", fmt.Errorf("there are no cells in the topo")
+	}
+	return cells[0], nil
+}
+
 func (vde *Engine) handleCreateResumeAction(ctx context.Context, dbClient binlogplayer.DBClient, action VDiffAction, req *tabletmanagerdatapb.VDiffRequest, resp *tabletmanagerdatapb.VDiffResponse) error {
 	var qr *sqltypes.Result
 	var err error
diff --git a/go/vt/vttablet/tabletmanager/vdiff/controller.go b/go/vt/vttablet/tabletmanager/vdiff/controller.go
index 63bc7a369c..e3a362d3c6 100644
--- a/go/vt/vttablet/tabletmanager/vdiff/controller.go
+++ b/go/vt/vttablet/tabletmanager/vdiff/controller.go
@@ -72,6 +72,8 @@ type controller struct {
 	options             *tabletmanagerdata.VDiffOptions // options initially from vtctld command and later from _vt.vdiff
 
 	sourceTimeZone, targetTimeZone string // named time zones if conversions are necessary for datetime values
+
+	externalCluster string // for Mount+Migrate
 }
 
 func newController(ctx context.Context, row sqltypes.RowNamedValues, dbClientFactory func() binlogplayer.DBClient,
@@ -202,6 +204,10 @@ func (ct *controller) start(ctx context.Context, dbClient binlogplayer.DBClient)
 		ct.sourceTimeZone = bls.SourceTimeZone
 		ct.targetTimeZone = bls.TargetTimeZone
 
+		if bls.ExternalCluster != "" {
+			ct.externalCluster = bls.ExternalCluster
+		}
+
 		ct.sources[source.shard] = source
 		if i == 0 {
 			ct.sourceKeyspace = bls.Keyspace
diff --git a/go/vt/vttablet/tabletmanager/vdiff/table_differ.go b/go/vt/vttablet/tabletmanager/vdiff/table_differ.go
index 1e58d93508..18a4a37a4c 100644
--- a/go/vt/vttablet/tabletmanager/vdiff/table_differ.go
+++ b/go/vt/vttablet/tabletmanager/vdiff/table_differ.go
@@ -194,12 +194,22 @@ func (td *tableDiffer) selectTablets(ctx context.Context, cell, tabletTypes stri
 	var wg sync.WaitGroup
 	ct := td.wd.ct
 	var err1, err2 error
+
+	// For Mount+Migrate, the source tablets will be in a different
+	// Vitess cluster with its own TopoServer.
+	sourceTopoServer := ct.ts
+	if ct.externalCluster != "" {
+		extTS, err := ct.ts.OpenExternalVitessClusterServer(ctx, ct.externalCluster)
+		if err != nil {
+			return err
+		}
+		sourceTopoServer = extTS
+	}
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		err1 = td.forEachSource(func(source *migrationSource) error {
-			// TODO: handle external sources to support Mount+Migrate
-			tablet, err := pickTablet(ctx, ct.ts, cell, ct.sourceKeyspace, source.shard, tabletTypes)
+			tablet, err := pickTablet(ctx, sourceTopoServer, cell, ct.sourceKeyspace, source.shard, tabletTypes)
 			if err != nil {
 				return err
 			}
diff --git a/tools/code_freeze.sh b/tools/code_freeze.sh
new file mode 100755
index 0000000000..7537ffbb13
--- /dev/null
+++ b/tools/code_freeze.sh
@@ -0,0 +1,65 @@
+#!/bin/bash
+
+# Copyright 2022 The Vitess Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#############
+#
+# This script changes the code_freeze workflow to either fail or succeed.
+#
+# The first argument of the script is a string, which should be either: "freeze" or "unfreeze".
+#   - If the argument == "freeze" then the workflow will always fail.
+#   - If the argument == "unfreeze" then the workflow will always succeed.
+#
+# The second argument is the name of the branch you want to freeze or unfreeze.
+# The script takes care of creating a branch that is based on top of the branch
+# you want to freeze/unfreeze. A commit is then created on this new branch, allowing
+# you to push the commit and create a Pull Request that you merge against the input
+# branch.
+#
+#############
+
+freeze=$1
+branch=$2
+code_freeze_workflow="./.github/workflows/code_freeze.yml"
+
+if [ "$freeze" != "freeze" && "$freeze" != "unfreeze" ]; then
+    echo "the first argument must be either 'freeze' or 'unfreeze'"
+    exit 1
+fi
+
+new_branch_name=$branch-code-freeze-1
+git checkout -b $new_branch_name $branch
+for (( i = 2; $? != 0; i++ )); do
+  new_branch_name=$branch-code-freeze-$i
+  git checkout -b $new_branch_name $branch
+done
+
+if [ "$freeze" == "freeze" ]; then
+    sed -i.bak -E "s/exit (.*)/exit 1/g" $code_freeze_workflow
+elif [ "$freeze" == "unfreeze" ]; then
+    sed -i.bak -E "s/exit (.*)/exit 0/g" $code_freeze_workflow
+fi
+
+rm -f $code_freeze_workflow.bak
+
+git add --all
+git commit -n -s -m "Code $freeze of $branch"
+
+echo " ------------------------------ "
+echo " "
+echo "The branch is ready to be pushed. Push it using the following line and create a PR against ${branch}"
+echo "      git push upstream ${new_branch_name}"
+echo " "
+echo "(make sure to replace upstream with vitessio/vitess' remote)"


Test patch:
diff --git a/go/mysql/collations/integration/collations_test.go b/go/mysql/collations/integration/collations_test.go
index c13e279deb..d9cc6f11c3 100644
--- a/go/mysql/collations/integration/collations_test.go
+++ b/go/mysql/collations/integration/collations_test.go
@@ -44,7 +44,7 @@ var collationEnv *collations.Environment
 func init() {
 	// We require MySQL 8.0 collations for the comparisons in the tests
 	mySQLVersion := "8.0.0"
-	servenv.MySQLServerVersion = &mySQLVersion
+	servenv.SetMySQLServerVersionForTest(mySQLVersion)
 	collationEnv = collations.NewEnvironment(mySQLVersion)
 }
 
diff --git a/go/test/endtoend/tabletmanager/replication_manager/tablet_test.go b/go/test/endtoend/tabletmanager/replication_manager/tablet_test.go
new file mode 100644
index 0000000000..3284ab65d4
--- /dev/null
+++ b/go/test/endtoend/tabletmanager/replication_manager/tablet_test.go
@@ -0,0 +1,191 @@
+/*
+Copyright 2022 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tabletmanager
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/test/endtoend/cluster"
+	replicationdatapb "vitess.io/vitess/go/vt/proto/replicationdata"
+	tabletpb "vitess.io/vitess/go/vt/proto/topodata"
+	tmc "vitess.io/vitess/go/vt/vttablet/grpctmclient"
+)
+
+var (
+	clusterInstance *cluster.LocalProcessCluster
+	tmClient        *tmc.Client
+	primaryTablet   cluster.Vttablet
+	replicaTablet   cluster.Vttablet
+	hostname        = "localhost"
+	keyspaceName    = "ks"
+	cell            = "zone1"
+	sqlSchema       = `
+	create table t1(
+		id bigint,
+		value varchar(16),
+		primary key(id)
+	) Engine=InnoDB DEFAULT CHARSET=utf8;
+	CREATE VIEW v1 AS SELECT id, value FROM t1;
+`
+
+	vSchema = `
+	{
+    "sharded": false,
+    "vindexes": {
+      "hash": {
+        "type": "hash"
+      }
+    },
+    "tables": {
+      "t1": {
+        "column_vindexes": [
+          {
+            "column": "id",
+            "name": "hash"
+          }
+        ]
+      }
+    }
+  }`
+)
+
+func TestMain(m *testing.M) {
+	defer cluster.PanicHandler(nil)
+	flag.Parse()
+
+	exitCode := func() int {
+		clusterInstance = cluster.NewCluster(cell, hostname)
+		defer clusterInstance.Teardown()
+
+		// Start topo server
+		err := clusterInstance.StartTopo()
+		if err != nil {
+			return 1
+		}
+
+		// We do not need semiSync for this test case.
+		clusterInstance.EnableSemiSync = false
+
+		// Start keyspace
+		keyspace := &cluster.Keyspace{
+			Name:      keyspaceName,
+			SchemaSQL: sqlSchema,
+			VSchema:   vSchema,
+		}
+
+		if err = clusterInstance.StartUnshardedKeyspace(*keyspace, 1, false); err != nil {
+			return 1
+		}
+
+		// Collect table paths and ports
+		tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+		for _, tablet := range tablets {
+			if tablet.Type == "primary" {
+				primaryTablet = *tablet
+			} else {
+				replicaTablet = *tablet
+			}
+		}
+
+		// create tablet manager client
+		tmClient = tmc.NewClient()
+
+		return m.Run()
+	}()
+	os.Exit(exitCode)
+}
+
+func tmcGetReplicationStatus(ctx context.Context, tabletGrpcPort int) (*replicationdatapb.Status, error) {
+	vtablet := getTablet(tabletGrpcPort)
+	return tmClient.ReplicationStatus(ctx, vtablet)
+}
+
+func getTablet(tabletGrpcPort int) *tabletpb.Tablet {
+	portMap := make(map[string]int32)
+	portMap["grpc"] = int32(tabletGrpcPort)
+	return &tabletpb.Tablet{Hostname: hostname, PortMap: portMap}
+}
+
+// resurrectTablet is used to resurrect the given tablet
+func resurrectTablet(t *testing.T, tab cluster.Vttablet) {
+	// initialize config again to regenerate the my.cnf file which has the port to use
+	_, err := tab.MysqlctlProcess.ExecuteCommandWithOutput("--log_dir", tab.MysqlctlProcess.LogDirectory,
+		"--tablet_uid", fmt.Sprintf("%d", tab.MysqlctlProcess.TabletUID),
+		"--mysql_port", fmt.Sprintf("%d", tab.MysqlctlProcess.MySQLPort),
+		"init_config")
+	require.NoError(t, err)
+
+	tab.MysqlctlProcess.InitMysql = false
+	err = tab.MysqlctlProcess.Start()
+	require.NoError(t, err)
+
+	// Start the tablet
+	tab.VttabletProcess.ServingStatus = "SERVING"
+	err = tab.VttabletProcess.Setup()
+	require.NoError(t, err)
+}
+
+// stopTablet stops the tablet
+func stopTablet(t *testing.T, tab cluster.Vttablet) {
+	err := tab.VttabletProcess.TearDownWithTimeout(30 * time.Second)
+	require.NoError(t, err)
+	err = tab.MysqlctlProcess.Stop()
+	require.NoError(t, err)
+}
+
+func waitForSourcePort(ctx context.Context, t *testing.T, tablet cluster.Vttablet, expectedPort int32) error {
+	timeout := time.Now().Add(15 * time.Second)
+	for time.Now().Before(timeout) {
+		// Check that initially replication is setup correctly on the replica tablet
+		replicaStatus, err := tmcGetReplicationStatus(ctx, tablet.GrpcPort)
+		require.NoError(t, err)
+		if replicaStatus.SourcePort == expectedPort {
+			return nil
+		}
+		time.Sleep(300 * time.Millisecond)
+	}
+	return fmt.Errorf("time out before source port became %v for %v", expectedPort, tablet.Alias)
+}
+
+func TestReplicationRepairAfterPrimaryTabletChange(t *testing.T) {
+	ctx := context.Background()
+	// Check that initially replication is setup correctly on the replica tablet
+	err := waitForSourcePort(ctx, t, replicaTablet, int32(primaryTablet.MySQLPort))
+	require.NoError(t, err)
+
+	// Stop the primary tablet
+	stopTablet(t, primaryTablet)
+
+	// Change the MySQL port of the primary tablet
+	newMysqlPort := clusterInstance.GetAndReservePort()
+	primaryTablet.MySQLPort = newMysqlPort
+	primaryTablet.MysqlctlProcess.MySQLPort = newMysqlPort
+
+	// Start the primary tablet again
+	resurrectTablet(t, primaryTablet)
+
+	// Let replication manager repair replication
+	err = waitForSourcePort(ctx, t, replicaTablet, int32(newMysqlPort))
+	require.NoError(t, err)
+}
diff --git a/go/test/endtoend/vreplication/migrate_test.go b/go/test/endtoend/vreplication/migrate_test.go
index be7e8d4434..cfedfce2f6 100644
--- a/go/test/endtoend/vreplication/migrate_test.go
+++ b/go/test/endtoend/vreplication/migrate_test.go
@@ -48,15 +48,6 @@ func TestMigrate(t *testing.T) {
 	allCellNames = "zone1"
 	vc = NewVitessCluster(t, "TestMigrate", cells, mainClusterConfig)
 
-	// VDiff2 does not support this mount+migrate use case today
-	// TODO: add support for this in the tablet picker phase
-	if runVDiffsSideBySide {
-		runVDiffsSideBySide = false
-		defer func() {
-			runVDiffsSideBySide = true
-		}()
-	}
-
 	require.NotNil(t, vc)
 	defaultReplicas = 0
 	defaultRdonly = 0
diff --git a/go/test/endtoend/vtorc/readtopologyinstance/main_test.go b/go/test/endtoend/vtorc/readtopologyinstance/main_test.go
index 19bd982b9b..908f2e41d4 100644
--- a/go/test/endtoend/vtorc/readtopologyinstance/main_test.go
+++ b/go/test/endtoend/vtorc/readtopologyinstance/main_test.go
@@ -117,6 +117,7 @@ func TestReadTopologyInstanceBufferable(t *testing.T) {
 	assert.False(t, primaryInstance.HasReplicationCredentials)
 	assert.Equal(t, primaryInstance.ReplicationIOThreadState, inst.ReplicationThreadStateNoThread)
 	assert.Equal(t, primaryInstance.ReplicationSQLThreadState, inst.ReplicationThreadStateNoThread)
+	assert.Equal(t, fmt.Sprintf("%v:%v", keyspace.Name, shard0.Name), primaryInstance.ClusterName)
 
 	replicaInstance, err := inst.ReadTopologyInstanceBufferable(&inst.InstanceKey{
 		Hostname: utils.Hostname,
@@ -168,4 +169,5 @@ func TestReadTopologyInstanceBufferable(t *testing.T) {
 	assert.False(t, replicaInstance.HasReplicationFilters)
 	assert.LessOrEqual(t, int(replicaInstance.SecondsBehindPrimary.Int64), 1)
 	assert.False(t, replicaInstance.AllowTLS)
+	assert.Equal(t, fmt.Sprintf("%v:%v", keyspace.Name, shard0.Name), replicaInstance.ClusterName)
 }
diff --git a/go/vt/orchestrator/inst/instance_dao_test.go b/go/vt/orchestrator/inst/instance_dao_test.go
index c4f3032912..08334b567a 100644
--- a/go/vt/orchestrator/inst/instance_dao_test.go
+++ b/go/vt/orchestrator/inst/instance_dao_test.go
@@ -60,17 +60,17 @@ func TestMkInsertOdkuSingle(t *testing.T) {
 									version, major_version, version_comment, binlog_server, read_only, binlog_format,
 									binlog_row_image, log_bin, log_replica_updates, binary_log_file, binary_log_pos, source_host, source_port,
 									replica_sql_running, replica_io_running, replication_sql_thread_state, replication_io_thread_state, has_replication_filters, supports_oracle_gtid, oracle_gtid, source_uuid, ancestry_uuid, executed_gtid_set, gtid_mode, gtid_purged, gtid_errant, mariadb_gtid, pseudo_gtid,
-									source_log_file, read_source_log_pos, relay_source_log_file, exec_source_log_pos, relay_log_file, relay_log_pos, last_sql_error, last_io_error, replication_lag_seconds, replica_lag_seconds, sql_delay, num_replica_hosts, replica_hosts, cluster_name, suggested_cluster_alias, data_center, region, physical_environment, replication_depth, is_co_primary, has_replication_credentials, allow_tls, semi_sync_enforced, semi_sync_primary_enabled, semi_sync_primary_timeout, semi_sync_primary_wait_for_replica_count, semi_sync_replica_enabled, semi_sync_primary_status, semi_sync_primary_clients, semi_sync_replica_status, instance_alias, last_discovery_latency, replication_group_name, replication_group_is_single_primary_mode, replication_group_member_state, replication_group_member_role, replication_group_members, replication_group_primary_host, replication_group_primary_port, last_seen)
+									source_log_file, read_source_log_pos, relay_source_log_file, exec_source_log_pos, relay_log_file, relay_log_pos, last_sql_error, last_io_error, replication_lag_seconds, replica_lag_seconds, sql_delay, num_replica_hosts, replica_hosts, cluster_name, data_center, region, physical_environment, replication_depth, is_co_primary, has_replication_credentials, allow_tls, semi_sync_enforced, semi_sync_primary_enabled, semi_sync_primary_timeout, semi_sync_primary_wait_for_replica_count, semi_sync_replica_enabled, semi_sync_primary_status, semi_sync_primary_clients, semi_sync_replica_status, instance_alias, last_discovery_latency, replication_group_name, replication_group_is_single_primary_mode, replication_group_member_state, replication_group_member_role, replication_group_members, replication_group_primary_host, replication_group_primary_port, last_seen)
         VALUES
-                (?, ?, NOW(), NOW(), 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+                (?, ?, NOW(), NOW(), 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
         ON DUPLICATE KEY UPDATE
-                hostname=VALUES(hostname), port=VALUES(port), last_checked=VALUES(last_checked), last_attempted_check=VALUES(last_attempted_check), last_check_partial_success=VALUES(last_check_partial_success), server_id=VALUES(server_id), server_uuid=VALUES(server_uuid), version=VALUES(version), major_version=VALUES(major_version), version_comment=VALUES(version_comment), binlog_server=VALUES(binlog_server), read_only=VALUES(read_only), binlog_format=VALUES(binlog_format), binlog_row_image=VALUES(binlog_row_image), log_bin=VALUES(log_bin), log_replica_updates=VALUES(log_replica_updates), binary_log_file=VALUES(binary_log_file), binary_log_pos=VALUES(binary_log_pos), source_host=VALUES(source_host), source_port=VALUES(source_port), replica_sql_running=VALUES(replica_sql_running), replica_io_running=VALUES(replica_io_running), replication_sql_thread_state=VALUES(replication_sql_thread_state), replication_io_thread_state=VALUES(replication_io_thread_state), has_replication_filters=VALUES(has_replication_filters), supports_oracle_gtid=VALUES(supports_oracle_gtid), oracle_gtid=VALUES(oracle_gtid), source_uuid=VALUES(source_uuid), ancestry_uuid=VALUES(ancestry_uuid), executed_gtid_set=VALUES(executed_gtid_set), gtid_mode=VALUES(gtid_mode), gtid_purged=VALUES(gtid_purged), gtid_errant=VALUES(gtid_errant), mariadb_gtid=VALUES(mariadb_gtid), pseudo_gtid=VALUES(pseudo_gtid), source_log_file=VALUES(source_log_file), read_source_log_pos=VALUES(read_source_log_pos), relay_source_log_file=VALUES(relay_source_log_file), exec_source_log_pos=VALUES(exec_source_log_pos), relay_log_file=VALUES(relay_log_file), relay_log_pos=VALUES(relay_log_pos), last_sql_error=VALUES(last_sql_error), last_io_error=VALUES(last_io_error), replication_lag_seconds=VALUES(replication_lag_seconds), replica_lag_seconds=VALUES(replica_lag_seconds), sql_delay=VALUES(sql_delay), num_replica_hosts=VALUES(num_replica_hosts), replica_hosts=VALUES(replica_hosts), cluster_name=VALUES(cluster_name), suggested_cluster_alias=VALUES(suggested_cluster_alias), data_center=VALUES(data_center), region=VALUES(region), physical_environment=VALUES(physical_environment), replication_depth=VALUES(replication_depth), is_co_primary=VALUES(is_co_primary), has_replication_credentials=VALUES(has_replication_credentials), allow_tls=VALUES(allow_tls),
+                hostname=VALUES(hostname), port=VALUES(port), last_checked=VALUES(last_checked), last_attempted_check=VALUES(last_attempted_check), last_check_partial_success=VALUES(last_check_partial_success), server_id=VALUES(server_id), server_uuid=VALUES(server_uuid), version=VALUES(version), major_version=VALUES(major_version), version_comment=VALUES(version_comment), binlog_server=VALUES(binlog_server), read_only=VALUES(read_only), binlog_format=VALUES(binlog_format), binlog_row_image=VALUES(binlog_row_image), log_bin=VALUES(log_bin), log_replica_updates=VALUES(log_replica_updates), binary_log_file=VALUES(binary_log_file), binary_log_pos=VALUES(binary_log_pos), source_host=VALUES(source_host), source_port=VALUES(source_port), replica_sql_running=VALUES(replica_sql_running), replica_io_running=VALUES(replica_io_running), replication_sql_thread_state=VALUES(replication_sql_thread_state), replication_io_thread_state=VALUES(replication_io_thread_state), has_replication_filters=VALUES(has_replication_filters), supports_oracle_gtid=VALUES(supports_oracle_gtid), oracle_gtid=VALUES(oracle_gtid), source_uuid=VALUES(source_uuid), ancestry_uuid=VALUES(ancestry_uuid), executed_gtid_set=VALUES(executed_gtid_set), gtid_mode=VALUES(gtid_mode), gtid_purged=VALUES(gtid_purged), gtid_errant=VALUES(gtid_errant), mariadb_gtid=VALUES(mariadb_gtid), pseudo_gtid=VALUES(pseudo_gtid), source_log_file=VALUES(source_log_file), read_source_log_pos=VALUES(read_source_log_pos), relay_source_log_file=VALUES(relay_source_log_file), exec_source_log_pos=VALUES(exec_source_log_pos), relay_log_file=VALUES(relay_log_file), relay_log_pos=VALUES(relay_log_pos), last_sql_error=VALUES(last_sql_error), last_io_error=VALUES(last_io_error), replication_lag_seconds=VALUES(replication_lag_seconds), replica_lag_seconds=VALUES(replica_lag_seconds), sql_delay=VALUES(sql_delay), num_replica_hosts=VALUES(num_replica_hosts), replica_hosts=VALUES(replica_hosts), cluster_name=VALUES(cluster_name), data_center=VALUES(data_center), region=VALUES(region), physical_environment=VALUES(physical_environment), replication_depth=VALUES(replication_depth), is_co_primary=VALUES(is_co_primary), has_replication_credentials=VALUES(has_replication_credentials), allow_tls=VALUES(allow_tls),
 								semi_sync_enforced=VALUES(semi_sync_enforced), semi_sync_primary_enabled=VALUES(semi_sync_primary_enabled), semi_sync_primary_timeout=VALUES(semi_sync_primary_timeout), semi_sync_primary_wait_for_replica_count=VALUES(semi_sync_primary_wait_for_replica_count), semi_sync_replica_enabled=VALUES(semi_sync_replica_enabled), semi_sync_primary_status=VALUES(semi_sync_primary_status), semi_sync_primary_clients=VALUES(semi_sync_primary_clients), semi_sync_replica_status=VALUES(semi_sync_replica_status),
 								instance_alias=VALUES(instance_alias), last_discovery_latency=VALUES(last_discovery_latency), replication_group_name=VALUES(replication_group_name), replication_group_is_single_primary_mode=VALUES(replication_group_is_single_primary_mode), replication_group_member_state=VALUES(replication_group_member_state), replication_group_member_role=VALUES(replication_group_member_role), replication_group_members=VALUES(replication_group_members), replication_group_primary_host=VALUES(replication_group_primary_host), replication_group_primary_port=VALUES(replication_group_primary_port), last_seen=VALUES(last_seen)
         `
 	a1 := `i710, 3306, 710, , 5.6.7, 5.6, MySQL, false, false, STATEMENT,
 	FULL, false, false, , 0, , 0,
-	false, false, 0, 0, false, false, false, , , , , , , false, false, , 0, mysql.000007, 10, , 0, , , {0 false}, {0 false}, 0, 0, [], , , , , , 0, false, false, false, false, false, 0, 0, false, false, 0, false, , 0, , false, , , [], , 0, `
+	false, false, 0, 0, false, false, false, , , , , , , false, false, , 0, mysql.000007, 10, , 0, , , {0 false}, {0 false}, 0, 0, [], , , , , 0, false, false, false, false, false, 0, 0, false, false, 0, false, , 0, , false, , , [], , 0, `
 
 	sql1, args1, err := mkInsertOdkuForInstances(instances[:1], false, true)
 	test.S(t).ExpectNil(err)
@@ -83,22 +83,22 @@ func TestMkInsertOdkuThree(t *testing.T) {
 
 	// three instances
 	s3 := `INSERT  INTO database_instance
-                (hostname, port, last_checked, last_attempted_check, last_check_partial_success, server_id, server_uuid, version, major_version, version_comment, binlog_server, read_only, binlog_format, binlog_row_image, log_bin, log_replica_updates, binary_log_file, binary_log_pos, source_host, source_port, replica_sql_running, replica_io_running, replication_sql_thread_state, replication_io_thread_state, has_replication_filters, supports_oracle_gtid, oracle_gtid, source_uuid, ancestry_uuid, executed_gtid_set, gtid_mode, gtid_purged, gtid_errant, mariadb_gtid, pseudo_gtid, source_log_file, read_source_log_pos, relay_source_log_file, exec_source_log_pos, relay_log_file, relay_log_pos, last_sql_error, last_io_error, replication_lag_seconds, replica_lag_seconds, sql_delay, num_replica_hosts, replica_hosts, cluster_name, suggested_cluster_alias, data_center, region, physical_environment, replication_depth, is_co_primary, has_replication_credentials, allow_tls, semi_sync_enforced, semi_sync_primary_enabled, semi_sync_primary_timeout, semi_sync_primary_wait_for_replica_count,
+                (hostname, port, last_checked, last_attempted_check, last_check_partial_success, server_id, server_uuid, version, major_version, version_comment, binlog_server, read_only, binlog_format, binlog_row_image, log_bin, log_replica_updates, binary_log_file, binary_log_pos, source_host, source_port, replica_sql_running, replica_io_running, replication_sql_thread_state, replication_io_thread_state, has_replication_filters, supports_oracle_gtid, oracle_gtid, source_uuid, ancestry_uuid, executed_gtid_set, gtid_mode, gtid_purged, gtid_errant, mariadb_gtid, pseudo_gtid, source_log_file, read_source_log_pos, relay_source_log_file, exec_source_log_pos, relay_log_file, relay_log_pos, last_sql_error, last_io_error, replication_lag_seconds, replica_lag_seconds, sql_delay, num_replica_hosts, replica_hosts, cluster_name, data_center, region, physical_environment, replication_depth, is_co_primary, has_replication_credentials, allow_tls, semi_sync_enforced, semi_sync_primary_enabled, semi_sync_primary_timeout, semi_sync_primary_wait_for_replica_count,
 								semi_sync_replica_enabled, semi_sync_primary_status, semi_sync_primary_clients, semi_sync_replica_status, instance_alias, last_discovery_latency, replication_group_name, replication_group_is_single_primary_mode, replication_group_member_state, replication_group_member_role, replication_group_members, replication_group_primary_host, replication_group_primary_port, last_seen)
         VALUES
-								(?, ?, NOW(), NOW(), 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()),
-								(?, ?, NOW(), NOW(), 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()),
-								(?, ?, NOW(), NOW(), 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
+								(?, ?, NOW(), NOW(), 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()),
+								(?, ?, NOW(), NOW(), 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW()),
+								(?, ?, NOW(), NOW(), 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
         ON DUPLICATE KEY UPDATE
-                hostname=VALUES(hostname), port=VALUES(port), last_checked=VALUES(last_checked), last_attempted_check=VALUES(last_attempted_check), last_check_partial_success=VALUES(last_check_partial_success), server_id=VALUES(server_id), server_uuid=VALUES(server_uuid), version=VALUES(version), major_version=VALUES(major_version), version_comment=VALUES(version_comment), binlog_server=VALUES(binlog_server), read_only=VALUES(read_only), binlog_format=VALUES(binlog_format), binlog_row_image=VALUES(binlog_row_image), log_bin=VALUES(log_bin), log_replica_updates=VALUES(log_replica_updates), binary_log_file=VALUES(binary_log_file), binary_log_pos=VALUES(binary_log_pos), source_host=VALUES(source_host), source_port=VALUES(source_port), replica_sql_running=VALUES(replica_sql_running), replica_io_running=VALUES(replica_io_running), replication_sql_thread_state=VALUES(replication_sql_thread_state), replication_io_thread_state=VALUES(replication_io_thread_state), has_replication_filters=VALUES(has_replication_filters), supports_oracle_gtid=VALUES(supports_oracle_gtid), oracle_gtid=VALUES(oracle_gtid), source_uuid=VALUES(source_uuid), ancestry_uuid=VALUES(ancestry_uuid), executed_gtid_set=VALUES(executed_gtid_set), gtid_mode=VALUES(gtid_mode), gtid_purged=VALUES(gtid_purged), gtid_errant=VALUES(gtid_errant), mariadb_gtid=VALUES(mariadb_gtid), pseudo_gtid=VALUES(pseudo_gtid), source_log_file=VALUES(source_log_file), read_source_log_pos=VALUES(read_source_log_pos), relay_source_log_file=VALUES(relay_source_log_file), exec_source_log_pos=VALUES(exec_source_log_pos), relay_log_file=VALUES(relay_log_file), relay_log_pos=VALUES(relay_log_pos), last_sql_error=VALUES(last_sql_error), last_io_error=VALUES(last_io_error), replication_lag_seconds=VALUES(replication_lag_seconds), replica_lag_seconds=VALUES(replica_lag_seconds), sql_delay=VALUES(sql_delay), num_replica_hosts=VALUES(num_replica_hosts), replica_hosts=VALUES(replica_hosts), cluster_name=VALUES(cluster_name), suggested_cluster_alias=VALUES(suggested_cluster_alias), data_center=VALUES(data_center), region=VALUES(region),
+                hostname=VALUES(hostname), port=VALUES(port), last_checked=VALUES(last_checked), last_attempted_check=VALUES(last_attempted_check), last_check_partial_success=VALUES(last_check_partial_success), server_id=VALUES(server_id), server_uuid=VALUES(server_uuid), version=VALUES(version), major_version=VALUES(major_version), version_comment=VALUES(version_comment), binlog_server=VALUES(binlog_server), read_only=VALUES(read_only), binlog_format=VALUES(binlog_format), binlog_row_image=VALUES(binlog_row_image), log_bin=VALUES(log_bin), log_replica_updates=VALUES(log_replica_updates), binary_log_file=VALUES(binary_log_file), binary_log_pos=VALUES(binary_log_pos), source_host=VALUES(source_host), source_port=VALUES(source_port), replica_sql_running=VALUES(replica_sql_running), replica_io_running=VALUES(replica_io_running), replication_sql_thread_state=VALUES(replication_sql_thread_state), replication_io_thread_state=VALUES(replication_io_thread_state), has_replication_filters=VALUES(has_replication_filters), supports_oracle_gtid=VALUES(supports_oracle_gtid), oracle_gtid=VALUES(oracle_gtid), source_uuid=VALUES(source_uuid), ancestry_uuid=VALUES(ancestry_uuid), executed_gtid_set=VALUES(executed_gtid_set), gtid_mode=VALUES(gtid_mode), gtid_purged=VALUES(gtid_purged), gtid_errant=VALUES(gtid_errant), mariadb_gtid=VALUES(mariadb_gtid), pseudo_gtid=VALUES(pseudo_gtid), source_log_file=VALUES(source_log_file), read_source_log_pos=VALUES(read_source_log_pos), relay_source_log_file=VALUES(relay_source_log_file), exec_source_log_pos=VALUES(exec_source_log_pos), relay_log_file=VALUES(relay_log_file), relay_log_pos=VALUES(relay_log_pos), last_sql_error=VALUES(last_sql_error), last_io_error=VALUES(last_io_error), replication_lag_seconds=VALUES(replication_lag_seconds), replica_lag_seconds=VALUES(replica_lag_seconds), sql_delay=VALUES(sql_delay), num_replica_hosts=VALUES(num_replica_hosts), replica_hosts=VALUES(replica_hosts), cluster_name=VALUES(cluster_name), data_center=VALUES(data_center), region=VALUES(region),
 								physical_environment=VALUES(physical_environment), replication_depth=VALUES(replication_depth), is_co_primary=VALUES(is_co_primary), has_replication_credentials=VALUES(has_replication_credentials), allow_tls=VALUES(allow_tls), semi_sync_enforced=VALUES(semi_sync_enforced),
 								semi_sync_primary_enabled=VALUES(semi_sync_primary_enabled), semi_sync_primary_timeout=VALUES(semi_sync_primary_timeout), semi_sync_primary_wait_for_replica_count=VALUES(semi_sync_primary_wait_for_replica_count), semi_sync_replica_enabled=VALUES(semi_sync_replica_enabled), semi_sync_primary_status=VALUES(semi_sync_primary_status), semi_sync_primary_clients=VALUES(semi_sync_primary_clients), semi_sync_replica_status=VALUES(semi_sync_replica_status),
 								instance_alias=VALUES(instance_alias), last_discovery_latency=VALUES(last_discovery_latency), replication_group_name=VALUES(replication_group_name), replication_group_is_single_primary_mode=VALUES(replication_group_is_single_primary_mode), replication_group_member_state=VALUES(replication_group_member_state), replication_group_member_role=VALUES(replication_group_member_role), replication_group_members=VALUES(replication_group_members), replication_group_primary_host=VALUES(replication_group_primary_host), replication_group_primary_port=VALUES(replication_group_primary_port), last_seen=VALUES(last_seen)
         `
 	a3 := `
-		i710, 3306, 710, , 5.6.7, 5.6, MySQL, false, false, STATEMENT, FULL, false, false, , 0, , 0, false, false, 0, 0, false, false, false, , , , , , , false, false, , 0, mysql.000007, 10, , 0, , , {0 false}, {0 false}, 0, 0, [], , , , , , 0, false, false, false, false, false, 0, 0, false, false, 0, false, , 0, , false, , , [], , 0,
-		i720, 3306, 720, , 5.6.7, 5.6, MySQL, false, false, STATEMENT, FULL, false, false, , 0, , 0, false, false, 0, 0, false, false, false, , , , , , , false, false, , 0, mysql.000007, 20, , 0, , , {0 false}, {0 false}, 0, 0, [], , , , , , 0, false, false, false, false, false, 0, 0, false, false, 0, false, , 0, , false, , , [], , 0,
-		i730, 3306, 730, , 5.6.7, 5.6, MySQL, false, false, STATEMENT, FULL, false, false, , 0, , 0, false, false, 0, 0, false, false, false, , , , , , , false, false, , 0, mysql.000007, 30, , 0, , , {0 false}, {0 false}, 0, 0, [], , , , , , 0, false, false, false, false, false, 0, 0, false, false, 0, false, , 0, , false, , , [], , 0,
+		i710, 3306, 710, , 5.6.7, 5.6, MySQL, false, false, STATEMENT, FULL, false, false, , 0, , 0, false, false, 0, 0, false, false, false, , , , , , , false, false, , 0, mysql.000007, 10, , 0, , , {0 false}, {0 false}, 0, 0, [], , , , , 0, false, false, false, false, false, 0, 0, false, false, 0, false, , 0, , false, , , [], , 0,
+		i720, 3306, 720, , 5.6.7, 5.6, MySQL, false, false, STATEMENT, FULL, false, false, , 0, , 0, false, false, 0, 0, false, false, false, , , , , , , false, false, , 0, mysql.000007, 20, , 0, , , {0 false}, {0 false}, 0, 0, [], , , , , 0, false, false, false, false, false, 0, 0, false, false, 0, false, , 0, , false, , , [], , 0,
+		i730, 3306, 730, , 5.6.7, 5.6, MySQL, false, false, STATEMENT, FULL, false, false, , 0, , 0, false, false, 0, 0, false, false, false, , , , , , , false, false, , 0, mysql.000007, 30, , 0, , , {0 false}, {0 false}, 0, 0, [], , , , , 0, false, false, false, false, false, 0, 0, false, false, 0, false, , 0, , false, , , [], , 0,
 		`
 
 	sql3, args3, err := mkInsertOdkuForInstances(instances[:3], true, true)
diff --git a/go/vt/orchestrator/test/recovery_analysis.go b/go/vt/orchestrator/test/recovery_analysis.go
index 03871382a3..8548374789 100644
--- a/go/vt/orchestrator/test/recovery_analysis.go
+++ b/go/vt/orchestrator/test/recovery_analysis.go
@@ -46,9 +46,7 @@ type InfoForRecoveryAnalysis struct {
 	LogFile                                   string
 	LogPos                                    int64
 	IsStaleBinlogCoordinates                  int
-	SuggestedClusterAlias                     string
 	ClusterName                               string
-	ClusterAlias                              string
 	ClusterDomain                             string
 	GTIDMode                                  string
 	LastCheckValid                            int
@@ -92,7 +90,6 @@ func (info *InfoForRecoveryAnalysis) ConvertToRowMap() sqlutils.RowMap {
 	rowMap := make(sqlutils.RowMap)
 	rowMap["binary_log_file"] = sqlutils.CellData{String: info.LogFile, Valid: true}
 	rowMap["binary_log_pos"] = sqlutils.CellData{String: fmt.Sprintf("%v", info.LogPos), Valid: true}
-	rowMap["cluster_alias"] = sqlutils.CellData{String: info.ClusterAlias, Valid: true}
 	rowMap["cluster_domain"] = sqlutils.CellData{String: info.ClusterDomain, Valid: true}
 	rowMap["cluster_name"] = sqlutils.CellData{String: info.ClusterName, Valid: true}
 	rowMap["count_binlog_server_replicas"] = sqlutils.CellData{Valid: false}
@@ -154,7 +151,6 @@ func (info *InfoForRecoveryAnalysis) ConvertToRowMap() sqlutils.RowMap {
 	rowMap["semi_sync_replica_enabled"] = sqlutils.CellData{String: fmt.Sprintf("%v", info.SemiSyncReplicaEnabled), Valid: true}
 	rowMap["source_host"] = sqlutils.CellData{String: info.SourceHost, Valid: true}
 	rowMap["source_port"] = sqlutils.CellData{String: fmt.Sprintf("%v", info.SourcePort), Valid: true}
-	rowMap["suggested_cluster_alias"] = sqlutils.CellData{String: info.SuggestedClusterAlias, Valid: true}
 	res, _ := prototext.Marshal(info.TabletInfo)
 	rowMap["tablet_info"] = sqlutils.CellData{String: string(res), Valid: true}
 	return rowMap
@@ -165,7 +161,6 @@ func (info *InfoForRecoveryAnalysis) SetValuesFromTabletInfo() {
 	info.Port = int(info.TabletInfo.MysqlPort)
 	info.DataCenter = info.TabletInfo.Alias.Cell
 	info.Keyspace = info.TabletInfo.Keyspace
-	info.ClusterName = fmt.Sprintf("%v:%d", info.TabletInfo.MysqlHostname, info.TabletInfo.MysqlPort)
-	info.ClusterAlias = fmt.Sprintf("%v:%d", info.TabletInfo.MysqlHostname, info.TabletInfo.MysqlPort)
+	info.ClusterName = fmt.Sprintf("%v:%v", info.TabletInfo.Keyspace, info.TabletInfo.Shard)
 	info.ClusterDomain = fmt.Sprintf("%v:%d", info.TabletInfo.MysqlHostname, info.TabletInfo.MysqlPort)
 }
diff --git a/go/vt/vtgate/engine/memory_sort_test.go b/go/vt/vtgate/engine/memory_sort_test.go
index 80b75b6177..93b7687678 100644
--- a/go/vt/vtgate/engine/memory_sort_test.go
+++ b/go/vt/vtgate/engine/memory_sort_test.go
@@ -35,7 +35,7 @@ import (
 func init() {
 	// We require MySQL 8.0 collations for the comparisons in the tests
 	mySQLVersion := "8.0.0"
-	servenv.MySQLServerVersion = &mySQLVersion
+	servenv.SetMySQLServerVersionForTest(mySQLVersion)
 	collationEnv = collations.NewEnvironment(mySQLVersion)
 }
 
diff --git a/go/vt/vtgate/engine/ordered_aggregate_test.go b/go/vt/vtgate/engine/ordered_aggregate_test.go
index 461df17796..a24ba88fd5 100644
--- a/go/vt/vtgate/engine/ordered_aggregate_test.go
+++ b/go/vt/vtgate/engine/ordered_aggregate_test.go
@@ -39,7 +39,7 @@ var collationEnv *collations.Environment
 func init() {
 	// We require MySQL 8.0 collations for the comparisons in the tests
 	mySQLVersion := "8.0.0"
-	servenv.MySQLServerVersion = &mySQLVersion
+	servenv.SetMySQLServerVersionForTest(mySQLVersion)
 	collationEnv = collations.NewEnvironment(mySQLVersion)
 }
 
diff --git a/go/vt/vtgate/engine/route_test.go b/go/vt/vtgate/engine/route_test.go
index e7d58d66cd..1866465bf5 100644
--- a/go/vt/vtgate/engine/route_test.go
+++ b/go/vt/vtgate/engine/route_test.go
@@ -53,7 +53,7 @@ var defaultSelectResult = sqltypes.MakeTestResult(
 func init() {
 	// We require MySQL 8.0 collations for the comparisons in the tests
 	mySQLVersion := "8.0.0"
-	servenv.MySQLServerVersion = &mySQLVersion
+	servenv.SetMySQLServerVersionForTest(mySQLVersion)
 	collationEnv = collations.NewEnvironment(mySQLVersion)
 }
 
diff --git a/go/vt/vtgate/evalengine/comparisons_test.go b/go/vt/vtgate/evalengine/comparisons_test.go
index 020e9c4a13..d0899ba159 100644
--- a/go/vt/vtgate/evalengine/comparisons_test.go
+++ b/go/vt/vtgate/evalengine/comparisons_test.go
@@ -54,7 +54,7 @@ var (
 func init() {
 	// We require MySQL 8.0 collations for the comparisons in the tests
 	mySQLVersion := "8.0.0"
-	servenv.MySQLServerVersion = &mySQLVersion
+	servenv.SetMySQLServerVersionForTest(mySQLVersion)
 	collationEnv = collations.NewEnvironment(mySQLVersion)
 }
 
diff --git a/go/vt/vtgate/evalengine/integration/comparison_test.go b/go/vt/vtgate/evalengine/integration/comparison_test.go
index b2e5da3e46..cb3020e822 100644
--- a/go/vt/vtgate/evalengine/integration/comparison_test.go
+++ b/go/vt/vtgate/evalengine/integration/comparison_test.go
@@ -37,7 +37,7 @@ var collationEnv *collations.Environment
 func init() {
 	// We require MySQL 8.0 collations for the comparisons in the tests
 	mySQLVersion := "8.0.0"
-	servenv.MySQLServerVersion = &mySQLVersion
+	servenv.SetMySQLServerVersionForTest(mySQLVersion)
 	collationEnv = collations.NewEnvironment(mySQLVersion)
 }
 
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index 28eebbdf09..635d6a89e4 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -4114,6 +4114,51 @@ Gen4 plan same as above
 "unsupported: cross-shard correlated subquery"
 Gen4 error: exists sub-queries are only supported with AND clause
 
+# correlated subquery that is dependent on one side of a join, fully mergeable
+"SELECT music.id FROM music INNER JOIN user ON music.user_id = user.id WHERE music.user_id = 5 AND music.id = (SELECT MAX(m2.id) FROM music m2 WHERE m2.user_id = user.id)"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music INNER JOIN user ON music.user_id = user.id WHERE music.user_id = 5 AND music.id = (SELECT MAX(m2.id) FROM music m2 WHERE m2.user_id = user.id)",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select music.id from music join `user` on music.user_id = `user`.id where 1 != 1",
+    "Query": "select music.id from music join `user` on music.user_id = `user`.id where music.user_id = 5 and music.id = (select max(m2.id) from music as m2 where m2.user_id = `user`.id)",
+    "Table": "music, `user`",
+    "Values": [
+      "INT64(5)"
+    ],
+    "Vindex": "user_index"
+  }
+}
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music INNER JOIN user ON music.user_id = user.id WHERE music.user_id = 5 AND music.id = (SELECT MAX(m2.id) FROM music m2 WHERE m2.user_id = user.id)",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select music.id from music, `user` where 1 != 1",
+    "Query": "select music.id from music, `user` where music.user_id = 5 and music.id = (select max(m2.id) from music as m2 where m2.user_id = `user`.id) and music.user_id = `user`.id",
+    "Table": "`user`, music",
+    "Values": [
+      "INT64(5)"
+    ],
+    "Vindex": "user_index"
+  },
+  "TablesUsed": [
+    "user.music",
+    "user.user"
+  ]
+}
+
 # union as a derived table
 "select found from (select id as found from user union all (select id from unsharded)) as t"
 {
@@ -5542,3 +5587,72 @@ Gen4 plan same as above
     "user.user"
   ]
 }
+
+# Treating single value tuples as `EqualUnique` routes
+"SELECT music.id FROM music WHERE music.id IN (SELECT music.id FROM music WHERE music.user_id IN (5)) AND music.user_id = 5"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music WHERE music.id IN (SELECT music.id FROM music WHERE music.user_id IN (5)) AND music.user_id = 5",
+  "Instructions": {
+    "OperatorType": "Subquery",
+    "Variant": "PulloutIn",
+    "PulloutVars": [
+      "__sq_has_values1",
+      "__sq1"
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "Route",
+        "Variant": "IN",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select music.id from music where 1 != 1",
+        "Query": "select music.id from music where music.user_id in ::__vals",
+        "Table": "music",
+        "Values": [
+          "(INT64(5))"
+        ],
+        "Vindex": "user_index"
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "EqualUnique",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select music.id from music where 1 != 1",
+        "Query": "select music.id from music where music.user_id = 5 and :__sq_has_values1 = 1 and music.id in ::__sq1",
+        "Table": "music",
+        "Values": [
+          "INT64(5)"
+        ],
+        "Vindex": "user_index"
+      }
+    ]
+  }
+}
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music WHERE music.id IN (SELECT music.id FROM music WHERE music.user_id IN (5)) AND music.user_id = 5",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select music.id from music where 1 != 1",
+    "Query": "select music.id from music where music.id in (select music.id from music where music.user_id in (5)) and music.user_id = 5",
+    "Table": "music",
+    "Values": [
+      "INT64(5)"
+    ],
+    "Vindex": "user_index"
+  },
+  "TablesUsed": [
+    "user.music"
+  ]
+}
diff --git a/go/vt/vttablet/tabletmanager/rpc_replication_test.go b/go/vt/vttablet/tabletmanager/rpc_replication_test.go
index b37d88518b..0aac8c971e 100644
--- a/go/vt/vttablet/tabletmanager/rpc_replication_test.go
+++ b/go/vt/vttablet/tabletmanager/rpc_replication_test.go
@@ -74,3 +74,19 @@ func TestPromoteReplicaReplicationManagerFailure(t *testing.T) {
 	// At the end we expect the replication manager to be stopped.
 	require.True(t, tm.replManager.ticks.Running())
 }
+
+// TestDisableReplicationManager checks that the replication manager doesn't start if it is disabled
+func TestDisableReplicationManager(t *testing.T) {
+	ts := memorytopo.NewServer("cell1")
+	statsTabletTypeCount.ResetAll()
+	prevDisableReplicationManager := disableReplicationManager
+	disableReplicationManager = true
+	defer func() {
+		disableReplicationManager = prevDisableReplicationManager
+	}()
+
+	tm := newTestTM(t, ts, 100, keyspace, shard)
+	defer tm.Stop()
+
+	require.False(t, tm.replManager.ticks.Running())
+}
diff --git a/go/vt/vttest/vtprocess.go b/go/vt/vttest/vtprocess.go
index 3ad4d0e975..cafc0482a6 100644
--- a/go/vt/vttest/vtprocess.go
+++ b/go/vt/vttest/vtprocess.go
@@ -259,8 +259,8 @@ func VtcomboProcess(environment Environment, args *Config, mysql MySQLManager) (
 	if args.VSchemaDDLAuthorizedUsers != "" {
 		vt.ExtraArgs = append(vt.ExtraArgs, []string{"--vschema_ddl_authorized_users", args.VSchemaDDLAuthorizedUsers}...)
 	}
-	if *servenv.MySQLServerVersion != "" {
-		vt.ExtraArgs = append(vt.ExtraArgs, "--mysql_server_version", *servenv.MySQLServerVersion)
+	if mySQLVersion := servenv.MySQLServerVersion(); mySQLVersion != "" {
+		vt.ExtraArgs = append(vt.ExtraArgs, "--mysql_server_version", mySQLVersion)
 	}
 
 	if socket != "" {
diff --git a/test/config.json b/test/config.json
index 1bbede299b..51a7a6218c 100644
--- a/test/config.json
+++ b/test/config.json
@@ -463,6 +463,15 @@
 				"site_test"
 			]
 		},
+		"tabletmanager_replication_manager": {
+			"File": "unused.go",
+			"Args": ["vitess.io/vitess/go/test/endtoend/tabletmanager/replication_manager"],
+			"Command": [],
+			"Manual": false,
+			"Shard": "18",
+			"RetryMax": 1,
+			"Tags": []
+		},
 		"tabletmanager_consul": {
 			"File": "unused.go",
 			"Args": [